refactor(harness): inject streaming into UnifiedEmitter + cover auto_send_turn + doc tracer modes

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

Author: declan-scale

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

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import AsyncIterator
+from typing import AsyncGenerator
 
 from agentex.lib.core.harness.auto_send import auto_send
 from agentex.lib.core.harness.tracer import SpanTracer
@@ -13,8 +13,13 @@
 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.
+    Tracing modes (the `tracer` arg):
+    - tracer=None (default): auto-construct a SpanTracer if `trace_id` is present.
+    - tracer=False: disable tracing entirely, regardless of `trace_id`.
+    - tracer=<SpanTracer>: use the supplied instance.
+
+    `tracing` and `streaming` are injection escape-hatches for tests/advanced
+    use; leave them None in production so the real adk modules are used.
     """
 
     tracer: SpanTracer | None
@@ -26,10 +31,12 @@ def __init__(
         parent_span_id: str | None,
         tracer: SpanTracer | bool | None = None,
         tracing: object | None = None,
+        streaming: object | None = None,
     ):
         self.task_id = task_id
         self.trace_id = trace_id
         self.parent_span_id = parent_span_id
+        self._streaming = streaming
         if tracer is False:
             self.tracer = None
         elif isinstance(tracer, SpanTracer):
@@ -44,7 +51,7 @@ def __init__(
         else:
             self.tracer = None
 
-    async def yield_turn(self, turn: HarnessTurn) -> AsyncIterator[StreamTaskMessage]:
+    async def yield_turn(self, turn: HarnessTurn) -> AsyncGenerator[StreamTaskMessage, None]:
         """Sync HTTP ACP delivery: forward events, trace as side effect."""
         async for event in yield_events(turn.events, tracer=self.tracer):
             yield event
@@ -55,5 +62,6 @@ async def auto_send_turn(self, turn: HarnessTurn) -> TurnResult:
             turn.events,
             task_id=self.task_id,
             tracer=self.tracer,
+            streaming=self._streaming,
             usage=turn.usage(),
         )

tests/lib/core/harness/test_emitter.py

@@ -2,7 +2,13 @@
 
 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.task_message import TaskMessage
+from agentex.types.task_message_delta import TextDelta
+from agentex.types.task_message_update import (
+    StreamTaskMessageDelta,
+    StreamTaskMessageDone,
+    StreamTaskMessageStart,
+)
 from agentex.types.text_content import TextContent
 
 
@@ -14,6 +20,42 @@ async def end_span(self, **kw):
         pass
 
 
+class _FakeCtx:
+    """Minimal StreamingTaskMessageContext fake (see test_auto_send.py)."""
+
+    def __init__(self, sink, content_type, initial_content):
+        self.sink = sink
+        self.content_type = content_type
+        self.task_message = TaskMessage(id="msg-1", task_id="task1", content=initial_content)
+
+    async def __aenter__(self):
+        self.sink.append(("open", self.content_type))
+        return self
+
+    async def __aexit__(self, *a):
+        await self.close()
+        return False
+
+    async def close(self):
+        self.sink.append(("close", self.content_type))
+
+    async def stream_update(self, update):
+        self.sink.append(("update", update))
+        return update
+
+
+class _FakeStreaming:
+    def __init__(self):
+        self.sink = []
+
+    def streaming_task_message_context(
+        self, task_id, initial_content, streaming_mode="coalesced", created_at=None
+    ):
+        ctype = getattr(initial_content, "type", None)
+        self.sink.append(("ctx", ctype))
+        return _FakeCtx(self.sink, ctype, initial_content)
+
+
 class _Turn:
     def __init__(self, events_list, usage):
         self._events_list = events_list
@@ -54,3 +96,21 @@ async def test_emitter_tracing_default_on_when_trace_id_present():
 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
+
+
+@pytest.mark.asyncio
+async def test_emitter_auto_send_turn_returns_usage():
+    usage = TurnUsage(model="m", input_tokens=5)
+    events = [
+        StreamTaskMessageStart(type="start", index=0,
+            content=TextContent(type="text", author="agent", content="")),
+        StreamTaskMessageDelta(type="delta", index=0,
+            delta=TextDelta(type="text", text_delta="Hello")),
+        StreamTaskMessageDone(type="done", index=0),
+    ]
+    turn = _Turn(events, usage)
+    fake = _FakeStreaming()
+    emitter = UnifiedEmitter(task_id="t", trace_id=None, parent_span_id=None, streaming=fake)
+    result = await emitter.auto_send_turn(turn)
+    assert result.usage == usage
+    assert result.final_text == "Hello"