[Phase 4] Codex/OpenAI Engine Adapter (App-Server Protocol)

[frankbria]

Summary

Implement an AgentAdapter for OpenAI's Codex using the app-server JSON-RPC protocol defined in the Symphony spec. This adapter speaks the same protocol Symphony uses.

Parent issue: #408

Motivation

OpenAI's Codex has a well-defined app-server protocol (JSON-RPC over stdio) that Symphony already documents in detail (spec Section 10). Implementing this adapter makes CodeFrame compatible with the Codex ecosystem and any future agent that speaks this protocol.

Scope

New module: core/engines/codex.py

1. CodexAdapter implementing AgentAdapter:

class CodexAdapter:
    name = "codex"
    requires_api_key = {"OPENAI_API_KEY": "OpenAI API key"}
    
    def execute(self, task_prompt, workspace_path, context, timeout_ms):
        # 1. Launch codex app-server subprocess
        # 2. Send initialize → initialized → thread/start → turn/start
        # 3. Stream turn events
        # 4. Handle turn/completed, turn/failed, turn/cancelled
        # 5. Return AgentResult
        ...

2. App-server protocol implementation (per Symphony spec Section 10):

   • JSON-RPC message framing over stdio
   • Startup handshake: initialize → initialized → thread/start → turn/start
   • Turn streaming: read line-delimited JSON events
   • Completion detection: turn/completed, turn/failed, turn/cancelled
   • Approval handling: auto-approve by default (configurable)
   • Timeout: turn_timeout_ms, read_timeout_ms, stall_timeout_ms

3. Configuration:

   • codex_command: path to Codex binary (default: codex app-server)
   • approval_policy: auto-approve, require-approval
   • sandbox_mode: configurable per Symphony spec
   • turn_timeout_ms: default 3600000 (1 hour)

4. Event mapping:

   • Codex session_started → AgentEvent(type="progress")
   • Codex turn_completed → AgentEvent(type="completed")
   • Codex notification → AgentEvent(type="progress")
   • Token usage extraction from Codex events

Acceptance Criteria

• CodexAdapter implements AgentAdapter protocol
• Full app-server handshake (initialize → thread/start → turn/start)
• Turn streaming with proper line-delimited JSON parsing
• Auto-approval of command/file change requests (configurable)
• Timeout enforcement (turn, read, stall)
• Token usage tracked from Codex events
• Works with cf work start <id> --execute --engine codex
• Integration test with mock app-server

Dependencies

• Depends on [Phase 4] Agent Adapter Protocol Definition #409 (AgentAdapter protocol)
• Depends on [Phase 4] Task Context Packager for External Agents #410 (context packager for prompt assembly)
• Requires Codex CLI installed

[traycerai]

Plan

Observations

• codeframe/core/runtime.py's execute_agent() function dispatches engine selection via a string check ("react" → ReactAgent, "plan" → Agent). Adding "codex" is a well-defined extension point.
• No AgentAdapter base class or AgentResult/AgentEvent types exist yet — these are part of issue [Phase 4] Agent Adapter Protocol Definition #409 which this ticket depends on. The plan assumes those land before or alongside this work.
• ContextLoader in codeframe/core/context.py with TaskContext.to_prompt_context() is the existing prompt-assembly path for both current engines.
• codeframe/adapters/llm/ is the pattern namespace for adapters; engine-level adapters belong in codeframe/core/engines/ per the ticket's naming.
• openai>=1.12.0 is already in pyproject.toml — no new dependency needed for the API key check.

Approach

Create codeframe/core/engines/codex.py as a new module containing CodexAdapter. It speaks the Codex app-server JSON-RPC protocol over a subprocess stdio channel. The adapter bridges CodeFrame's existing execute_agent() dispatch path and the Codex CLI, mapping Codex turn events to AgentStatus. CLI and runtime changes are minimal: one new engine value in the allowed set and a new require_openai_api_key validator.

---

Implementation Plan

1. Create codeframe/core/engines/ package

Create file:codeframe/core/engines/__init__.py as an empty package marker so the directory becomes importable.

---

2. Define shared engine data structures

If issue #409 (AgentAdapter protocol) is not yet merged, define the following in codeframe/core/engines/__init__.py so CodexAdapter can import them locally. Otherwise import from wherever #409 places them.

Define two dataclasses:

• AgentEvent — fields: type: str (e.g. "progress", "completed"), message: str, data: dict
• AgentResult — fields: status: AgentStatus (reuse existing enum from codeframe/core/agent.py), events: list[AgentEvent], input_tokens: int, output_tokens: int, error: str | None

These mirror how ReactAgent.run() returns AgentStatus and how events are emitted via on_event.

---

3. Create codeframe/core/engines/codex.py — CodexAdapter

Class-level attributes (following the ticket spec):

name = "codex"
requires_api_key = {"OPENAI_API_KEY": "OpenAI API key"}

Constructor — accept:

• workspace: Workspace
• on_event: Callable[[str, dict], None] | None — mirrors the callback signature used by ReactAgent and Agent
• verbose: bool
• output_logger — same RunOutputLogger type passed to other engines
• codex_command: str — defaulting to "codex" (binary name)
• approval_policy: str — default "auto" (auto-approve all requests)
• sandbox_mode: str | None — passed through to Codex
• turn_timeout_ms: int — default 3600000 (1 hour)
• read_timeout_ms: int — default 30000 (30 s line-read timeout)
• stall_timeout_ms: int — default 300000 (5 min no-progress timeout)

execute(task_id: str) -> AgentStatus — main entry point:

1. Use ContextLoader(workspace).load(task_id) to build a TaskContext, then call context.to_prompt_context() to assemble the task prompt string — identical to how ReactAgent builds its context.
2. Call the private _run_app_server(task_prompt, workspace.repo_path) method.
3. Map the returned AgentResult to AgentStatus and return.

_run_app_server(task_prompt: str, workspace_path: Path) -> AgentResult:

Build the subprocess launch args from self.codex_command (split by shlex), append ["app-server"]. Use subprocess.Popen with stdin=PIPE, stdout=PIPE, stderr=PIPE, cwd set to workspace_path.

Then carry out the handshake and streaming via the private helper methods below.

---

3a. JSON-RPC message framing helpers

_send(proc, method: str, params: dict, msg_id: int | None = None) -> None
Serialize {"jsonrpc": "2.0", "method": method, "params": params, "id": msg_id} as a single JSON line and write to proc.stdin, flushing immediately.

_recv_line(proc, timeout_s: float) -> dict | None
Read one line from proc.stdout using select/readline with the given timeout. Return the parsed JSON dict, or None on timeout/EOF.

---

3b. Startup handshake — _handshake(proc) -> bool

Perform the four-step initialization sequence:

Step	Direction	Method / Event
1	→ Codex	initialize (send capabilities)
2	← Codex	wait for initialized notification
3	→ Codex	thread/start (send thread_id, workspace)
4	→ Codex	turn/start (send turn_id, prompt)

Return True if all four steps complete without timeout; False otherwise.

---

3c. Turn streaming — _stream_turn(proc) -> AgentResult

Loop reading line-delimited JSON events from proc.stdout until a terminal event is received or timeout fires.

Event routing table:

Codex event type	Action
session_started	Emit AgentEvent(type="progress", message="Session started") via on_event
notification	Extract message text; emit AgentEvent(type="progress")
tool_call / command request	Call _handle_approval(proc, event)
turn_completed	Extract token usage; return AgentResult(status=AgentStatus.COMPLETED, ...)
turn_failed	Return AgentResult(status=AgentStatus.FAILED, error=...)
turn_cancelled	Return AgentResult(status=AgentStatus.FAILED, error="turn cancelled")

Track last_event_time for stall timeout enforcement. If stall_timeout_ms elapses with no new event, terminate proc and return a FAILED result.

If turn_timeout_ms is exceeded (wall-clock), kill the subprocess and return FAILED.

---

3d. Approval handling — _handle_approval(proc, event: dict) -> None

When approval_policy == "auto":

• Send {"method": "tool_call/approved", "params": {"id": event["id"]}} immediately.
• Emit a progress event noting the auto-approved action.

When approval_policy == "require":

• Create a CodeFrame blocker via codeframe/core/blockers.py asking the human to approve, then wait (blocking loop) for the blocker answer, then send approved/rejected accordingly.

---

3e. Token usage extraction — _extract_token_usage(event: dict) -> tuple[int, int]

Parse input_tokens and output_tokens from the turn_completed event payload. Return (0, 0) if keys are absent (graceful degradation).

---

4. Update codeframe/core/runtime.py — extend execute_agent()

In execute_agent():

1. Add "codex" to valid_engines tuple (currently ("plan", "react")).
2. Remove the hard-coded ANTHROPIC_API_KEY check at the top of execute_agent and replace it with a per-engine key check:
   • "react" / "plan" → check ANTHROPIC_API_KEY (existing behavior)
   • "codex" → check OPENAI_API_KEY
3. In the engine dispatch block, add an elif engine == "codex": branch:
   • Instantiate CodexAdapter(workspace=workspace, verbose=verbose, on_event=on_agent_event, output_logger=output_logger, ...)
   • Call codex_status = adapter.execute(run.task_id)
   • Wrap into AgentState(status=codex_status) to match the return contract already used by the "react" branch.

---

5. Update codeframe/cli/validators.py — add require_openai_api_key()

Mirror the existing require_anthropic_api_key() pattern exactly, but check/load OPENAI_API_KEY. Print a matching error message referencing OPENAI_API_KEY.

---

6. Update codeframe/cli/app.py — work_start command

Two small changes in work_start:

1. Update the --engine option help string to include "codex" as a valid value:
   "Agent engine: 'react' (default), 'plan' (legacy step-based), or 'codex' (OpenAI Codex app-server)"

2. In the pre-execution API key check block (currently calls require_anthropic_api_key()), branch by engine:

   • If engine == "codex" → call the new require_openai_api_key() from codeframe/cli/validators.py
   • Otherwise → call require_anthropic_api_key() (existing behavior)

---

Data & Control Flow

sequenceDiagram
    participant CLI as cf work start --engine codex
    participant Runtime as runtime.execute_agent()
    participant Adapter as CodexAdapter.execute()
    participant Proc as codex app-server (subprocess)

    CLI->>Runtime: execute_agent(workspace, run, engine="codex")
    Runtime->>Adapter: CodexAdapter(workspace, ...).execute(task_id)
    Adapter->>Adapter: ContextLoader.load(task_id) → to_prompt_context()
    Adapter->>Proc: Popen("codex app-server", cwd=workspace_path)
    Adapter->>Proc: _send initialize
    Proc-->>Adapter: initialized
    Adapter->>Proc: _send thread/start
    Adapter->>Proc: _send turn/start (prompt)
    loop Stream events
        Proc-->>Adapter: session_started / notification / tool_call
        Adapter-->>CLI: on_event("progress", {...})
        alt tool_call (approval_policy=auto)
            Adapter->>Proc: tool_call/approved
        end
        Proc-->>Adapter: turn_completed / turn_failed / turn_cancelled
    end
    Adapter-->>Runtime: AgentResult → AgentStatus
    Runtime->>Runtime: complete_run / fail_run / block_run
    Runtime-->>CLI: AgentState

Loading

Import In IDE

🤖 Prompt for AI Agents

## Observations

- `codeframe/core/runtime.py`'s `execute_agent()` function dispatches engine selection via a string check (`"react"` → `ReactAgent`, `"plan"` → `Agent`). Adding `"codex"` is a well-defined extension point.
- No `AgentAdapter` base class or `AgentResult`/`AgentEvent` types exist yet — these are part of issue #409 which this ticket depends on. The plan assumes those land before or alongside this work.
- `ContextLoader` in `codeframe/core/context.py` with `TaskContext.to_prompt_context()` is the existing prompt-assembly path for both current engines.
- `codeframe/adapters/llm/` is the pattern namespace for adapters; engine-level adapters belong in `codeframe/core/engines/` per the ticket's naming.
- `openai>=1.12.0` is already in `pyproject.toml` — no new dependency needed for the API key check.

## Approach

Create `codeframe/core/engines/codex.py` as a new module containing `CodexAdapter`. It speaks the Codex app-server JSON-RPC protocol over a subprocess stdio channel. The adapter bridges CodeFrame's existing `execute_agent()` dispatch path and the Codex CLI, mapping Codex turn events to `AgentStatus`. CLI and runtime changes are minimal: one new `engine` value in the allowed set and a new `require_openai_api_key` validator.

---

## Implementation Plan

### 1. Create `codeframe/core/engines/` package

Create `file:codeframe/core/engines/__init__.py` as an empty package marker so the directory becomes importable.

---

### 2. Define shared engine data structures

If issue #409 (`AgentAdapter` protocol) is not yet merged, define the following **in** `codeframe/core/engines/__init__.py` so `CodexAdapter` can import them locally. Otherwise import from wherever #409 places them.

Define two dataclasses:

- **`AgentEvent`** — fields: `type: str` (e.g. `"progress"`, `"completed"`), `message: str`, `data: dict`
- **`AgentResult`** — fields: `status: AgentStatus` (reuse existing enum from `codeframe/core/agent.py`), `events: list[AgentEvent]`, `input_tokens: int`, `output_tokens: int`, `error: str | None`

These mirror how `ReactAgent.run()` returns `AgentStatus` and how events are emitted via `on_event`.

---

### 3. Create `codeframe/core/engines/codex.py` — `CodexAdapter`

**Class-level attributes** (following the ticket spec):

```
name = "codex"
requires_api_key = {"OPENAI_API_KEY": "OpenAI API key"}
```

**Constructor** — accept:

- `workspace: Workspace`
- `on_event: Callable[[str, dict], None] | None` — mirrors the callback signature used by `ReactAgent` and `Agent`
- `verbose: bool`
- `output_logger` — same `RunOutputLogger` type passed to other engines
- `codex_command: str` — defaulting to `"codex"` (binary name)
- `approval_policy: str` — default `"auto"` (auto-approve all requests)
- `sandbox_mode: str | None` — passed through to Codex
- `turn_timeout_ms: int` — default `3600000` (1 hour)
- `read_timeout_ms: int` — default `30000` (30 s line-read timeout)
- `stall_timeout_ms: int` — default `300000` (5 min no-progress timeout)

**`execute(task_id: str) -> AgentStatus`** — main entry point:

1. Use `ContextLoader(workspace).load(task_id)` to build a `TaskContext`, then call `context.to_prompt_context()` to assemble the task prompt string — identical to how `ReactAgent` builds its context.
2. Call the private `_run_app_server(task_prompt, workspace.repo_path)` method.
3. Map the returned `AgentResult` to `AgentStatus` and return.

**`_run_app_server(task_prompt: str, workspace_path: Path) -> AgentResult`**:

Build the subprocess launch args from `self.codex_command` (split by shlex), append `["app-server"]`. Use `subprocess.Popen` with `stdin=PIPE`, `stdout=PIPE`, `stderr=PIPE`, cwd set to `workspace_path`.

Then carry out the handshake and streaming via the private helper methods below.

---

#### 3a. JSON-RPC message framing helpers

**`_send(proc, method: str, params: dict, msg_id: int | None = None) -> None`**  
Serialize `{"jsonrpc": "2.0", "method": method, "params": params, "id": msg_id}` as a single JSON line and write to `proc.stdin`, flushing immediately.

**`_recv_line(proc, timeout_s: float) -> dict | None`**  
Read one line from `proc.stdout` using `select`/`readline` with the given timeout. Return the parsed JSON dict, or `None` on timeout/EOF.

---

#### 3b. Startup handshake — `_handshake(proc) -> bool`

Perform the four-step initialization sequence:

| Step | Direction | Method / Event |
|------|-----------|----------------|
| 1 | → Codex | `initialize` (send capabilities) |
| 2 | ← Codex | wait for `initialized` notification |
| 3 | → Codex | `thread/start` (send `thread_id`, `workspace`) |
| 4 | → Codex | `turn/start` (send `turn_id`, `prompt`) |

Return `True` if all four steps complete without timeout; `False` otherwise.

---

#### 3c. Turn streaming — `_stream_turn(proc) -> AgentResult`

Loop reading line-delimited JSON events from `proc.stdout` until a terminal event is received or timeout fires.

**Event routing table:**

| Codex event type | Action |
|---|---|
| `session_started` | Emit `AgentEvent(type="progress", message="Session started")` via `on_event` |
| `notification` | Extract message text; emit `AgentEvent(type="progress")` |
| `tool_call` / command request | Call `_handle_approval(proc, event)` |
| `turn_completed` | Extract token usage; return `AgentResult(status=AgentStatus.COMPLETED, ...)` |
| `turn_failed` | Return `AgentResult(status=AgentStatus.FAILED, error=...)` |
| `turn_cancelled` | Return `AgentResult(status=AgentStatus.FAILED, error="turn cancelled")` |

Track `last_event_time` for stall timeout enforcement. If `stall_timeout_ms` elapses with no new event, terminate `proc` and return a `FAILED` result.

If `turn_timeout_ms` is exceeded (wall-clock), kill the subprocess and return `FAILED`.

---

#### 3d. Approval handling — `_handle_approval(proc, event: dict) -> None`

When `approval_policy == "auto"`:
- Send `{"method": "tool_call/approved", "params": {"id": event["id"]}}` immediately.
- Emit a `progress` event noting the auto-approved action.

When `approval_policy == "require"`:
- Create a CodeFrame blocker via `codeframe/core/blockers.py` asking the human to approve, then wait (blocking loop) for the blocker answer, then send approved/rejected accordingly.

---

#### 3e. Token usage extraction — `_extract_token_usage(event: dict) -> tuple[int, int]`

Parse `input_tokens` and `output_tokens` from the `turn_completed` event payload. Return `(0, 0)` if keys are absent (graceful degradation).

---

### 4. Update `codeframe/core/runtime.py` — extend `execute_agent()`

In `execute_agent()`:

1. Add `"codex"` to `valid_engines` tuple (currently `("plan", "react")`).
2. Remove the hard-coded `ANTHROPIC_API_KEY` check at the top of `execute_agent` and replace it with a per-engine key check:
   - `"react"` / `"plan"` → check `ANTHROPIC_API_KEY` (existing behavior)
   - `"codex"` → check `OPENAI_API_KEY`
3. In the engine dispatch block, add an `elif engine == "codex":` branch:
   - Instantiate `CodexAdapter(workspace=workspace, verbose=verbose, on_event=on_agent_event, output_logger=output_logger, ...)`
   - Call `codex_status = adapter.execute(run.task_id)`
   - Wrap into `AgentState(status=codex_status)` to match the return contract already used by the `"react"` branch.

---

### 5. Update `codeframe/cli/validators.py` — add `require_openai_api_key()`

Mirror the existing `require_anthropic_api_key()` pattern exactly, but check/load `OPENAI_API_KEY`. Print a matching error message referencing `OPENAI_API_KEY`.

---

### 6. Update `codeframe/cli/app.py` — `work_start` command

Two small changes in `work_start`:

1. Update the `--engine` option help string to include `"codex"` as a valid value:  
   `"Agent engine: 'react' (default), 'plan' (legacy step-based), or 'codex' (OpenAI Codex app-server)"`

2. In the pre-execution API key check block (currently calls `require_anthropic_api_key()`), branch by `engine`:
   - If `engine == "codex"` → call the new `require_openai_api_key()` from `codeframe/cli/validators.py`
   - Otherwise → call `require_anthropic_api_key()` (existing behavior)

---

### Data & Control Flow

```mermaid
sequenceDiagram
    participant CLI as cf work start --engine codex
    participant Runtime as runtime.execute_agent()
    participant Adapter as CodexAdapter.execute()
    participant Proc as codex app-server (subprocess)

    CLI->>Runtime: execute_agent(workspace, run, engine="codex")
    Runtime->>Adapter: CodexAdapter(workspace, ...).execute(task_id)
    Adapter->>Adapter: ContextLoader.load(task_id) → to_prompt_context()
    Adapter->>Proc: Popen("codex app-server", cwd=workspace_path)
    Adapter->>Proc: _send initialize
    Proc-->>Adapter: initialized
    Adapter->>Proc: _send thread/start
    Adapter->>Proc: _send turn/start (prompt)
    loop Stream events
        Proc-->>Adapter: session_started / notification / tool_call
        Adapter-->>CLI: on_event("progress", {...})
        alt tool_call (approval_policy=auto)
            Adapter->>Proc: tool_call/approved
        end
        Proc-->>Adapter: turn_completed / turn_failed / turn_cancelled
    end
    Adapter-->>Runtime: AgentResult → AgentStatus
    Runtime->>Runtime: complete_run / fail_run / block_run
    Runtime-->>CLI: AgentState
```

---

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.

[coderabbitai]

🔗 Related PRs

#9 - Merge Sprint 4 Multi-Agent Coordination work into main [merged]
#31 - Create SDK migration implementation plan [merged]
#125 - Implement WorkerAgent execute_task method [merged]
#363 - feat: wire ReactAgent into runtime with --engine react CLI flag [merged]
#370 - feat: token budget tracking and message compaction for ReactAgent (#352) [merged]

---

📝 Issue Planner

_{Check the box below or use the @coderabbitai plan command to generate an implementation plan and prompts that you can use with your favorite coding assistant.}

• Create Plan

---

🧪 Issue enrichment is currently in open beta.

You can configure auto-planning by selecting labels in the issue_enrichment configuration.

To disable automatic issue enrichment, add the following to your .coderabbit.yaml:

issue_enrichment:
  auto_enrich:
    enabled: false

💬 Have feedback or questions? Drop into our discord!