refactor(harness): simplify yield_events guard + cover finally-flush on early close

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

Author: declan-scale

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

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import AsyncIterator
+from typing import AsyncGenerator, AsyncIterator
 
 from agentex.lib.core.harness.span_derivation import SpanDeriver
 from agentex.lib.core.harness.tracer import SpanTracer
@@ -12,7 +12,7 @@
 async def yield_events(
     events: AsyncIterator[StreamTaskMessage],
     tracer: SpanTracer | None = None,
-) -> AsyncIterator[StreamTaskMessage]:
+) -> AsyncGenerator[StreamTaskMessage, None]:
     """Forward each event to the caller; derive + trace spans as a side effect.
 
     For sync HTTP ACP agents that yield events back over the response. When
@@ -21,11 +21,11 @@ async def yield_events(
     deriver = SpanDeriver() if tracer is not None else None
     try:
         async for event in events:
-            if deriver is not None and tracer is not None:
+            if deriver is not None:  # tracer is non-None whenever deriver is set
                 for signal in deriver.observe(event):
                     await tracer.handle(signal)
             yield event
     finally:
-        if deriver is not None and tracer is not None:
+        if deriver is not None:  # tracer is non-None whenever deriver is set
             for signal in deriver.flush():
                 await tracer.handle(signal)

tests/lib/core/harness/test_yield_delivery.py

@@ -56,3 +56,22 @@ async def test_yield_without_tracer_is_pure_passthrough():
     ]
     out = [e async for e in yield_events(_gen(events), tracer=None)]
     assert out == events
+
+
+@pytest.mark.asyncio
+async def test_flush_runs_on_early_close():
+    fake = _RecordTracing()
+    tracer = SpanTracer(trace_id="t", parent_span_id="p", tracing=fake)
+    events = [
+        StreamTaskMessageStart(type="start", index=0,
+            content=ToolRequestContent(type="tool_request", author="agent",
+                                       tool_call_id="c", name="Bash", arguments={})),
+        StreamTaskMessageDone(type="done", index=0),
+        # response intentionally never arrives
+    ]
+    gen = yield_events(_gen(events), tracer=tracer)
+    first = await gen.__anext__()   # Start
+    second = await gen.__anext__()  # Done -> tool span opens here
+    await gen.aclose()              # triggers the finally -> flush()
+    assert fake.started == ["Bash"]
+    assert fake.ended == [None]     # flush closed the unpaired span (incomplete, no output)