feat(codex): event-stream parser tap for the unified harness surface

[declan-scale]

Summary

• Ports the golden agent's _CodexEventProcessor to a pure SDK tap: convert_codex_to_agentex_events + CodexTurn + usage normalization (codex_usage_to_turn_usage)
• _CodexStreamProcessor in _codex_sync.py maps all codex exec JSON event types (agent_message, reasoning, command_execution, file_change, mcp_tool_call, web_search, todo_list, collab_tool_call, error events) to canonical StreamTaskMessage* events
• CodexTurn in _codex_turn.py wraps the parser as a HarnessTurn so callers can pass it directly to UnifiedEmitter.yield_turn / UnifiedEmitter.auto_send_turn
• Cross-channel conformance fixtures added in test_codex_conformance.py (4 fixtures: text, tool, reasoning, multi-step)
• Orchestration (subprocess management, sandbox provisioning, secret injection, MCP config) remains in the golden agent — this module is a pure library tap with no deployable agent

🤖 Generated with Claude Code

Greptile Summary

This PR ports the golden agent's _CodexEventProcessor into a reusable SDK tap: _CodexStreamProcessor / convert_codex_to_agentex_events maps every codex exec --json event type to canonical StreamTaskMessage* events, CodexTurn wraps it as a HarnessTurn, and codex_usage_to_turn_usage normalises OpenAI-style token counts. Three tutorial variants (sync, async-base, Temporal) demonstrate the tap against a live subprocess, and a cross-channel conformance suite verifies deterministic span derivation alongside other harness taps.

• _codex_sync.py: stateful _CodexStreamProcessor handles agent_message (delta streaming), reasoning, all tool types (command_execution, file_change, mcp_tool_call, web_search, todo_list, collab_tool_call), and error events; previous review issues with tool_call_id consistency and collab_tool_call args/errors are now resolved.
• _codex_turn.py: CodexTurn caches the events generator so _on_result fires exactly once; codex_usage_to_turn_usage defensively distinguishes None ("not reported") from explicit 0.
• Test assertions for num_llm_calls in test_openai_turn.py and test_pydantic_ai_turn.py are updated from == 0 to is None for the pre-exhaustion case, aligning with TurnUsage's None-default semantics.

Confidence Score: 5/5

Safe to merge; the core parser and turn wrapper are well-tested and the three previous review issues are all addressed.

The parser correctly handles all documented codex event types, the generator-caching fix prevents stale usage data, and the tool_call_id consistency fix prevents mismatched request/response pairs. The only gaps are the web_search/todo_list error-path question (needs clarification, not a known breakage) and the missing try/finally in the Temporal activity (tutorial code that degrades gracefully in practice).

src/agentex/lib/adk/_modules/_codex_sync.py for the web_search/todo_list error payload question; examples/tutorials/10_async/10_temporal/harness_codex/project/activities.py for the missing try/finally around auto_send_turn.

Important Files Changed

Filename	Overview
src/agentex/lib/adk/_modules/_codex_sync.py	Core event-stream parser tap: maps all codex exec JSON event types to canonical StreamTaskMessage* events; previous review issues (tool_call_id drift, collab_tool_call args/errors) are now fixed; web_search/todo_list error handling falls through to generic is_error=False path.
src/agentex/lib/adk/_modules/_codex_turn.py	CodexTurn HarnessTurn implementation: generator caching, on_result wiring, and codex_usage_to_turn_usage normalization all look correct; previous reviewer's events-property caching issue is resolved.
examples/tutorials/10_async/10_temporal/harness_codex/project/activities.py	Temporal activity for codex turn: process spawning and CodexTurn wiring look correct but process.wait() is not guarded in try/finally, which could orphan subprocesses on activity failure/retry.
tests/lib/adk/test_codex_sync.py	Comprehensive offline tests covering text, tool, reasoning, error, edge-case, on_result, and multi-step scenarios including the new empty-item-id request/response ID match test.
tests/lib/adk/test_codex_turn.py	Thorough tests for CodexTurn protocol compliance, usage normalization, generator stability, and on_result wiring; events_property_stable_across_accesses explicitly validates the caching fix.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant Caller as Caller (ACP handler / Activity)
    participant CT as CodexTurn
    participant CS as convert_codex_to_agentex_events
    participant CSP as _CodexStreamProcessor
    participant UE as UnifiedEmitter

    Caller->>CT: "CodexTurn(events=stdout_lines, model=)"
    Caller->>UE: yield_turn(turn) / auto_send_turn(turn)
    UE->>CT: async for msg in turn.events
    CT->>CS: create generator (once, cached)
    loop per raw event line
        CS->>CS: JSON parse / pass-through dict
        CS->>CSP: process(evt)
        CSP-->>CS: "list[StreamTaskMessage*]"
        CS-->>UE: "yield StreamTaskMessage*"
        UE-->>Caller: forward / push to Redis
    end
    Note over CS,CSP: turn.completed - _pending_usage captured
    CS->>CT: "on_result({usage, session_id, counts})"
    CT->>CT: "self._result = result"
    Caller->>CT: turn.usage()
    CT->>CT: codex_usage_to_turn_usage(_result)
    CT-->>Caller: TurnUsage

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant Caller as Caller (ACP handler / Activity)
    participant CT as CodexTurn
    participant CS as convert_codex_to_agentex_events
    participant CSP as _CodexStreamProcessor
    participant UE as UnifiedEmitter

    Caller->>CT: "CodexTurn(events=stdout_lines, model=)"
    Caller->>UE: yield_turn(turn) / auto_send_turn(turn)
    UE->>CT: async for msg in turn.events
    CT->>CS: create generator (once, cached)
    loop per raw event line
        CS->>CS: JSON parse / pass-through dict
        CS->>CSP: process(evt)
        CSP-->>CS: "list[StreamTaskMessage*]"
        CS-->>UE: "yield StreamTaskMessage*"
        UE-->>Caller: forward / push to Redis
    end
    Note over CS,CSP: turn.completed - _pending_usage captured
    CS->>CT: "on_result({usage, session_id, counts})"
    CT->>CT: "self._result = result"
    Caller->>CT: turn.usage()
    CT->>CT: codex_usage_to_turn_usage(_result)
    CT-->>Caller: TurnUsage

Loading

_{Reviews (12): Last reviewed commit: "test(harness): num_llm_calls is None (no..." | Re-trigger Greptile}

[danielmillerp]

is this a new integration?