feat(langgraph): migrate LangGraph harness onto unified surface

[declan-scale]

Summary

Migrates the LangGraph harness onto the unified harness surface introduced in PR 4 (pydantic-ai). Implements 12 tasks covering the new LangGraphTurn adapter, bespoke helper rewrites, offline integration tests, conformance fixtures, tutorial agents, and CI matrix.

New surface:

turn = LangGraphTurn(stream, model=model_name)
# Sync HTTP ACP
async for event in emitter.yield_turn(turn):
    yield event
# Async / temporal
result = await emitter.auto_send_turn(turn)

Key implementation points:

• LangGraphTurn wraps LangGraph astream() and implements HarnessTurn (tasks 1-2)
• stream_langgraph_events reimplemented on UnifiedEmitter (task 4)
• _langgraph_tracing.py create_langgraph_tracing_handler marked deprecated with warnings.warn(DeprecationWarning) (task 3)
• AGX1-377 documented: LangGraph emits tool requests as StreamTaskMessageFull (not Start+Delta+Done); SpanDeriver does not produce tool spans from Full events today (tracked in AGX1-373)
• Usage timing: LangGraphTurn.usage() is populated via on_final_ai_message callback during event iteration; TurnResult.usage is a pre-iteration snapshot — callers should read turn.usage() after auto_send_turn returns
• Added AsyncGenerator return type annotation to convert_langgraph_to_agentex_events and _generate_events to fix pyright inference (was treating them as coroutines)

Tests added (tasks 5-8, 219 passing):

• test_langgraph_sync.py: 11 unit tests for convert_langgraph_to_agentex_events + deprecation
• test_langgraph_turn.py: 19 unit tests for LangGraphTurn + langgraph_usage_to_turn_usage
• test_langgraph_async.py: 6 characterization tests for the unified stream_langgraph_events
• test_langgraph_sync_unified.py: 6 passthrough + span derivation tests
• test_langgraph_conformance.py: 4 conformance fixtures (text-only, single-tool, reasoning, multi-step)
• test_harness_langgraph_sync.py: 6 offline integration tests (yield channel)
• test_harness_langgraph_async.py: 7 offline integration tests (auto_send channel)
• test_harness_langgraph_temporal.py: 5 offline integration tests (temporal channel)

Tutorial agents (task 9):

• examples/tutorials/00_sync/harness_langgraph/ (s-harness-langgraph) — sync, yield_turn
• examples/tutorials/10_async/00_base/harness_langgraph/ (a-harness-langgraph) — async, auto_send_turn
• examples/tutorials/10_async/10_temporal/harness_langgraph/ (at-harness-langgraph) — temporal, LangGraphPlugin + emit_langgraph_messages

CI (task 10): Enabled live-matrix job in harness-integration.yml with 3-way matrix over [sync, async, temporal] running offline LangGraph integration tests.

Test plan

• uv run --all-packages --all-extras pytest tests/lib/core/harness/ tests/lib/adk/ -v — 219 passed
• ./scripts/lint — 0 errors, 0 warnings (ruff + pyright)
• Live agent smoke test (requires running AgentEx server + LLM keys)

🤖 Generated with Claude Code

Greptile Summary

Migrates the LangGraph harness onto the unified LangGraphTurn + UnifiedEmitter surface introduced for pydantic-ai, eliminating ~165 lines of bespoke Redis-streaming code and aligning all three channels (sync HTTP yield, async Redis, Temporal) behind the same adapter pattern.

• LangGraphTurn (new): wraps graph.astream() and accumulates token usage additively across multi-step LLM calls via _accumulate_turn_usage, fixing the silent per-call overwrite from the old bespoke path.
• _langgraph_sync.py: adds on_final_ai_message callback for lazy usage capture and fixes the reasoning-block StreamTaskMessageStart content type from TextContent to ReasoningContent (matching the conformance fixture and downstream schema expectations).
• CI matrix extended to [pydantic_ai, langgraph] × [sync, async, temporal] with 219 new offline tests covering unit, integration, and conformance layers.

Confidence Score: 5/5

Safe to merge — core logic is correct, all three channels work, and the one reasoning-content-type fix is guarded by a new assertion in the test suite.

The implementation is well-structured: bespoke Redis code is cleanly replaced, multi-step usage now accumulates additively instead of overwriting, and the reasoning-block content-type fix is confirmed by a targeted test. The only finding is a stale docstring claiming SpanDeriver does not handle Full tool events, while the tests in this same PR assert it does — a documentation inconsistency that does not affect runtime behavior.

The module docstring in tests/lib/core/harness/test_harness_langgraph_async.py and both acp.py tutorial files carry a stale claim about SpanDeriver not producing tool spans from Full events; those comments should be updated before they mislead future contributors.

Important Files Changed

Filename	Overview
src/agentex/lib/adk/_modules/_langgraph_turn.py	New LangGraphTurn HarnessTurn adapter — correctly accumulates usage across multi-step LLM calls via _accumulate_turn_usage; the known eager-snapshot limitation for TurnResult.usage is documented.
src/agentex/lib/adk/_modules/_langgraph_sync.py	Fixes reasoning-block Start content type (now ReasoningContent instead of TextContent); adds on_final_ai_message callback for usage capture; overall event sequencing is correct.
src/agentex/lib/adk/_modules/_langgraph_async.py	Stripped 165 lines of bespoke Redis-streaming code and replaced with a two-line LangGraphTurn + UnifiedEmitter.auto_send_turn delegation; public signature preserved.
src/agentex/lib/adk/_modules/_langgraph_tracing.py	Marked deprecated via docstrings only (no runtime warnings.warn); test confirms no DeprecationWarning is emitted, keeping callers safe under -W error.
tests/lib/core/harness/test_harness_langgraph_async.py	Good coverage of the async auto_send channel; module docstring incorrectly states that tool spans are not produced, contradicting the passing test assertions.
tests/lib/core/harness/test_harness_langgraph_sync.py	Thorough coverage of sync yield channel — text, tool calls, multi-step, tracing, and usage capture all tested.
tests/lib/core/harness/conformance/test_langgraph_conformance.py	Four canonical fixtures (text-only, single-tool, reasoning, multi-step) registered with the cross-channel runner; span-derivation determinism guard also added.
.github/workflows/harness-integration.yml	Extends CI matrix to [pydantic_ai, langgraph] x [sync, async, temporal]; path triggers and job names updated consistently.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant Agent as Agent ACP Handler
    participant LGT as LangGraphTurn
    participant Conv as convert_langgraph_to_agentex_events
    participant UE as UnifiedEmitter
    participant SD as SpanDeriver
    participant Backend as Streaming Backend (Redis / HTTP)

    Agent->>LGT: LangGraphTurn(graph.astream(), model)
    Agent->>UE: yield_turn(turn) OR auto_send_turn(turn)

    UE->>LGT: iterate turn.events
    LGT->>Conv: "convert_langgraph_to_agentex_events(stream, on_final_ai_message=_capture)"

    loop LangGraph messages events
        Conv-->>LGT: StreamTaskMessageStart / Delta / Done (text or reasoning)
        LGT-->>UE: event
        UE->>SD: derive span signal
        UE-->>Backend: stream text delta
    end

    loop LangGraph updates events
        Conv-->>LGT: StreamTaskMessageFull(ToolRequestContent)
        LGT-->>UE: Full event
        UE->>SD: open tool span
        UE-->>Backend: Full message
        Conv->>Conv: on_final_ai_message(_capture) accumulate TurnUsage
        Conv-->>LGT: StreamTaskMessageFull(ToolResponseContent)
        LGT-->>UE: Full event
        UE->>SD: close tool span
        UE-->>Backend: Full message
    end

    UE-->>Agent: TurnResult(final_text) / yield events
    Agent->>LGT: turn.usage() accumulated 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 Agent as Agent ACP Handler
    participant LGT as LangGraphTurn
    participant Conv as convert_langgraph_to_agentex_events
    participant UE as UnifiedEmitter
    participant SD as SpanDeriver
    participant Backend as Streaming Backend (Redis / HTTP)

    Agent->>LGT: LangGraphTurn(graph.astream(), model)
    Agent->>UE: yield_turn(turn) OR auto_send_turn(turn)

    UE->>LGT: iterate turn.events
    LGT->>Conv: "convert_langgraph_to_agentex_events(stream, on_final_ai_message=_capture)"

    loop LangGraph messages events
        Conv-->>LGT: StreamTaskMessageStart / Delta / Done (text or reasoning)
        LGT-->>UE: event
        UE->>SD: derive span signal
        UE-->>Backend: stream text delta
    end

    loop LangGraph updates events
        Conv-->>LGT: StreamTaskMessageFull(ToolRequestContent)
        LGT-->>UE: Full event
        UE->>SD: open tool span
        UE-->>Backend: Full message
        Conv->>Conv: on_final_ai_message(_capture) accumulate TurnUsage
        Conv-->>LGT: StreamTaskMessageFull(ToolResponseContent)
        LGT-->>UE: Full event
        UE->>SD: close tool span
        UE-->>Backend: Full message
    end

    UE-->>Agent: TurnResult(final_text) / yield events
    Agent->>LGT: turn.usage() accumulated TurnUsage

Loading

Comments Outside Diff (1)

1. src/agentex/lib/adk/_modules/_langgraph_sync.py, line 147-153 (link)

   Reasoning block StreamTaskMessageStart uses wrong content type

   When a reasoning model emits a block of type "reasoning", the code opens the stream with TextContent(type="text", ...) instead of ReasoningContent. Downstream consumers that dispatch on content.type (e.g. rendering pipelines, the SpanDeriver text-span logic) will receive a TextContent wrapper for what is actually a reasoning block, then see a ReasoningContentDelta arrive — a type mismatch that will confuse or break those consumers. ReasoningContent is also not imported in this file, confirming the intended type was never used. The conformance fixture _REASONING correctly shows ReasoningContent as the expected start content, but it constructs the events by hand and never runs them through the actual converter, so no test catches this today.

   Prompt To Fix With AI

   This is a comment left during a code review.
   Path: src/agentex/lib/adk/_modules/_langgraph_sync.py
   Line: 147-153
   
   Comment:
   **Reasoning block `StreamTaskMessageStart` uses wrong content type**
   
   When a reasoning model emits a block of type `"reasoning"`, the code opens the stream with `TextContent(type="text", ...)` instead of `ReasoningContent`. Downstream consumers that dispatch on `content.type` (e.g. rendering pipelines, the `SpanDeriver` text-span logic) will receive a `TextContent` wrapper for what is actually a reasoning block, then see a `ReasoningContentDelta` arrive — a type mismatch that will confuse or break those consumers. `ReasoningContent` is also not imported in this file, confirming the intended type was never used. The conformance fixture `_REASONING` correctly shows `ReasoningContent` as the expected start content, but it constructs the events by hand and never runs them through the actual converter, so no test catches this today.
   
   How can I resolve this? If you propose a fix, please make it concise.

   

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

[greptile-apps]

TurnResult.usage is always empty when using auto_send_turn

LangGraphTurn populates self._usage lazily via the on_final_ai_message callback, which fires during event iteration. However, UnifiedEmitter.auto_send_turn passes usage=turn.usage() as an argument to auto_send before iteration begins (Python evaluates all arguments before the call). By the time the stream is consumed and _capture updates self._usage, the pre-iteration snapshot has already been handed to TurnResult.

Concretely: every caller that reads result.usage after await emitter.auto_send_turn(turn) gets TurnUsage(model=model) — zero token counts regardless of what the model reported. The PR description documents the workaround ("callers should read turn.usage() after auto_send_turn returns"), but TurnResult.usage existing with silent stale data is a trap for every future user of this API.

The fix belongs in emitter.py: call turn.usage() after await auto_send(turn.events, ...) returns, then construct the TurnResult from the now-populated usage.

Prompt To Fix With AI

This is a comment left during a code review.
Path: src/agentex/lib/adk/_modules/_langgraph_turn.py
Line: 95-119

Comment:
**`TurnResult.usage` is always empty when using `auto_send_turn`**

`LangGraphTurn` populates `self._usage` lazily via the `on_final_ai_message` callback, which fires _during_ event iteration. However, `UnifiedEmitter.auto_send_turn` passes `usage=turn.usage()` as an argument to `auto_send` _before_ iteration begins (Python evaluates all arguments before the call). By the time the stream is consumed and `_capture` updates `self._usage`, the pre-iteration snapshot has already been handed to `TurnResult`.

Concretely: every caller that reads `result.usage` after `await emitter.auto_send_turn(turn)` gets `TurnUsage(model=model)` — zero token counts regardless of what the model reported. The PR description documents the workaround ("callers should read `turn.usage()` after `auto_send_turn` returns"), but `TurnResult.usage` existing with silent stale data is a trap for every future user of this API.

The fix belongs in `emitter.py`: call `turn.usage()` _after_ `await auto_send(turn.events, ...)` returns, then construct the `TurnResult` from the now-populated usage.

How can I resolve this? If you propose a fix, please make it concise.

[declan-scale]

@greptile review