fix(tracing): drain sync tracing processors on ACP shutdown

A sync ACP agent silently lost whatever business spans were still queued when the
pod stopped. The lifespan drained `shutdown_default_span_queue`, which is the ASYNC
path only; the sync tracing processors hold their own queue and nothing in the SDK
ever shut them down. `get_sync_tracing_processors()` had exactly one caller —
tracer.py, to CONSTRUCT a Trace — and no shutdown path at all.

This is the same class of bug as the missing `sgp_obs.shutdown()` in the previous
commit, and it compounds it from the other end. The business span is what an obs
span's `agentex.business_trace_id` resolves to, so dropping it breaks the pivot from
Tempo back to the SGP store — the backward edge points at a record that was never
written.

Found while working out what the obs-test-* agents in agentex-agents#2183 would still
need after the SDK absorbs their bootstrap: each one carries a `sgp_flush_lifespan`
that does exactly this, which is the tell that the SDK should have been doing it.

Each processor is isolated — one that hangs or raises must not strand the spans held
by the ones after it, and none of them may stop the pod shutting down.

The last test pins the wiring rather than just the helper: a drain nothing calls is
worthless, so it asserts the lifespan actually invokes both this and shutdown_sgp_obs.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: stephen-wang24

src/agentex/lib/sdk/fastacp/base/base_acp_server.py

@@ -119,6 +119,40 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             _detach_otel_context(otel_token)
 
 
+def _shutdown_sync_tracing_processors() -> None:
+    """Drain the sync tracing processors' queues at shutdown. Never raises.
+
+    ``shutdown_default_span_queue`` covers the async path only. The sync processors
+    keep their own queue and nothing in the SDK ever shut them down, so a sync ACP
+    agent dropped whatever business spans were still queued when the pod stopped.
+    That matters beyond the lost spans: the business span is what an obs span's
+    ``agentex.business_trace_id`` resolves to, so losing it breaks the pivot from
+    Tempo back to the SGP store.
+
+    Each processor is isolated: one that hangs or raises must not stop the others,
+    and none of them may stop the pod from shutting down.
+    """
+    try:
+        from agentex.lib.core.tracing.tracing_processor_manager import (
+            get_sync_tracing_processors,
+        )
+
+        processors = get_sync_tracing_processors()
+    except Exception:  # pragma: no cover - nothing to drain if this can't import
+        logger.debug("sync tracing processors unavailable at shutdown", exc_info=True)
+        return
+
+    for processor in processors:
+        try:
+            processor.shutdown()
+        except Exception:  # noqa: PERF203 - one bad processor must not block the rest
+            logger.warning(
+                "a sync tracing processor failed to flush on shutdown; "
+                "some business spans may be lost",
+                exc_info=True,
+            )
+
+
 class BaseACPServer(FastAPI):
     """
     AsyncAgentACP provides RPC-style hooks for agent events and commands asynchronously.
@@ -191,6 +225,11 @@ async def lifespan_context(app: FastAPI):  # noqa: ARG001
                 yield
             finally:
                 await shutdown_default_span_queue()
+                # The queue above is the ASYNC path only. Sync tracing processors
+                # hold their own queue and nothing ever drained it, so a sync ACP
+                # agent lost whatever business spans were still queued when the pod
+                # stopped — including the ones the obs correlation points at.
+                _shutdown_sync_tracing_processors()
                 # Flush whatever sgp-obs still holds. A periodic exporter's buffer
                 # is otherwise dropped when the pod stops, which for a short-lived
                 # or scaled-to-zero agent can be most of what it recorded. No-op
@@ -199,6 +238,7 @@ async def lifespan_context(app: FastAPI):  # noqa: ARG001
 
         return lifespan_context
 
+
     async def _healthz(self):
         """Health check endpoint"""
         result = {"status": "healthy"}

src/agentex/lib/sdk/fastacp/base/tests/__init__.py

@@ -0,0 +1,73 @@
+"""Tests for the ACP lifespan's shutdown drains.
+
+``shutdown_default_span_queue`` covers the async span path. The SYNC tracing
+processors keep their own queue, and nothing in the SDK ever shut them down, so a
+sync ACP agent dropped whatever business spans were still queued when the pod
+stopped. That is worse than the spans themselves: the business span is what an obs
+span's ``agentex.business_trace_id`` resolves to, so losing it breaks the pivot from
+Tempo back to the SGP store.
+"""
+
+from __future__ import annotations
+
+from agentex.lib.sdk.fastacp.base import base_acp_server
+from agentex.lib.sdk.fastacp.base.base_acp_server import _shutdown_sync_tracing_processors
+
+
+class _Processor:
+    def __init__(self, explode: bool = False) -> None:
+        self.calls = 0
+        self._explode = explode
+
+    def shutdown(self) -> None:
+        self.calls += 1
+        if self._explode:
+            raise RuntimeError("flush timed out")
+
+
+def _patch_processors(monkeypatch, processors):
+    import agentex.lib.core.tracing.tracing_processor_manager as mgr
+
+    monkeypatch.setattr(mgr, "get_sync_tracing_processors", lambda: processors)
+
+
+class TestSyncProcessorDrain:
+    def test_every_processor_is_flushed(self, monkeypatch):
+        a, b = _Processor(), _Processor()
+        _patch_processors(monkeypatch, [a, b])
+        _shutdown_sync_tracing_processors()
+        assert (a.calls, b.calls) == (1, 1)
+
+    def test_one_failure_does_not_stop_the_others(self, monkeypatch):
+        """A processor that hangs or raises must not strand the spans held by the
+        ones after it in the list."""
+        bad, good = _Processor(explode=True), _Processor()
+        _patch_processors(monkeypatch, [bad, good])
+        _shutdown_sync_tracing_processors()
+        assert good.calls == 1
+
+    def test_no_processors_is_a_no_op(self, monkeypatch):
+        _patch_processors(monkeypatch, [])
+        _shutdown_sync_tracing_processors()  # must not raise
+
+    def test_an_unimportable_manager_does_not_fail_shutdown(self, monkeypatch):
+        """Nothing here may stop the pod from shutting down."""
+        import builtins
+
+        real_import = builtins.__import__
+
+        def blocked(name, *args, **kwargs):
+            if "tracing_processor_manager" in name:
+                raise ImportError("boom")
+            return real_import(name, *args, **kwargs)
+
+        monkeypatch.setattr(builtins, "__import__", blocked)
+        _shutdown_sync_tracing_processors()  # must not raise
+
+    def test_the_lifespan_calls_it(self):
+        """Pin the wiring, not just the helper: a drain nothing calls is worthless."""
+        import inspect
+
+        source = inspect.getsource(base_acp_server.BaseACPServer.get_lifespan_function)
+        assert "_shutdown_sync_tracing_processors()" in source
+        assert "shutdown_sgp_obs()" in source