symfonic.services.observability.bridge

bridge

The observability event bridge — one consumer of the kernel event stream.

T2.3.3 established that blocking results, text deltas, typed events and callback fan-out are all projections of one internal stream owned by one RequestContext. Observability was the consumer still outside that rule: metrics read LLMEndEvent, the OTel bridge read six event types, and both were fed by emission sites hand-placed in the engine.

This bridge is a callable suitable for ServiceBindings.event_sink. It reads KernelEvent and nothing else, and projects it onto the segregated ports in :mod:symfonic.services.observability.ports.

Three properties it must never lose:

• Telemetry failure is not run failure. Every observer call is isolated, and so is the pricing call that builds one observer's argument; a sink or a registry that is down increments a counter and the run continues.
• Exactly one terminal. The kernel guarantees one terminal event per run (EVT-1); the bridge enforces the same on its own output, so a stream that somehow repeated one cannot double-bill a cost observer.
• Nothing costs anything when nobody is watching. No observers means compose_event_sink (in :mod:.suite, which owns composition) returns None and the kernel never constructs a callback adapter at all.

ObservabilityBridge

ObservabilityBridge(scope: RunScope, observers: Iterable[Any], *, capture: PromptCapturePolicy = CAPTURE_ALL, accountant: CostAccountant | None = None)

Projects one run's kernel events onto whichever ports are implemented.

Source code in src/symfonic/services/observability/bridge.py

def __init__(
    self,
    scope: RunScope,
    observers: Iterable[Any],
    *,
    capture: PromptCapturePolicy = CAPTURE_ALL,
    accountant: CostAccountant | None = None,
) -> None:
    self._scope = scope
    self._capture = capture
    self._accountant = accountant or CostAccountant()
    self._observers = tuple(observers)
    self._subscribers = resolve_ports(self._observers)
    self._opened = False
    self._closed = False
    self._tokens: tuple[object, object] | None = None
    self.failures = 0

aclose async

aclose() -> None

Release what a run that never terminated left held. Never raises.

Idempotent, and safe after a normal terminal: _close has already cleared _tokens by then. Whoever owns the sink owns this call — :func:compose_event_sink builds the object, it does not run it. Why it exists at all, and why closing is duck-typed rather than a seventh port, is in :mod:symfonic.services.observability.closing.

The release is scoped to this run: a bridge is per-run, an observer need not be, and releasing a shared OTelObserver here would end another live run's root span and reset its carrier under it.

The abandoned end is owed only by a bridge that OPENED (TA8.29): one composed but never fed an event announced no run, so finishing it would be a start-less end.

Source code in src/symfonic/services/observability/bridge.py

async def aclose(self) -> None:
    """Release what a run that never terminated left held. Never raises.

    Idempotent, and safe after a normal terminal: ``_close`` has already
    cleared ``_tokens`` by then. Whoever owns the sink owns this call —
    :func:`compose_event_sink` builds the object, it does not run it. Why
    it exists at all, and why closing is duck-typed rather than a seventh
    port, is in :mod:`symfonic.services.observability.closing`.

    The release is scoped to *this run*: a bridge is per-run, an observer
    need not be, and releasing a shared ``OTelObserver`` here would end
    another live run's root span and reset its carrier under it.

    The abandoned end is owed only by a bridge that OPENED (TA8.29): one composed but
    never fed an event announced no run, so finishing it would be a start-less end.
    """
    abandoned = self._opened and not self._closed
    self._closed = True
    if abandoned:
        await self._emit("run", "on_run_finished", abandoned_finish(self._scope))
    if self._tokens is not None:
        unbind_run(self._tokens)
        self._tokens = None
    self.failures += await close_observers(self._observers, run_id=self._scope.run_id)