feat(harness): UnifiedEmitter facade tying delivery + tracing + usage

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: declan-scale

src/agentex/lib/core/harness/__init__.py

@@ -4,3 +4,27 @@
 package derives spans from it and delivers it (yield or auto-send), so every
 harness tap gets streaming + tracing + turn usage uniformly.
 """
+
+from agentex.lib.core.harness.emitter import UnifiedEmitter
+from agentex.lib.core.harness.tracer import SpanTracer
+from agentex.lib.core.harness.types import (
+    CloseSpan,
+    HarnessTurn,
+    OpenSpan,
+    SpanSignal,
+    StreamTaskMessage,
+    TurnResult,
+    TurnUsage,
+)
+
+__all__ = [
+    "UnifiedEmitter",
+    "SpanTracer",
+    "OpenSpan",
+    "CloseSpan",
+    "SpanSignal",
+    "StreamTaskMessage",
+    "TurnUsage",
+    "TurnResult",
+    "HarnessTurn",
+]

src/agentex/lib/core/harness/emitter.py

@@ -0,0 +1,59 @@
+"""UnifiedEmitter: the single facade agent authors use for either delivery mode."""
+
+from __future__ import annotations
+
+from typing import AsyncIterator
+
+from agentex.lib.core.harness.auto_send import auto_send
+from agentex.lib.core.harness.tracer import SpanTracer
+from agentex.lib.core.harness.types import HarnessTurn, StreamTaskMessage, TurnResult
+from agentex.lib.core.harness.yield_delivery import yield_events
+
+
+class UnifiedEmitter:
+    """Ties trace context + chosen delivery together.
+
+    Tracing is default-on whenever `trace_id` is truthy; pass `tracer=False` to
+    disable, or a custom `SpanTracer` to override.
+    """
+
+    tracer: SpanTracer | None
+
+    def __init__(
+        self,
+        task_id: str,
+        trace_id: str | None,
+        parent_span_id: str | None,
+        tracer: SpanTracer | bool | None = None,
+        tracing: object | None = None,
+    ):
+        self.task_id = task_id
+        self.trace_id = trace_id
+        self.parent_span_id = parent_span_id
+        if tracer is False:
+            self.tracer = None
+        elif isinstance(tracer, SpanTracer):
+            self.tracer = tracer
+        elif trace_id:
+            self.tracer = SpanTracer(
+                trace_id=trace_id,
+                parent_span_id=parent_span_id,
+                task_id=task_id,
+                tracing=tracing,
+            )
+        else:
+            self.tracer = None
+
+    async def yield_turn(self, turn: HarnessTurn) -> AsyncIterator[StreamTaskMessage]:
+        """Sync HTTP ACP delivery: forward events, trace as side effect."""
+        async for event in yield_events(turn.events, tracer=self.tracer):
+            yield event
+
+    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,
+            usage=turn.usage(),
+        )

tests/lib/core/harness/test_emitter.py

@@ -0,0 +1,56 @@
+import pytest
+
+from agentex.lib.core.harness.emitter import UnifiedEmitter
+from agentex.lib.core.harness.types import TurnUsage
+from agentex.types.task_message_update import StreamTaskMessageStart, StreamTaskMessageDone
+from agentex.types.text_content import TextContent
+
+
+class _FakeTracing:
+    async def start_span(self, **kw):
+        return None
+
+    async def end_span(self, **kw):
+        pass
+
+
+class _Turn:
+    def __init__(self, events_list, usage):
+        self._events_list = events_list
+        self._usage = usage
+
+    @property
+    async def events(self):
+        for e in self._events_list:
+            yield e
+
+    def usage(self):
+        return self._usage
+
+
+@pytest.mark.asyncio
+async def test_emitter_yield_mode_passes_through():
+    events = [
+        StreamTaskMessageStart(type="start", index=0,
+            content=TextContent(type="text", author="agent", content="hi")),
+        StreamTaskMessageDone(type="done", index=0),
+    ]
+    turn = _Turn(events, TurnUsage(model="m"))
+    emitter = UnifiedEmitter(task_id="t", trace_id=None, parent_span_id=None)
+    out = [e async for e in emitter.yield_turn(turn)]
+    assert out == events
+
+
+@pytest.mark.asyncio
+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())
+    assert emitter.tracer is not None
+
+
+@pytest.mark.asyncio
+async def test_emitter_tracing_overridable_off():
+    emitter = UnifiedEmitter(task_id="t", trace_id="trace1", parent_span_id="p", tracer=False)
+    assert emitter.tracer is None