fix(tracing): never let span export block the event loop or the run path

cubepi/tracing/meter.py

@@ -16,6 +16,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import contextlib
 import time
 from dataclasses import dataclass, field
@@ -200,7 +201,12 @@ def detach() -> None:
         return detach
 
     async def force_flush(self, timeout_seconds: float = 30.0) -> bool:
-        return self._provider.force_flush(timeout_millis=int(timeout_seconds * 1000))
+        # Sync provider flush can hold for seconds on a slow OTLP
+        # collector; run it in a worker thread so awaiting this never
+        # stalls the event loop (mirrors Tracer.force_flush).
+        return await asyncio.to_thread(
+            self._provider.force_flush, timeout_millis=int(timeout_seconds * 1000)
+        )
 
     async def shutdown(self, timeout_seconds: float = 30.0) -> None:
         if self._shutdown:

cubepi/tracing/tracer.py

@@ -8,12 +8,13 @@
 
 from __future__ import annotations
 
+import asyncio
 import atexit
 import contextlib
 import logging
 import uuid
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, AsyncIterator, Callable
+from typing import TYPE_CHECKING, Any, AsyncIterator, Callable, Literal
 
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import TracerProvider
@@ -189,6 +190,10 @@ def __init__(
             schema_url=SCHEMA_URL,
         )
         self._shutdown = False
+        # Strong refs to background flush Tasks (``trace(...,
+        # flush="background")``): the loop only holds weak refs, so an
+        # untracked task could be GC'd mid-flush.
+        self._pending_flushes: set[asyncio.Task[Any]] = set()
         self._atexit_flush_timeout_ms = int(atexit_flush_timeout_seconds * 1000)
         self._atexit_unregister: Callable[[], None] | None = None
         if atexit_flush:
@@ -313,18 +318,40 @@ def detach():
 
         return detach
 
+    def _track_flush_task(self, task: "asyncio.Task[Any]") -> None:
+        """Hold a strong reference to a background flush Task until it ends.
+
+        The event loop keeps only weak references to Tasks, so a
+        fire-and-forget flush could be garbage-collected mid-export
+        without this. ``shutdown()`` awaits whatever is still pending.
+        """
+        self._pending_flushes.add(task)
+        task.add_done_callback(self._pending_flushes.discard)
+
     async def force_flush(self, timeout_seconds: float = 30.0) -> bool:
         """Block until all currently buffered spans are exported.
 
+        The provider's ``force_flush`` is synchronous and can hold for
+        seconds when an exporter is slow (e.g. OTLP HTTP to a remote or
+        backlogged collector), so it runs in a worker thread — awaiting
+        this never stalls the event loop.
+
         Returns ``False`` on timeout.
         """
         timeout_millis = int(timeout_seconds * 1000)
-        return self._provider.force_flush(timeout_millis=timeout_millis)
+        return await asyncio.to_thread(
+            self._provider.force_flush, timeout_millis=timeout_millis
+        )
 
     async def shutdown(self, timeout_seconds: float = 30.0) -> None:
         """Flush and close all exporters. Tracer is unusable after this."""
         if self._shutdown:
             return
+        # Settle in-flight background flushes first (trace(...,
+        # flush="background")) so their spans can't race the provider
+        # shutdown below; failures were already logged by the supervisor.
+        if self._pending_flushes:
+            await asyncio.gather(*list(self._pending_flushes), return_exceptions=True)
         # SpanProcessor.shutdown is sync; flush first to bound wait.
         await self.force_flush(timeout_seconds=timeout_seconds)
         self._provider.shutdown()
@@ -738,6 +765,8 @@ def _build_resource(
 async def trace(
     tracer: "Tracer | None",
     agent: "Agent",
+    *,
+    flush: Literal["await", "background"] = "await",
 ) -> AsyncIterator[None]:
     """Best-effort tracing scope for one agent run.
 
@@ -748,6 +777,19 @@ async def trace(
     block a no-op, which lets callers gate tracing on config without branching
     at the call site.
 
+    ``flush`` picks what exiting the block waits for:
+
+    - ``"await"`` (default) — block until this run's spans are exported.
+      The exit is only as fast as the slowest exporter; use it when the
+      caller has nothing latency-sensitive after the block.
+    - ``"background"`` — detach synchronously (listeners are off before
+      the next line) but let the export run as a supervised background
+      task. Use this on request/serving paths where a caller is waiting
+      on the block's completion: span export to a slow collector must
+      not gate a user-visible response. The tracer keeps a strong
+      reference to the task and ``await tracer.shutdown()`` settles any
+      still-pending flushes, so spans are not lost on clean shutdown.
+
     This does **not** shut the tracer down: the tracer is reusable across runs,
     so build it once (e.g. per process) and call ``await tracer.shutdown()``
     when the owning process stops.
@@ -760,7 +802,7 @@ async def trace(
     -------
     ::
 
-        async with trace(tracer, agent):
+        async with trace(tracer, agent, flush="background"):
             await agent.prompt("...")
     """
     if tracer is None:
@@ -780,10 +822,26 @@ async def trace(
         if detach is not None:
             try:
                 result = detach()
-                # In an async context ``detach()`` returns a flush Task; await
-                # it so this run's spans are exported before the block exits.
+                # In an async context ``detach()`` returns a flush Task.
                 if result is not None and hasattr(result, "__await__"):
-                    await result
+                    if flush == "background":
+                        # Supervise instead of awaiting: log failures
+                        # (matching this helper's swallow-and-log
+                        # contract) and keep a strong ref via the
+                        # tracer so the task can't be GC'd mid-export.
+                        async def _supervise(flush_task: Any) -> None:
+                            try:
+                                await flush_task
+                            except Exception as exc:  # noqa: BLE001
+                                _log_tracing_warning("background flush failed", exc)
+
+                        tracer._track_flush_task(
+                            asyncio.get_running_loop().create_task(_supervise(result))
+                        )
+                    else:
+                        # Await so this run's spans are exported before
+                        # the block exits.
+                        await result
             except Exception as exc:  # noqa: BLE001 — flush/detach must never break the run
                 _log_tracing_warning("detach/flush failed", exc)
 

tests/tracing/test_nonblocking_flush.py

@@ -0,0 +1,221 @@
+"""Non-blocking span flush: ``force_flush`` off the loop, background trace mode.
+
+The provider's ``force_flush`` is synchronous and blocks its calling
+thread until every processor drains its queue through
+``exporter.export()`` — seconds when an OTLP collector is remote or
+backlogged. Two invariants pinned here:
+
+1. ``Tracer.force_flush`` runs that sync flush in a worker thread, so
+   awaiting it never stalls the event loop (a stalled loop freezes
+   every concurrent request in the host process).
+2. ``trace(tracer, agent, flush="background")`` exits the block without
+   waiting for export, while ``await tracer.shutdown()`` still settles
+   any in-flight background flush so spans are not lost on clean
+   shutdown.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import threading
+import time
+
+from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
+
+from cubepi.agent.agent import Agent
+from cubepi.providers.base import Model
+from cubepi.providers.faux import FauxProvider
+from cubepi.tracing import Tracer, trace
+
+MODEL = Model(id="faux-1", provider_id="faux")
+
+
+class _SlowExporter(SpanExporter):
+    """Sleeps in ``export`` to simulate a slow/backlogged collector."""
+
+    def __init__(self, export_seconds: float = 0.0) -> None:
+        self.export_seconds = export_seconds
+        self.export_count = 0
+
+    def export(self, spans):  # noqa: ANN001
+        self.export_count += 1
+        if self.export_seconds:
+            time.sleep(self.export_seconds)
+        return SpanExportResult.SUCCESS
+
+    def shutdown(self) -> None:
+        pass
+
+    def force_flush(self, timeout_millis: int = 30_000) -> bool:
+        return True
+
+
+def _build(exporter: SpanExporter) -> tuple[Agent, Tracer]:
+    provider = FauxProvider(provider_id="faux")
+    agent = Agent(model=provider.model(MODEL.id), system_prompt="t")
+    tracer = Tracer(
+        service_name="t",
+        agent_name="t",
+        exporters=[exporter],
+        atexit_flush=False,
+    )
+    return agent, tracer
+
+
+def _queue_span(tracer: Tracer) -> None:
+    """Put one ended span into the BatchSpanProcessor queue so
+    ``force_flush`` actually has something to export."""
+    tracer._otel_tracer.start_span("flush-probe").end()
+
+
+async def test_force_flush_calls_provider_off_the_loop_thread():
+    exporter = _SlowExporter()
+    _agent, tracer = _build(exporter)
+    loop_thread = threading.current_thread().name
+    flush_threads: list[str] = []
+    inner = tracer._provider.force_flush
+
+    def _spy(timeout_millis: int = 30_000) -> bool:
+        flush_threads.append(threading.current_thread().name)
+        return inner(timeout_millis=timeout_millis)
+
+    tracer._provider.force_flush = _spy  # type: ignore[method-assign]
+    try:
+        assert await tracer.force_flush() is True
+        assert flush_threads, "provider.force_flush must have run"
+        assert flush_threads[0] != loop_thread, (
+            "sync provider flush must run in a worker thread, not on the loop"
+        )
+    finally:
+        tracer._provider.force_flush = inner  # type: ignore[method-assign]
+        await tracer.shutdown()
+
+
+async def test_force_flush_does_not_stall_concurrent_tasks():
+    # With a span queued against a 0.5s exporter, force_flush blocks its
+    # calling thread for >=0.5s. On the loop thread that would freeze the
+    # ticker below; off-loop it keeps ticking. Generous threshold to
+    # avoid CI flakes.
+    exporter = _SlowExporter(export_seconds=0.5)
+    _agent, tracer = _build(exporter)
+    _queue_span(tracer)
+    ticks = 0
+
+    async def _ticker() -> None:
+        nonlocal ticks
+        while True:
+            await asyncio.sleep(0.01)
+            ticks += 1
+
+    ticker = asyncio.create_task(_ticker())
+    try:
+        await tracer.force_flush()
+        assert exporter.export_count >= 1, "flush must have exported the queued span"
+        assert ticks >= 10, (
+            f"event loop stalled during flush (ticks={ticks}); "
+            "provider.force_flush must not run on the loop thread"
+        )
+    finally:
+        ticker.cancel()
+        await tracer.shutdown()
+
+
+async def test_meter_force_flush_calls_provider_off_the_loop_thread():
+    # Meter mirrors Tracer: its provider force_flush is synchronous and
+    # must run in a worker thread, never on the loop.
+    from opentelemetry.sdk.metrics import MeterProvider
+    from opentelemetry.sdk.metrics.export import InMemoryMetricReader
+    from opentelemetry.sdk.resources import Resource
+
+    from cubepi.tracing import Meter
+    from cubepi.tracing.schema import SCHEMA_URL
+
+    reader = InMemoryMetricReader()
+    resource = Resource.create({"service.name": "test"}, schema_url=SCHEMA_URL)
+    provider = MeterProvider(resource=resource, metric_readers=[reader])
+    meter = Meter.__new__(Meter)
+    meter._provider = provider  # type: ignore[attr-defined]
+    meter._shutdown = False  # type: ignore[attr-defined]
+
+    loop_thread = threading.current_thread().name
+    flush_threads: list[str] = []
+    inner = provider.force_flush
+
+    def _spy(timeout_millis: int = 30_000) -> bool:
+        flush_threads.append(threading.current_thread().name)
+        return inner(timeout_millis=timeout_millis)
+
+    provider.force_flush = _spy  # type: ignore[method-assign]
+    try:
+        assert await meter.force_flush() is True
+        assert flush_threads, "provider.force_flush must have run"
+        assert flush_threads[0] != loop_thread, (
+            "sync metric flush must run in a worker thread, not on the loop"
+        )
+    finally:
+        provider.force_flush = inner  # type: ignore[method-assign]
+        provider.shutdown()
+
+
+async def test_trace_background_exits_before_export_completes():
+    exporter = _SlowExporter(export_seconds=0.5)
+    agent, tracer = _build(exporter)
+    try:
+        t0 = time.monotonic()
+        async with trace(tracer, agent, flush="background"):
+            _queue_span(tracer)
+        exited_after = time.monotonic() - t0
+        # Give the background task a tick to start without waiting for it.
+        await asyncio.sleep(0)
+        assert exited_after < 0.3, (
+            f"background mode must not gate block exit on export "
+            f"(waited {exited_after:.2f}s against a 0.5s exporter)"
+        )
+        # Detach is still synchronous: listeners are gone at exit.
+        assert agent._listeners == []
+        # The flush is tracked, not dropped.
+        assert tracer._pending_flushes, "background flush task must be tracked"
+    finally:
+        await tracer.shutdown()
+    assert exporter.export_count >= 1, "shutdown must settle the background flush"
+    assert not tracer._pending_flushes
+
+
+async def test_trace_background_flush_failure_is_swallowed_and_logged(caplog):
+    exporter = _SlowExporter()
+    agent, tracer = _build(exporter)
+
+    async def _boom_flush(*_args, **_kwargs):  # noqa: ANN202
+        raise RuntimeError("flush boom")
+
+    tracer.force_flush = _boom_flush  # type: ignore[method-assign]
+    try:
+        async with trace(tracer, agent, flush="background"):
+            pass
+        # Let the supervisor observe the failure.
+        for _ in range(10):
+            if not tracer._pending_flushes:
+                break
+            await asyncio.sleep(0.01)
+        assert not tracer._pending_flushes
+        assert any(
+            "background flush failed" in r.getMessage() for r in caplog.records
+        ), "background flush failures must be logged, not silently dropped"
+    finally:
+        del tracer.force_flush  # restore the real method for shutdown
+        await tracer.shutdown()
+
+
+async def test_trace_default_still_awaits_flush():
+    # Regression pin: default mode blocks until export is done, so callers
+    # relying on "block exit == spans persisted" keep that guarantee.
+    exporter = _SlowExporter(export_seconds=0.2)
+    agent, tracer = _build(exporter)
+    try:
+        t0 = time.monotonic()
+        async with trace(tracer, agent):
+            _queue_span(tracer)
+        assert time.monotonic() - t0 >= 0.2
+        assert exporter.export_count >= 1
+    finally:
+        await tracer.shutdown()