feat(tracing): auto-apply sgp-obs @business_trace to ACP handlers

Every agent's message/send and task/create handler is now wrapped with sgp-obs
`business_trace(business_trace_id=lambda p: p.task.id)` at registration, so an
agent gets the full business<->obs correlation edge — the turn anchor plus the
process-local correlation context the openai-agents bridge and auto-egress read
for their reverse tag — with NO per-agent code. Proven manually on analyst-agent
(trace 9ad202d0b0bbdc0a78d8e5719c830811); this bakes it into the base server so
no agent has to decorate.

The decorator preserves the handler's shape (sync / async / async-generator), so
streaming message/send handlers keep streaming.

Fail-open + version-tolerant: `_with_business_trace` no-ops when the installed
sgp-obs predates `business_trace` (defensive import, same stance as init_tracing),
so this is inert until the sgp-obs pin is bumped to the release that ships it
(0.5.0) — no code change needed here at that point.

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

Author: NiteshDhanpal

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

@@ -360,9 +360,33 @@ async def _process_request(
     - on_message_send
     """
 
+    @staticmethod
+    def _with_business_trace(fn: Callable[..., Any], id_getter: Callable[[Any], Any]) -> Callable[..., Any]:
+        """Auto-apply sgp-obs ``@business_trace`` so every agent handler gets the
+        business<->obs correlation edge with NO per-agent code: it opens the turn
+        anchor and binds the process-local correlation context that the
+        openai-agents bridge and auto-egress read for their reverse tag. The
+        decorator preserves the handler's shape (sync / async / async-generator),
+        so streaming ``message/send`` handlers keep streaming.
+
+        Fail-open + version-tolerant: a no-op (returns ``fn`` unchanged) when the
+        installed sgp-obs predates ``business_trace`` — the same defensive stance
+        as the ``init_tracing`` wiring above, so bumping sgp-obs turns it on with
+        no code change here.
+        """
+        try:
+            from sgp_obs.traces import business_trace
+        except Exception:
+            return fn
+        try:
+            return business_trace(business_trace_id=id_getter)(fn)
+        except Exception:  # pragma: no cover - correlation must never break registration
+            return fn
+
     # Type: Async
     def on_task_create(self, fn: Callable[[CreateTaskParams], Awaitable[Any]]):
         """Handle task/init method"""
+        fn = self._with_business_trace(fn, lambda p: p.task.id)
         wrapped = self._wrap_handler(fn)
         self._handlers[RPCMethod.TASK_CREATE] = wrapped
         return fn
@@ -417,6 +441,9 @@ def on_message_send(
         For non-streaming: return a single TaskMessage
         For streaming: return an AsyncGenerator that yields TaskMessageUpdate objects
         """
+        # One annotation, applied for the agent: correlate this turn end-to-end.
+        # Preserves the handler's async-generator shape so streaming still streams.
+        fn = self._with_business_trace(fn, lambda p: p.task.id)
 
         async def message_send_wrapper(params: SendMessageParams):
             """Special wrapper for message_send that handles both regular async functions and async generators"""

tests/test_business_trace_autowrap.py

@@ -0,0 +1,63 @@
+"""BaseACPServer auto-applies sgp-obs @business_trace to handlers, fail-open and
+version-tolerant (no-op when the installed sgp-obs predates business_trace)."""
+
+import sys
+import types
+
+from agentex.lib.sdk.fastacp.base.base_acp_server import BaseACPServer
+
+
+def test_noop_when_sgp_obs_lacks_business_trace(monkeypatch):
+    # sgp-obs too old (no business_trace) -> import fails -> handler returned unchanged
+    monkeypatch.setitem(sys.modules, "sgp_obs.traces", None)
+
+    def handler(params):
+        return "ok"
+
+    assert BaseACPServer._with_business_trace(handler, lambda p: p.task.id) is handler
+
+
+def test_applies_business_trace_and_passes_id_getter(monkeypatch):
+    calls: dict[str, object] = {}
+
+    fake = types.ModuleType("sgp_obs.traces")
+
+    def business_trace(*, business_trace_id):
+        calls["getter"] = business_trace_id
+
+        def deco(fn):
+            def wrapped(*a, **k):
+                return ("wrapped", fn(*a, **k))
+
+            return wrapped
+
+        return deco
+
+    fake.business_trace = business_trace  # type: ignore[attr-defined]
+    monkeypatch.setitem(sys.modules, "sgp_obs.traces", fake)
+
+    def handler(params):
+        return "ok"
+
+    wrapped = BaseACPServer._with_business_trace(handler, lambda p: p.task.id)
+    assert wrapped is not handler
+    assert wrapped(None) == ("wrapped", "ok")
+    # the id resolver threads through to business_trace
+    params = types.SimpleNamespace(task=types.SimpleNamespace(id="T-9"))
+    assert calls["getter"](params) == "T-9"
+
+
+def test_fail_open_when_decorator_raises(monkeypatch):
+    fake = types.ModuleType("sgp_obs.traces")
+
+    def business_trace(*, business_trace_id):
+        raise RuntimeError("boom")
+
+    fake.business_trace = business_trace  # type: ignore[attr-defined]
+    monkeypatch.setitem(sys.modules, "sgp_obs.traces", fake)
+
+    def handler(params):
+        return "ok"
+
+    # a decorator that blows up must not break handler registration
+    assert BaseACPServer._with_business_trace(handler, lambda p: p.task.id) is handler