feat(harness): unified harness surface — foundation (span derivation, delivery adapters, emitter)

[declan-scale]

What this is

Foundation (PRs 1–3 of the rollout) for a unified harness tracing/message-emitting surface: the Agentex StreamTaskMessage* stream is the single source of truth, and shared harness-independent machinery derives spans from it and delivers it over both channels:

• yield — pass the canonical stream through to the caller (sync HTTP ACP agents),
• auto-send — push to the task stream via adk.streaming (async + temporal agents, from inside an activity),

with tracing on by default (derived from the same stream) and overridable, and a unified TurnUsage/TurnResult shape for per-harness usage normalization.

Design: docs/superpowers/specs/2026-06-18-unified-harness-surface-design.md
Plan: docs/superpowers/plans/2026-06-18-unified-harness-surface-foundation.md

What's in src/agentex/lib/core/harness/

• types.py — StreamTaskMessage, OpenSpan/CloseSpan/SpanSignal, TurnUsage, TurnResult, HarnessTurn protocol.
• span_derivation.py — SpanDeriver: pure reducer (no adk dep), canonical stream → span signals. Tool span opens on the Done of a ToolRequestContent index, closes on the matching ToolResponseContent by tool_call_id; reasoning span open-on-Start / close-on-Done; parallel-safe; flush() closes unclosed spans.
• tracer.py — SpanTracer: best-effort adapter from span signals to adk.tracing (never raises; overridable; guarded make_logger).
• yield_delivery.py / auto_send.py — the two delivery adapters (both feed the same SpanDeriver/SpanTracer; finally-flush on early close/error).
• emitter.py — UnifiedEmitter: ties trace context + delivery + usage; default-on/overridable tracing; injectable tracing/streaming backends.
• conformance/ — shared conformance scaffold each future harness tap registers fixtures with.
• .github/workflows/harness-integration.yml — conformance CI job (via ./scripts/test) + an if: false live-matrix placeholder enabled by the migration PRs.

Scope / what's NOT here

Per-harness migration (pydantic-ai / langgraph / openai) and parser taps (claude-code / codex), plus their 3 e2e test agents each (sync/async/temporal), are future migration PRs (4–8) — not in this branch.

Quality gates

• 30 tests passing on Python 3.12 + 3.13 (via ./scripts/test tests/lib/core/harness/).
• pyright clean (0 errors/warnings), no # type: ignore in the package.
• Each task spec- + quality-reviewed; final whole-branch review passed with no Critical issues.

Follow-ups (filed)

• AGX1-373 (High) — make conformance assert true yield-vs-auto-send equivalence + reconcile Full tool-message wire shape (blocks migration backward-compat claims).
• AGX1-374 (Medium) — auto_send reasoning + mixed-ordering tests.
• AGX1-375 (Medium) — expose the surface via the public adk facade before the first consumer migration.
• AGX1-376 (Low) — widen CI paths to agentex.types; SpanTracer duplicate-open guard.
• AGX1-371 — deferred optional is_error on ToolResponseContent (tool-span error status).

Note: total diff is ~3k lines but ~1.6k of that is the spec + plan docs; the package code + tests + CI is ~1.4k. Reviewable per-commit (one commit per plan task).

🤖 Generated with Claude Code

Greptile Summary

This PR introduces the foundation layer for a unified harness tracing and message-emitting surface — SpanDeriver, SpanTracer, yield_delivery, auto_send, UnifiedEmitter, and shared conformance scaffolding — so that every future harness tap gets streaming, tracing, and usage normalization from a single set of shared components.

• Core event pipeline: SpanDeriver is a pure, adk-free reducer that maps the canonical StreamTaskMessage* stream to OpenSpan/CloseSpan signals; both delivery adapters (yield_events and auto_send) feed it identically, with a finally-flush covering early exits and errors.
• auto_send delivery: Now correctly routes events by index via ctx_map, handles Start(ToolRequestContent) (opens a streaming context), supports last-segment final_text semantics (resets on each new TextContent start), and accumulates text from Full(TextContent) — all verified by 10 dedicated tests.
• Deferred gaps: Turn-usage ordering in the emitter (turn.usage() evaluated before events drain), wire-shape consistency for Full tool messages, and the conformance suite's yield-vs-auto-send equivalence assertion are explicitly tracked in AGX1-373/374/375/377 for resolution in the per-harness migration PRs.

Confidence Score: 4/5

Safe to merge as a foundation layer; the deferred usage-ordering behaviour in the emitter means that real harness taps accumulating token counts during iteration will receive stale usage until the migration PRs align the call site.

The core event routing, span derivation, and delivery logic are correct and well-tested. The one known semantic gap — turn.usage() is evaluated as an argument before auto_send drains turn.events, so any real tap that accumulates usage during iteration returns stale data — is a present defect in the changed emitter, deliberately deferred to the migration PRs where real implementations will be provided.

emitter.py — usage collection ordering; span_derivation.py — _on_done double-open guard asymmetry

Important Files Changed

Filename	Overview
src/agentex/lib/core/harness/span_derivation.py	Pure reducer from stream events to span signals. Handles Start+Done and Full paths for tool/reasoning spans. Minor asymmetry: _on_full guards against double-open but _on_done does not, risking orphaned spans on malformed mixed streams.
src/agentex/lib/core/harness/auto_send.py	Async/temporal delivery adapter. Index-keyed ctx_map correctly handles interleaved streams, ToolRequestContent Start opens contexts, last-segment final_text semantics are in place, and finally-flush covers early exits. Previous review gaps have been addressed.
src/agentex/lib/core/harness/emitter.py	UnifiedEmitter facade wiring trace context, delivery, and usage. turn.usage() is evaluated as a positional argument before auto_send drains turn.events, which violates the HarnessTurn protocol contract; deferred to migration PRs.
src/agentex/lib/core/harness/tracer.py	Best-effort adapter over adk.tracing. Never raises, gracefully handles missing spans. Duplicate-open orphan behavior is documented and tracked (AGX1-376).
src/agentex/lib/core/harness/yield_delivery.py	Pure passthrough async generator with side-effect tracing. finally-flush correctly handles early consumer cancellation.
src/agentex/lib/core/harness/types.py	Clean type definitions: StreamTaskMessage union, OpenSpan/CloseSpan signals, TurnUsage, TurnResult, HarnessTurn protocol.
tests/lib/core/harness/conformance/runner.py	Process-global fixture registry with a documented cross-module ordering hazard. Correct for the current single-module setup; future harness taps are advised to parametrize their own fixtures.
tests/lib/core/harness/conformance/test_conformance.py	Conformance test only asserts derive_all idempotency, not correctness of emitted signals; true yield-vs-auto-send equivalence assertion is deferred to AGX1-373.
.github/workflows/harness-integration.yml	Conformance CI job correctly defers to ./scripts/test for isolated uv environment. Narrow paths filter (missing agentex.types) is a known gap filed as AGX1-376. Placeholder live-matrix job kept as if: false until migration PRs land.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant HT as HarnessTurn
    participant UE as UnifiedEmitter
    participant YD as yield_events / auto_send
    participant SD as SpanDeriver
    participant ST as SpanTracer
    participant S as adk.streaming

    UE->>YD: pass turn.events + tracer
    loop each StreamTaskMessage
        HT-->>YD: event (Start / Delta / Done / Full)
        YD->>SD: observe(event)
        SD-->>YD: []  or  [OpenSpan / CloseSpan]
        opt signal emitted
            YD->>ST: handle(signal)
            ST->>S: start_span / end_span
        end
        alt yield mode
            YD-->>UE: yield event
        else auto_send mode
            YD->>S: streaming_task_message_context + stream_update + close
        end
    end
    Note over YD,ST: finally: deriver.flush() closes unclosed spans
    YD-->>UE: TurnResult(final_text, usage)

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 HT as HarnessTurn
    participant UE as UnifiedEmitter
    participant YD as yield_events / auto_send
    participant SD as SpanDeriver
    participant ST as SpanTracer
    participant S as adk.streaming

    UE->>YD: pass turn.events + tracer
    loop each StreamTaskMessage
        HT-->>YD: event (Start / Delta / Done / Full)
        YD->>SD: observe(event)
        SD-->>YD: []  or  [OpenSpan / CloseSpan]
        opt signal emitted
            YD->>ST: handle(signal)
            ST->>S: start_span / end_span
        end
        alt yield mode
            YD-->>UE: yield event
        else auto_send mode
            YD->>S: streaming_task_message_context + stream_update + close
        end
    end
    Note over YD,ST: finally: deriver.flush() closes unclosed spans
    YD-->>UE: TurnResult(final_text, usage)

Loading

Comments Outside Diff (1)

1. General comment

   UnifiedEmitter ignores duck-typed tracer override and constructs real SpanTracer

   • Bug
     • Passing a custom tracer object with an async handle() method to UnifiedEmitter(..., tracer=custom_tracer) did not use that object. Instead, the emitter fell through to default construction of SpanTracer, which attempted to import the real ADK stack and failed in this environment with ModuleNotFoundError: No module named 'temporalio'. This contradicts the requested override behavior for tracing in the unified emitter surface.
   • Cause
     • src/agentex/lib/core/harness/emitter.py only accepts overrides when isinstance(tracer, SpanTracer) is true. A valid injected/duck-typed tracer object is ignored, causing fallback to SpanTracer(...) whenever trace_id is truthy.
   • Fix
     • Relax the override branch to accept any non-None, non-False tracer object that implements the expected handle(signal) contract, or define and use a runtime-checkable tracer Protocol instead of requiring isinstance(tracer, SpanTracer).

   _{Ran code and verified through T-Rex}

_{Reviews (7): Last reviewed commit: "feat(harness): thread created_at through..." | Re-trigger Greptile}

[declan-scale]

@greptile review

[greptile-apps]

turn.usage() called before events are exhausted

turn.usage() is evaluated as a Python argument before auto_send is awaited, meaning it runs before turn.events is ever iterated. The HarnessTurn protocol explicitly states that usage() is "valid only after events is exhausted." Any real implementation that accumulates token counts or cost while iterating events will always return stale/zero usage here. The test passes only because _Turn.usage() returns a pre-set value regardless of iteration state.

The fix is to call turn.usage() after auto_send has drained the stream.

Suggested change

	async def auto_send_turn(self, turn: HarnessTurn) -> TurnResult:
	"""Async/temporal delivery: push to the task stream, return TurnResult."""
	return await auto_send(
	turn.events,
	task_id=self.task_id,
	tracer=self.tracer,
	streaming=self._streaming,
	usage=turn.usage(),
	)
	async def auto_send_turn(self, turn: HarnessTurn) -> TurnResult:
	"""Async/temporal delivery: push to the task stream, return TurnResult."""
	result = await auto_send(
	turn.events,
	task_id=self.task_id,
	tracer=self.tracer,
	streaming=self._streaming,
	)
	usage = turn.usage() # valid now: events exhausted by auto_send
	return TurnResult(final_text=result.final_text, usage=usage)

Prompt To Fix With AI

This is a comment left during a code review.
Path: src/agentex/lib/core/harness/emitter.py
Line: 59-67

Comment:
**`turn.usage()` called before events are exhausted**

`turn.usage()` is evaluated as a Python argument before `auto_send` is awaited, meaning it runs before `turn.events` is ever iterated. The `HarnessTurn` protocol explicitly states that `usage()` is "valid only after `events` is exhausted." Any real implementation that accumulates token counts or cost while iterating events will always return stale/zero usage here. The test passes only because `_Turn.usage()` returns a pre-set value regardless of iteration state.

The fix is to call `turn.usage()` after `auto_send` has drained the stream.

```suggestion
    async def auto_send_turn(self, turn: HarnessTurn) -> TurnResult:
        """Async/temporal delivery: push to the task stream, return TurnResult."""
        result = await auto_send(
            turn.events,
            task_id=self.task_id,
            tracer=self.tracer,
            streaming=self._streaming,
        )
        usage = turn.usage()  # valid now: events exhausted by auto_send
        return TurnResult(final_text=result.final_text, usage=usage)
```

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

[declan-scale]

@greptile review