[Phase 4] Workspace Hooks: Execution Engine with Timeout and Error Handling

[frankbria]

Summary

Implement the hook execution engine that runs shell scripts with timeout enforcement, error handling, and structured logging.

Parent issue: #392

Scope

New module: core/hooks.py

1. async def run_hook(script: str, cwd: Path, timeout_ms: int, context: dict) -> HookResult

   • Execute shell script via subprocess with cwd set to workspace path
   • Enforce timeout (kill process on expiry)
   • Capture stdout/stderr (truncated to reasonable limit, ~10KB)
   • Return structured HookResult(success, exit_code, stdout, stderr, duration_ms)

2. Error handling:

   • Timeout → HookResult(success=False, error="timeout")
   • Non-zero exit → HookResult(success=False, exit_code=N)
   • Exception → HookResult(success=False, error=str(e))

3. Logging:

   • Log hook start with name and script preview
   • Log hook completion with duration
   • Log hook failures with stderr excerpt

Acceptance Criteria

• Hook scripts execute with correct cwd
• Timeout kills the subprocess after timeout_ms
• stdout/stderr captured and truncated
• Structured HookResult returned for all outcomes
• Integration test: successful hook, failing hook, timeout hook

Dependencies

• Depends on [Phase 4] Workspace Hooks: Configuration Schema and Parser #393 (HooksConfig schema)

[traycerai]

Plan

Observations

The codebase uses @dataclass (not Pydantic) for result types in core/ modules (e.g., StepResult, GateCheck, InstallResult). Async subprocess is already established via asyncio.create_subprocess_exec in codeframe/testing/lint_runner.py. Timeout enforcement uses asyncio.wait_for. Logging follows the logging.getLogger(__name__) module-level pattern. There is no existing HooksConfig from issue #393 yet, so HookResult and the context: dict parameter must be self-contained. The codebase targets Python 3.11+.

Approach

Create codeframe/core/hooks.py as a standalone async module following the exact patterns of lint_runner.py (for asyncio.create_subprocess_exec) and the @dataclass result conventions from executor.py and installer.py. The run_hook function wraps subprocess execution with asyncio.wait_for for timeout, captures and truncates stdout/stderr, and returns a structured HookResult for all outcomes.

---

Implementation Steps

1. Define HookResult dataclass

In the new file:codeframe/core/hooks.py, define HookResult as a @dataclass following the same convention as StepResult in file:codeframe/core/executor.py and InstallResult in file:codeframe/core/installer.py:

Field	Type	Notes
success	bool	True only on zero exit and no exception
exit_code	Optional[int]	None when process never started or timed out
stdout	str	Captured and truncated
stderr	str	Captured and truncated
duration_ms	int	Wall-clock time in milliseconds
error	Optional[str]	Human-readable reason: "timeout", str(e), or None

2. Define the output truncation constant

Define HOOK_OUTPUT_MAX_BYTES = 10_240 (10 KB) at module level. This mirrors the SSE_OUTPUT_MAX_CHARS pattern already present in file:codeframe/core/streaming.py. Apply this limit to both stdout and stderr independently using simple slicing on the decoded string.

3. Implement async def run_hook

Signature:

async def run_hook(script: str, cwd: Path, timeout_ms: int, context: dict) -> HookResult

Flow:

sequenceDiagram
    participant Caller
    participant run_hook
    participant asyncio.create_subprocess_exec
    participant Shell Process

    Caller->>run_hook: script, cwd, timeout_ms, context
    run_hook->>run_hook: log hook start (name from context, script preview[:80])
    run_hook->>asyncio.create_subprocess_exec: shell=True, cwd=cwd, PIPE stdout/stderr
    asyncio.create_subprocess_exec-->>run_hook: proc
    run_hook->>asyncio.wait_for: proc.communicate(), timeout=timeout_ms/1000
    alt success (exit_code == 0)
        asyncio.wait_for-->>run_hook: stdout, stderr
        run_hook-->>Caller: HookResult(success=True, exit_code=0, ...)
    else non-zero exit
        asyncio.wait_for-->>run_hook: stdout, stderr
        run_hook-->>Caller: HookResult(success=False, exit_code=N, ...)
    else asyncio.TimeoutError
        run_hook->>Shell Process: proc.kill()
        run_hook-->>Caller: HookResult(success=False, error="timeout")
    else Exception
        run_hook-->>Caller: HookResult(success=False, error=str(e))
    end
    run_hook->>run_hook: log completion with duration_ms
    run_hook->>run_hook: if not success → log stderr excerpt (first 500 chars)

Loading

Key implementation details:

• Use asyncio.create_subprocess_exec("sh", "-c", script, ...) with stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=cwd — matching the pattern from file:codeframe/testing/lint_runner.py.
• Wrap await proc.communicate() inside asyncio.wait_for(proc.communicate(), timeout=timeout_ms / 1000). This matches the asyncio.wait_for pattern already used throughout the codebase.
• On asyncio.TimeoutError: call proc.kill() then await proc.wait() to reap the zombie, then return HookResult(success=False, error="timeout", duration_ms=elapsed).
• Decode stdout/stderr with errors="replace" and truncate to HOOK_OUTPUT_MAX_BYTES characters.
• Compute duration_ms using datetime.now(timezone.utc) before and after (same as file:codeframe/core/executor.py).
• The context: dict is expected to carry a "name" key used in log messages (e.g. context.get("name", "unknown")). It is passed through but not otherwise required by the engine.

4. Logging instrumentation

Follow the logger = logging.getLogger(__name__) module-level pattern from every other core/ module:

• Hook start (logger.info): hook name, first 80 chars of script.
• Hook completion (logger.info): hook name, duration_ms.
• Hook failure (logger.warning): hook name, stderr excerpt (first 500 chars), exit_code or error.
• Timeout (logger.warning): hook name, timeout_ms.
• Unexpected exception (logger.exception): full traceback via logger.exception.

5. Integration test file

Create file:tests/integration/test_hooks_integration.py following the pattern of file:tests/integration/test_quality_gates_integration.py. Use pytest.mark.asyncio (already available via pytest-asyncio in pyproject.toml) and tmp_path for workspace directories.

Three integration test cases as required by the acceptance criteria:

Test	Script	Expected HookResult
test_run_hook_success	echo "hello"	success=True, exit_code=0, stdout contains "hello"
test_run_hook_failure	exit 1	success=False, exit_code=1
test_run_hook_timeout	sleep 10 with timeout_ms=200	success=False, error="timeout"

Each test constructs a real tmp_path directory as cwd, invokes run_hook directly, and asserts on the returned HookResult fields — no subprocess mocking needed for integration tests.

Import In IDE

🤖 Prompt for AI Agents

## Observations

The codebase uses `@dataclass` (not Pydantic) for result types in `core/` modules (e.g., `StepResult`, `GateCheck`, `InstallResult`). Async subprocess is already established via `asyncio.create_subprocess_exec` in `codeframe/testing/lint_runner.py`. Timeout enforcement uses `asyncio.wait_for`. Logging follows the `logging.getLogger(__name__)` module-level pattern. There is no existing `HooksConfig` from issue #393 yet, so `HookResult` and the `context: dict` parameter must be self-contained. The codebase targets Python 3.11+.

## Approach

Create `codeframe/core/hooks.py` as a standalone async module following the exact patterns of `lint_runner.py` (for `asyncio.create_subprocess_exec`) and the `@dataclass` result conventions from `executor.py` and `installer.py`. The `run_hook` function wraps subprocess execution with `asyncio.wait_for` for timeout, captures and truncates stdout/stderr, and returns a structured `HookResult` for all outcomes.

---

## Implementation Steps

### 1. Define `HookResult` dataclass

In the new `file:codeframe/core/hooks.py`, define `HookResult` as a `@dataclass` following the same convention as `StepResult` in `file:codeframe/core/executor.py` and `InstallResult` in `file:codeframe/core/installer.py`:

| Field | Type | Notes |
|---|---|---|
| `success` | `bool` | `True` only on zero exit and no exception |
| `exit_code` | `Optional[int]` | `None` when process never started or timed out |
| `stdout` | `str` | Captured and truncated |
| `stderr` | `str` | Captured and truncated |
| `duration_ms` | `int` | Wall-clock time in milliseconds |
| `error` | `Optional[str]` | Human-readable reason: `"timeout"`, `str(e)`, or `None` |

### 2. Define the output truncation constant

Define `HOOK_OUTPUT_MAX_BYTES = 10_240` (10 KB) at module level. This mirrors the `SSE_OUTPUT_MAX_CHARS` pattern already present in `file:codeframe/core/streaming.py`. Apply this limit to both stdout and stderr independently using simple slicing on the decoded string.

### 3. Implement `async def run_hook`

Signature:
```
async def run_hook(script: str, cwd: Path, timeout_ms: int, context: dict) -> HookResult
```

Flow:

```mermaid
sequenceDiagram
    participant Caller
    participant run_hook
    participant asyncio.create_subprocess_exec
    participant Shell Process

    Caller->>run_hook: script, cwd, timeout_ms, context
    run_hook->>run_hook: log hook start (name from context, script preview[:80])
    run_hook->>asyncio.create_subprocess_exec: shell=True, cwd=cwd, PIPE stdout/stderr
    asyncio.create_subprocess_exec-->>run_hook: proc
    run_hook->>asyncio.wait_for: proc.communicate(), timeout=timeout_ms/1000
    alt success (exit_code == 0)
        asyncio.wait_for-->>run_hook: stdout, stderr
        run_hook-->>Caller: HookResult(success=True, exit_code=0, ...)
    else non-zero exit
        asyncio.wait_for-->>run_hook: stdout, stderr
        run_hook-->>Caller: HookResult(success=False, exit_code=N, ...)
    else asyncio.TimeoutError
        run_hook->>Shell Process: proc.kill()
        run_hook-->>Caller: HookResult(success=False, error="timeout")
    else Exception
        run_hook-->>Caller: HookResult(success=False, error=str(e))
    end
    run_hook->>run_hook: log completion with duration_ms
    run_hook->>run_hook: if not success → log stderr excerpt (first 500 chars)
```

**Key implementation details:**

- Use `asyncio.create_subprocess_exec("sh", "-c", script, ...)` with `stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=cwd` — matching the pattern from `file:codeframe/testing/lint_runner.py`.
- Wrap `await proc.communicate()` inside `asyncio.wait_for(proc.communicate(), timeout=timeout_ms / 1000)`. This matches the `asyncio.wait_for` pattern already used throughout the codebase.
- On `asyncio.TimeoutError`: call `proc.kill()` then `await proc.wait()` to reap the zombie, then return `HookResult(success=False, error="timeout", duration_ms=elapsed)`.
- Decode stdout/stderr with `errors="replace"` and truncate to `HOOK_OUTPUT_MAX_BYTES` characters.
- Compute `duration_ms` using `datetime.now(timezone.utc)` before and after (same as `file:codeframe/core/executor.py`).
- The `context: dict` is expected to carry a `"name"` key used in log messages (e.g. `context.get("name", "unknown")`). It is passed through but not otherwise required by the engine.

### 4. Logging instrumentation

Follow the `logger = logging.getLogger(__name__)` module-level pattern from every other `core/` module:

- **Hook start** (`logger.info`): hook name, first 80 chars of script.
- **Hook completion** (`logger.info`): hook name, `duration_ms`.
- **Hook failure** (`logger.warning`): hook name, stderr excerpt (first 500 chars), `exit_code` or `error`.
- **Timeout** (`logger.warning`): hook name, `timeout_ms`.
- **Unexpected exception** (`logger.exception`): full traceback via `logger.exception`.

### 5. Integration test file

Create `file:tests/integration/test_hooks_integration.py` following the pattern of `file:tests/integration/test_quality_gates_integration.py`. Use `pytest.mark.asyncio` (already available via `pytest-asyncio` in `pyproject.toml`) and `tmp_path` for workspace directories.

Three integration test cases as required by the acceptance criteria:

| Test | Script | Expected `HookResult` |
|---|---|---|
| `test_run_hook_success` | `echo "hello"` | `success=True, exit_code=0, stdout contains "hello"` |
| `test_run_hook_failure` | `exit 1` | `success=False, exit_code=1` |
| `test_run_hook_timeout` | `sleep 10` with `timeout_ms=200` | `success=False, error="timeout"` |

Each test constructs a real `tmp_path` directory as `cwd`, invokes `run_hook` directly, and asserts on the returned `HookResult` fields — no subprocess mocking needed for integration tests.

---

Execution Information

Branch: main
Commit: 705b160

---

💡 Tips

Supported Commands (Inside Comments)

• Use @traycerai generate to iterate on the previous version of the implementation plan.

Supported Commands (Inside Description)

• Add @traycerai ignore anywhere in the ticket description to prevent this ticket from being processed.
• Add @traycerai branch:<branch-name> anywhere in the ticket description to specify the target branch for the implementation plan.

Community

• Join our Discord Community to get help, request features, and share feedback.
• Follow us on X/Twitter for updates and announcements.

[frankbria]

Implemented in PR #438 (merged). All hook system components (config schema, execution engine, template rendering, runtime integration, CLI commands) were delivered as a single PR implementing the parent issue #392.