test(harness): extract shared FakeSpan/FakeTracing + parametrize determinism test once

Adds tests/lib/core/harness/_fakes.py with a single superset FakeSpan/FakeTracing
(started/ended/ended_spans plus started_names/started_pairs/ended_outputs views) and
migrates every consumer off its local copy. Keeps the conformance determinism test
once (parametrized over all_fixtures) and drops the per-harness copies. run_yield_turn
and test_langgraph_sync_unified's _FakeTracingBackend left in place (genuinely divergent).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: declan-scale

tests/lib/adk/test_pydantic_ai_sync_unified.py

@@ -25,6 +25,7 @@
 )
 
 from agentex.lib.core.harness import UnifiedEmitter
+from tests.lib.core.harness._fakes import FakeTracing
 from agentex.lib.adk._modules._pydantic_ai_turn import PydanticAITurn
 
 
@@ -37,25 +38,6 @@ async def _collect(stream: AsyncIterator[Any]) -> list[Any]:
     return [e async for e in stream]
 
 
-class _FakeSpan:
-    def __init__(self, name: str):
-        self.name = name
-        self.output: Any = None
-
-
-class _FakeTracing:
-    def __init__(self) -> None:
-        self.started: list[tuple[str, str | None, Any]] = []
-        self.ended: list[tuple[str, Any]] = []
-
-    async def start_span(self, *, trace_id, name, input=None, parent_id=None, data=None, task_id=None):
-        self.started.append((name, parent_id, input))
-        return _FakeSpan(name)
-
-    async def end_span(self, *, trace_id, span):
-        self.ended.append((span.name, span.output))
-
-
 def _make_result_event(usage: RunUsage | None = None) -> AgentRunResultEvent:
     result = AgentRunResult(output="done", _output_tool_name=None)
     if usage is not None:
@@ -129,7 +111,7 @@ async def test_tool_span_opened_and_closed(self):
             ),
         ]
 
-        fake = _FakeTracing()
+        fake = FakeTracing()
         turn = PydanticAITurn(_aiter(tool_events), model="openai:gpt-4o")
         emitter = UnifiedEmitter(task_id="t", trace_id="tr", parent_span_id="p", tracing=fake)
 
@@ -152,7 +134,7 @@ async def test_reasoning_span_opened_and_closed(self):
             PartEndEvent(index=0, part=ThinkingPart(content="let me think")),
         ]
 
-        fake = _FakeTracing()
+        fake = FakeTracing()
         turn = PydanticAITurn(_aiter(reasoning_events), model="openai:gpt-4o")
         emitter = UnifiedEmitter(task_id="t", trace_id="tr", parent_span_id="p", tracing=fake)
 
@@ -177,7 +159,7 @@ async def test_no_trace_id_means_no_spans(self):
             ),
         ]
 
-        fake = _FakeTracing()
+        fake = FakeTracing()
         turn = PydanticAITurn(_aiter(raw_events), model="openai:gpt-4o")
         emitter = UnifiedEmitter(task_id="t", trace_id=None, parent_span_id=None, tracing=fake)
 
@@ -199,7 +181,7 @@ async def test_tracer_false_suppresses_spans_even_with_trace_id(self):
             ),
         ]
 
-        fake = _FakeTracing()
+        fake = FakeTracing()
         turn = PydanticAITurn(_aiter(raw_events), model="openai:gpt-4o")
         emitter = UnifiedEmitter(task_id="t", trace_id="tr", parent_span_id="p", tracer=False, tracing=fake)
 

tests/lib/core/harness/_fakes.py

@@ -0,0 +1,63 @@
+"""Shared test doubles for the unified harness test suites.
+
+A single superset implementation of the in-memory tracing backend used across
+the harness tests. Three recording shapes were previously duplicated:
+
+- Shape-1 (richest): ``started`` = ``(name, parent_id, input)`` 3-tuples,
+  ``ended`` = ``(name, output)`` 2-tuples, plus an ``ended_spans`` list of the
+  closed ``FakeSpan`` objects (which carry ``.name``, ``.output``, ``.data``).
+- Shape-2: ``started`` = ``(name, parent_id)`` 2-tuples, ``ended`` =
+  ``(name, output)``.
+- Shape-3: ``started`` = bare names, ``ended`` = bare outputs.
+
+``FakeTracing`` records the richest (shape-1) form and exposes read-only
+convenience properties (``started_names``, ``started_pairs``,
+``ended_outputs``) so shape-2 and shape-3 assertions stay clean.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+class FakeSpan:
+    def __init__(self, name: str) -> None:
+        self.name = name
+        self.output: Any = None
+        self.data: Any = None
+
+
+class FakeTracing:
+    def __init__(self) -> None:
+        self.started: list[tuple[str, Any, Any]] = []
+        self.ended: list[tuple[str, Any]] = []
+        self.ended_spans: list[FakeSpan] = []
+
+    async def start_span(
+        self,
+        *,
+        trace_id: str,
+        name: str,
+        input: Any = None,
+        parent_id: Any = None,
+        data: Any = None,
+        task_id: Any = None,
+    ) -> FakeSpan:
+        self.started.append((name, parent_id, input))
+        return FakeSpan(name)
+
+    async def end_span(self, *, trace_id: str, span: FakeSpan) -> None:
+        self.ended.append((span.name, span.output))
+        self.ended_spans.append(span)
+
+    @property
+    def started_names(self) -> list[str]:
+        return [name for (name, _parent, _input) in self.started]
+
+    @property
+    def started_pairs(self) -> list[tuple[str, Any]]:
+        return [(name, parent) for (name, parent, _input) in self.started]
+
+    @property
+    def ended_outputs(self) -> list[Any]:
+        return [output for (_name, output) in self.ended]

tests/lib/core/harness/conformance/runner.py

@@ -64,12 +64,12 @@
 from __future__ import annotations
 
 import json
-import types as _types
 from typing import Any, NamedTuple, override
 from dataclasses import dataclass
 
 from agentex.types.text_delta import TextDelta
 from agentex.types.task_message import TaskMessage
+from tests.lib.core.harness._fakes import FakeTracing
 from agentex.lib.core.harness.types import SpanSignal, StreamTaskMessage
 from agentex.lib.core.harness.tracer import SpanTracer
 from agentex.types.task_message_update import (
@@ -296,30 +296,6 @@ def streaming_task_message_context(
         return _FakeCtx(self.sink, ctype, initial_content)
 
 
-class _FakeTracing:
-    """Minimal tracing backend: records started/ended span names + outputs."""
-
-    def __init__(self) -> None:
-        self.started: list[str] = []
-        self.ended: list[Any] = []
-
-    async def start_span(
-        self,
-        *,
-        trace_id: str,
-        name: str,
-        input: Any = None,
-        parent_id: Any = None,
-        data: Any = None,
-        task_id: Any = None,
-    ) -> Any:
-        self.started.append(name)
-        return _types.SimpleNamespace()
-
-    async def end_span(self, *, trace_id: str, span: Any) -> None:
-        self.ended.append(getattr(span, "output", None))
-
-
 class _RecordingTracer(SpanTracer):
     """SpanTracer that records every SpanSignal it actually receives.
 
@@ -486,7 +462,7 @@ async def run_cross_channel_conformance(
     from agentex.lib.core.harness.yield_delivery import yield_events
 
     # --- yield channel ---
-    tracer_yield = _RecordingTracer(tracing=_FakeTracing())
+    tracer_yield = _RecordingTracer(tracing=FakeTracing())
     yield_out = [e async for e in yield_events(_gen(fixture.events), tracer=tracer_yield)]
 
     # Span signals the yield channel actually emitted to its tracer
@@ -496,7 +472,7 @@ async def run_cross_channel_conformance(
     yield_deliveries = _yield_text_reasoning_seq(_yield_logical_deliveries(yield_out))
 
     # --- auto_send channel ---
-    tracer_auto = _RecordingTracer(tracing=_FakeTracing())
+    tracer_auto = _RecordingTracer(tracing=FakeTracing())
     fake_streaming = _FakeStreaming()
     await auto_send(
         _gen(fixture.events),

tests/lib/core/harness/conformance/test_codex_conformance.py

@@ -19,7 +19,7 @@
 from agentex.lib.core.harness.types import StreamTaskMessage
 from agentex.lib.adk._modules._codex_sync import convert_codex_to_agentex_events
 
-from .runner import Fixture, register, derive_all
+from .runner import Fixture, register
 
 
 async def _aiter(items: list[Any]) -> AsyncIterator[Any]:
@@ -208,17 +208,6 @@ def _build(events: list[Any]) -> list[StreamTaskMessage]:
 _LOCAL_FIXTURES = [_CODEX_TEXT, _CODEX_TOOL, _CODEX_REASONING, _CODEX_MULTI]
 
 
-@pytest.mark.parametrize("fixture", _LOCAL_FIXTURES, ids=lambda f: f.name)
-def test_codex_span_derivation_is_deterministic(fixture: Fixture) -> None:
-    """Span derivation over codex events is deterministic (cross-channel guarantee).
-
-    Deriving twice over the same events yields identical signals. This is the
-    invariant that makes ``yield`` and ``auto_send`` delivery equivalent: both
-    observe the same event stream, so their tracing side effects are identical.
-    """
-    assert derive_all(fixture.events) == derive_all(fixture.events)
-
-
 @pytest.mark.parametrize("fixture", _LOCAL_FIXTURES, ids=lambda f: f.name)
 def test_codex_events_are_non_empty(fixture: Fixture) -> None:
     """Every codex fixture yields at least one StreamTaskMessage*."""

tests/lib/core/harness/conformance/test_langgraph_conformance.py

@@ -32,7 +32,7 @@
 from agentex.types.tool_response_content import ToolResponseContent
 from agentex.types.reasoning_content_delta import ReasoningContentDelta
 
-from .runner import Fixture, register, derive_all, run_cross_channel_conformance
+from .runner import Fixture, register, run_cross_channel_conformance
 
 # ---------------------------------------------------------------------------
 # Fixtures
@@ -216,14 +216,3 @@ async def test_cross_channel_equivalence(fixture: Fixture) -> None:
     assert yield_spans == auto_spans, (
         f"[{fixture.name}] span signals differ:\n  yield:     {yield_spans}\n  auto_send: {auto_spans}"
     )
-
-
-# ---------------------------------------------------------------------------
-# Backward-compatible determinism guard
-# ---------------------------------------------------------------------------
-
-
-@pytest.mark.parametrize("fixture", _LANGGRAPH_FIXTURES, ids=lambda f: f.name)
-def test_span_derivation_is_deterministic(fixture: Fixture) -> None:
-    """Span derivation over the same event list is idempotent."""
-    assert derive_all(fixture.events) == derive_all(fixture.events)

tests/lib/core/harness/conformance/test_pydantic_ai_conformance.py

@@ -39,7 +39,6 @@
 from .runner import (
     Fixture,
     register,
-    derive_all,
     run_cross_channel_conformance,
 )
 
@@ -181,14 +180,3 @@ async def test_cross_channel_equivalence(fixture: Fixture) -> None:
     assert yield_spans == auto_spans, (
         f"[{fixture.name}] span signals differ:\n  yield:     {yield_spans}\n  auto_send: {auto_spans}"
     )
-
-
-# ---------------------------------------------------------------------------
-# Backward-compatible determinism guard
-# ---------------------------------------------------------------------------
-
-
-@pytest.mark.parametrize("fixture", _FIXTURES, ids=lambda f: f.name)
-def test_span_derivation_is_deterministic(fixture: Fixture) -> None:
-    """Span derivation over the same event list is idempotent."""
-    assert derive_all(fixture.events) == derive_all(fixture.events)

tests/lib/core/harness/test_auto_send.py

@@ -9,13 +9,13 @@
 This mirrors _langgraph_async.py lines 62-78 and 100-127.
 """
 
-import types as _types
 from datetime import datetime
 
 import pytest
 
 from agentex.types.task_message import TaskMessage
 from agentex.types.text_content import TextContent
+from tests.lib.core.harness._fakes import FakeTracing
 from agentex.lib.core.harness.tracer import SpanTracer
 from agentex.types.task_message_delta import TextDelta
 from agentex.types.tool_request_delta import ToolRequestDelta
@@ -181,21 +181,9 @@ async def test_auto_send_posts_full_tool_messages():
 # ---------------------------------------------------------------------------
 
 
-class _RecordTracing:
-    def __init__(self):
-        self.started, self.ended = [], []
-
-    async def start_span(self, *, trace_id, name, input=None, parent_id=None, data=None, task_id=None):
-        self.started.append(name)
-        return _types.SimpleNamespace()
-
-    async def end_span(self, *, trace_id, span):
-        self.ended.append(getattr(span, "output", None))
-
-
 @pytest.mark.asyncio
 async def test_auto_send_derives_tool_spans_via_tracer():
-    fake_tracing = _RecordTracing()
+    fake_tracing = FakeTracing()
     tracer = SpanTracer(trace_id="t", parent_span_id="p", tracing=fake_tracing)
     streaming = _FakeStreaming()
 
@@ -228,8 +216,8 @@ async def test_auto_send_derives_tool_spans_via_tracer():
     result = await auto_send(_gen(events), task_id="task1", tracer=tracer, streaming=streaming)
 
     assert result.final_text == ""
-    assert fake_tracing.started == ["Bash"]
-    assert fake_tracing.ended == ["ok"]
+    assert fake_tracing.started_names == ["Bash"]
+    assert fake_tracing.ended_outputs == ["ok"]
 
 
 # ---------------------------------------------------------------------------

tests/lib/core/harness/test_emitter.py

@@ -2,6 +2,7 @@
 
 from agentex.types.task_message import TaskMessage
 from agentex.types.text_content import TextContent
+from tests.lib.core.harness._fakes import FakeTracing
 from agentex.lib.core.harness.types import TurnUsage
 from agentex.lib.core.harness.emitter import UnifiedEmitter
 from agentex.types.task_message_delta import TextDelta
@@ -12,14 +13,6 @@
 )
 
 
-class _FakeTracing:
-    async def start_span(self, **kw):
-        return None
-
-    async def end_span(self, **kw):
-        pass
-
-
 class _FakeCtx:
     """Minimal StreamingTaskMessageContext fake (see test_auto_send.py)."""
 
@@ -84,7 +77,7 @@ async def test_emitter_yield_mode_passes_through():
 async def test_emitter_tracing_default_on_when_trace_id_present():
     # Inject a fake tracing backend so the test env doesn't need temporalio.
     # This exercises the default-on path (tracer=None) when trace_id is truthy.
-    emitter = UnifiedEmitter(task_id="t", trace_id="trace1", parent_span_id="p", tracing=_FakeTracing())
+    emitter = UnifiedEmitter(task_id="t", trace_id="trace1", parent_span_id="p", tracing=FakeTracing())
     assert emitter.tracer is not None