fix(harness): coerce non-dict span input/output so spans aren't dropped

The SGP spans API requires a span's input/output to be an object: a scalar or
string payload is rejected with a 422 ("Input should be a valid dictionary") and
the async processor drops the span. The reasoning-span fix records the
chain-of-thought as a plain-string output, so reasoning spans were silently
dropped — observed as "0 reasoning traces" in the golden agent while tool spans
(dict output) survived. Some harnesses' tool results are also plain strings and
would hit the same 422.

Coerce at the tracer boundary: wrap any non-dict input/output under a single key
({"output": ...} / {"input": ...}); dicts and None pass through unchanged.

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

Author: declan-scale

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

@@ -16,6 +16,21 @@
     logger = logging.getLogger(__name__)
 
 
+def _as_span_payload(value: Any, *, key: str) -> Any:
+    """Coerce a span input/output payload into a dict.
+
+    The SGP spans API requires ``input`` and ``output`` to be objects: a scalar
+    or string is rejected with a 422 and the span is dropped by the async
+    processor. The SpanDeriver legitimately produces non-dict payloads — the
+    reasoning span's output is the chain-of-thought string, and some harnesses'
+    tool results are plain strings — so wrap anything that isn't already a dict
+    (``None`` passes through unchanged so an absent payload stays absent).
+    """
+    if value is None or isinstance(value, dict):
+        return value
+    return {key: value}
+
+
 class SpanTracer:
     """Opens/closes adk.tracing child spans in response to span signals.
 
@@ -60,7 +75,7 @@ async def handle(self, signal: SpanSignal) -> None:
                 span = await self._tracing.start_span(
                     trace_id=self.trace_id,
                     name=signal.name,
-                    input=signal.input,
+                    input=_as_span_payload(signal.input, key="input"),
                     parent_id=self.parent_span_id,
                     task_id=self.task_id,
                 )
@@ -73,7 +88,7 @@ async def handle(self, signal: SpanSignal) -> None:
                     # The real TracingModule.end_span signature is:
                     #   end_span(trace_id, span, start_to_close_timeout, heartbeat_timeout, retry_policy)
                     # It does not accept an output= kwarg.
-                    span.output = signal.output
+                    span.output = _as_span_payload(signal.output, key="output")
                     # Tool failure status (ToolResponseContent.is_error) is recorded
                     # on span.data when the harness reports one; Span has no dedicated
                     # error field. None means no status was reported, so leave data alone.

tests/lib/core/harness/test_auto_send.py

@@ -218,7 +218,8 @@ async def test_auto_send_derives_tool_spans_via_tracer():
 
     assert result.final_text == ""
     assert fake_tracing.started_names == ["Bash"]
-    assert fake_tracing.ended_outputs == ["ok"]
+    # String tool output is wrapped in a dict (SGP spans require an object).
+    assert fake_tracing.ended_outputs == [{"output": "ok"}]
 
 
 # ---------------------------------------------------------------------------

tests/lib/core/harness/test_tracer.py

@@ -15,7 +15,32 @@ async def test_open_then_close_starts_and_ends_span():
     await tracer.handle(OpenSpan(key="call_1", kind="tool", name="Bash", input={"cmd": "ls"}))
     await tracer.handle(CloseSpan(key="call_1", output="files", is_complete=True))
     assert fake.started == [("Bash", "p1", {"cmd": "ls"})]
-    assert fake.ended == [("Bash", "files")]
+    # A plain-string output is wrapped in a dict (SGP spans require an object).
+    assert fake.ended == [("Bash", {"output": "files"})]
+
+
+@pytest.mark.asyncio
+async def test_non_dict_payloads_are_wrapped_in_a_dict():
+    """SGP spans reject scalar input/output with a 422; the tracer wraps any
+    non-dict payload so reasoning spans (string output) are not dropped."""
+    fake = FakeTracing()
+    tracer = SpanTracer(trace_id="t1", parent_span_id="p1", tracing=fake)
+    await tracer.handle(OpenSpan(key="reasoning:0", kind="reasoning", name="reasoning", input={}))
+    await tracer.handle(CloseSpan(key="reasoning:0", output="chain of thought", is_complete=True))
+    # Empty-dict input stays a dict; string output is wrapped.
+    assert fake.started == [("reasoning", "p1", {})]
+    assert fake.ended == [("reasoning", {"output": "chain of thought"})]
+
+
+@pytest.mark.asyncio
+async def test_dict_and_none_payloads_pass_through_unchanged():
+    fake = FakeTracing()
+    tracer = SpanTracer(trace_id="t1", parent_span_id="p1", tracing=fake)
+    await tracer.handle(OpenSpan(key="c", kind="tool", name="T", input={"a": 1}))
+    await tracer.handle(CloseSpan(key="c", output={"result": "x"}, is_complete=True))
+    await tracer.handle(OpenSpan(key="d", kind="tool", name="U", input={}))
+    await tracer.handle(CloseSpan(key="d", output=None, is_complete=False))
+    assert fake.ended == [("T", {"result": "x"}), ("U", None)]
 
 
 @pytest.mark.asyncio

tests/lib/core/harness/test_yield_delivery.py

@@ -42,7 +42,8 @@ async def test_yield_passes_events_through_and_traces():
     out = [e async for e in yield_events(_gen(events), tracer=tracer)]
     assert out == events  # passthrough unchanged
     assert fake.started_names == ["Bash"]  # span derived + opened
-    assert fake.ended_outputs == ["ok"]  # span closed with response
+    # String tool output is wrapped in a dict (SGP spans require an object).
+    assert fake.ended_outputs == [{"output": "ok"}]  # span closed with response
 
 
 @pytest.mark.asyncio