symfonic.services.observability.suite

suite

Composition: the one place a config decides what observability it buys.

Before this task the decision was spread across the engine — the OTel bridge was appended to one list, the metrics collector registered on another, adopter callbacks merged into a third, each guarded by its own "is it on?" test. Three enablement tests are three chances to disagree, and the failure mode is silent: telemetry that is configured but wired nowhere looks exactly like telemetry that had nothing to report.

observers_from_config answers the question once, and answers it with an empty tuple when nothing is configured — which :func:compose_event_sink then turns into no event sink at all.

Both halves of that recipe live here rather than beside the bridge: choosing observers and wrapping them in a sink are the same decision, and :mod:.bridge is the projection of one run's events, not the place that decides whether a run has observability at all.

The config reaches both halves. The recipe is compose_event_sink(scope, observers_from_config(config, ...), config=config), and the second config is load-bearing rather than redundant: choosing observers and choosing what those observers are told are two decisions of the same configuration. While only observers_from_config read it, FrameworkConfig.observability_redact_payloads was a declared, documented control that the supported wiring path never consulted — prompts, completions and tool arguments reached every observer with the flag set, and only a caller who separately hand-plumbed capture=PromptCapturePolicy.from_config(config) got redaction. That is the same fail-open :mod:.capture argues against, reached through composition instead of through getattr.

compose_event_sink

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

Build the run's event sink, or None when nobody is watching.

Returning None is the zero-overhead guarantee expressed in a type: ServiceBindings.event_sink left unbound means InvocationRunner never constructs a CallbackEventAdapter, so a run with no observers pays for no buffer, no worker and no per-event delivery.

config is what makes the adopter's redaction opt-in reachable from the supported wiring path: pass the same object given to :func:observers_from_config and FrameworkConfig.observability_redact_payloads decides what the observers are handed. capture remains the explicit override for a caller who wants a policy the config cannot express (the three gates are independent; the flag maps onto all three), and omitting both keeps the shipped behaviour of delivering every payload in-process.

The sink it returns is not self-closing: the kernel's terminal releases it, and a run that never reaches one is released by ObservabilityBridge.aclose. Composing a sink does not make this module responsible for running it.

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

def compose_event_sink(
    scope: RunScope,
    observers: Iterable[Any],
    *,
    config: Any | None = None,
    capture: PromptCapturePolicy | None = None,
    accountant: CostAccountant | None = None,
) -> ObservabilityBridge | None:
    """Build the run's event sink, or ``None`` when nobody is watching.

    Returning ``None`` is the zero-overhead guarantee expressed in a type:
    ``ServiceBindings.event_sink`` left unbound means ``InvocationRunner``
    never constructs a ``CallbackEventAdapter``, so a run with no observers
    pays for no buffer, no worker and no per-event delivery.

    ``config`` is what makes the adopter's redaction opt-in reachable from the
    supported wiring path: pass the same object given to
    :func:`observers_from_config` and
    ``FrameworkConfig.observability_redact_payloads`` decides what the
    observers are handed. ``capture`` remains the explicit override for a
    caller who wants a policy the config cannot express (the three gates are
    independent; the flag maps onto all three), and omitting both keeps the
    shipped behaviour of delivering every payload in-process.

    The sink it returns is *not* self-closing: the kernel's terminal releases
    it, and a run that never reaches one is released by
    ``ObservabilityBridge.aclose``. Composing a sink does not make this module
    responsible for running it.
    """
    present = tuple(observer for observer in observers if observer is not None)
    if not present:
        return None
    if capture is None:
        capture = (
            CAPTURE_ALL if config is None else PromptCapturePolicy.from_config(config)
        )
    return ObservabilityBridge(
        scope, present, capture=capture, accountant=accountant
    )

observers_from_config

observers_from_config(config: Any, *, collector: Any | None = None, callback_handlers: Sequence[Any] | None = None, otel_builder: Any | None = None) -> tuple[Any, ...]

Return every observer this configuration asks for, in dispatch order.

Adopter handlers are wrapped once, not once per handler: the shipped CallbackManager already owns fan-out, ordering and per-handler error isolation, and a second layer of the same thing would only add a second place for that ordering to be defined.

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

def observers_from_config(
    config: Any,
    *,
    collector: Any | None = None,
    callback_handlers: Sequence[Any] | None = None,
    otel_builder: Any | None = None,
) -> tuple[Any, ...]:
    """Return every observer this configuration asks for, in dispatch order.

    Adopter handlers are wrapped **once**, not once per handler: the shipped
    ``CallbackManager`` already owns fan-out, ordering and per-handler error
    isolation, and a second layer of the same thing would only add a second
    place for that ordering to be defined.
    """
    observers: list[Any] = []
    otel = build_otel_observer(config, builder=otel_builder)
    if otel is not None:
        observers.append(otel)
    if collector is not None:
        observers.append(MetricsObserver(collector))
    handlers = tuple(callback_handlers or ())
    if handlers:
        observers.append(CallbackHandlerObserver(CallbackManager(handlers)))
    return tuple(observers)