What

PR 6 of the unified-harness-surface series: migrate the OpenAI Agents SDK integration onto the shared harness surface.

Library

• OpenAITurn (src/agentex/lib/adk/providers/_modules/openai_turn.py): a HarnessTurn adapter that wraps a Runner.run_streamed result. It converts the SDK's native events into the canonical StreamTaskMessage* stream via the existing convert_openai_to_agentex_events, and after the stream is exhausted reads result.raw_responses to aggregate per-response usage into a provider-independent TurnUsage.

  • openai_usage_to_turn_usage(usage, model) maps agents.Usage -> TurnUsage with defensive getattr access so present-but-zero values (e.g. 0 output tokens on a cache hit) survive as 0, not None.
  • _aggregate_usage(raw_responses) sums usage across ModelResponses via Usage.add, skipping responses without usage.
  • Accepts either result= (a streamed run) or stream= (a pre-built canonical stream, for tests); raises ValueError if neither. coalesce_tool_requests is a no-op kept for API parity.

• OpenAIService.run_agent_streamed_auto_send: replaced the ~270-line inline streaming/reasoning/span loop with UnifiedEmitter.auto_send_turn(OpenAITurn(result=result, model=model)). Guardrail tripwire handling and the RunResultStreaming return type are preserved. The created_at first-message ordering limitation under the unified path is documented in a comment. OpenAITurn is imported lazily inside the method to avoid a circular import at package init.

• SyncStreamingModel / SyncStreamingProvider: docstring-deprecated (no runtime warning), pointing at the harness pattern.

Tests

• tests/lib/adk/providers/test_openai_turn.py: usage mapping (full / None / real zeros), _aggregate_usage (empty / single / multiple), events driven by an injected canonical stream, usage() before/after exhaustion (including the result-backed path), and the ValueError guard.
• tests/lib/core/harness/conformance/test_openai_conformance.py: text-only, tool-call, reasoning, and multi-step canonical fixtures; registers module-locally and parametrizes over its own list to avoid the cross-module global-registry hazard.
• tests/lib/adk/providers/test_openai_activities.py: updated the streamed-auto-send activity test to the new contract (full tool messages are posted by opening a context with initial_content and closing it, no stream_update).

Tutorials

Three tutorials demonstrating the same OpenAITurn across delivery modes, each with an offline test (no server / Redis / Temporal / API key required):

• examples/tutorials/00_sync/060_harness_openai — UnifiedEmitter.yield_turn
• examples/tutorials/10_async/00_base/130_harness_openai — UnifiedEmitter.auto_send_turn
• examples/tutorials/10_async/10_temporal/140_harness_openai — auto_send_turn inside a custom Temporal activity

Verification

• ./scripts/lint — clean (ruff + pyright, 0 errors)
• Full tests/ suite — 1016 passed, 1376 skipped
• All three tutorial offline tests pass individually

🤖 Generated with Claude Code

Greptile Summary

This PR migrates the OpenAI Agents SDK integration onto the shared harness surface by introducing OpenAITurn (a HarnessTurn adapter) and replacing the ~270-line inline streaming loop in run_agent_streamed_auto_send with UnifiedEmitter.auto_send_turn. It also adds three tutorial projects, a new unit-test module for OpenAITurn, and cross-channel conformance fixtures.

• OpenAITurn wraps RunResultStreaming, converts native SDK events to canonical StreamTaskMessage* via convert_openai_to_agentex_events, and aggregates raw_responses usage after stream exhaustion; UnifiedEmitter.auto_send_turn correctly reads turn.usage() only after consuming the stream, fixing the previously reported stale-usage bug.
• run_agent_streamed_auto_send now uses the 4-branch Runner.run_streamed pattern to restore previous_response_id forwarding, and auto_send_turn receives created_at directly (stamping all streaming contexts rather than only the first).
• Tutorials (060_harness_openai, 130_harness_openai, 140_harness_openai) demonstrate sync, async, and Temporal delivery modes; the Temporal tutorial now correctly accumulates multi-turn conversation history via input_list + result.to_input_list().

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant W as Workflow / ACP
    participant S as OpenAIService
    participant R as Runner.run_streamed
    participant OT as OpenAITurn
    participant UE as UnifiedEmitter
    participant AS as auto_send
    participant ST as StreamingService

    W->>S: run_agent_streamed_auto_send(input, created_at)
    S->>R: run_streamed(agent, input, [max_turns], [prev_resp_id])
    R-->>S: RunResultStreaming
    S->>OT: OpenAITurn(result, model)
    S->>UE: UnifiedEmitter(task_id, tracer, streaming)
    S->>UE: auto_send_turn(turn, created_at)
    UE->>AS: "auto_send(turn.events, created_at=created_at)"
    loop For each canonical event
        AS->>OT: _iter_events() → convert_openai_to_agentex_events
        OT-->>AS: "StreamTaskMessage* (Start / Delta / Done / Full)"
        AS->>ST: "streaming_task_message_context(created_at=created_at)"
        AS->>ST: stream_update / close
    end
    Note over OT: After last event: aggregate raw_responses → TurnUsage
    AS-->>UE: "TurnResult(final_text, usage=empty_default)"
    UE->>OT: turn.usage()
    OT-->>UE: TurnUsage (populated)
    UE-->>S: TurnResult (with usage)
    S-->>W: RunResultStreaming

%%{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 W as Workflow / ACP
    participant S as OpenAIService
    participant R as Runner.run_streamed
    participant OT as OpenAITurn
    participant UE as UnifiedEmitter
    participant AS as auto_send
    participant ST as StreamingService

    W->>S: run_agent_streamed_auto_send(input, created_at)
    S->>R: run_streamed(agent, input, [max_turns], [prev_resp_id])
    R-->>S: RunResultStreaming
    S->>OT: OpenAITurn(result, model)
    S->>UE: UnifiedEmitter(task_id, tracer, streaming)
    S->>UE: auto_send_turn(turn, created_at)
    UE->>AS: "auto_send(turn.events, created_at=created_at)"
    loop For each canonical event
        AS->>OT: _iter_events() → convert_openai_to_agentex_events
        OT-->>AS: "StreamTaskMessage* (Start / Delta / Done / Full)"
        AS->>ST: "streaming_task_message_context(created_at=created_at)"
        AS->>ST: stream_update / close
    end
    Note over OT: After last event: aggregate raw_responses → TurnUsage
    AS-->>UE: "TurnResult(final_text, usage=empty_default)"
    UE->>OT: turn.usage()
    OT-->>UE: TurnUsage (populated)
    UE-->>S: TurnResult (with usage)
    S-->>W: RunResultStreaming

_{Reviews (11): Last reviewed commit: "fix(openai): forward previous_response_i..." | Re-trigger Greptile}