`symfonic.services.observability`¶

observability ¶

Segregated observability ports over the one kernel event stream (T4.2.3).

Callbacks, metrics, cost, traces and OpenTelemetry used to be five consumers with five sources: hand-placed emission sites in the engine, a boolean capture flag read at four of them, two cost calculations, and a trace context that only existed when an optional extra was installed.

This package gives all five one source — KernelEvent — and one seam each:

:mod:.ports — six narrow protocols; an observer implements what it cares about and hears nothing else.
:mod:.bridge — the single consumer of the event stream, and the only place fan-out, error isolation and terminal cardinality are decided.
:mod:.suite — composition: what observability a config buys, and the sink that wraps it (None when nobody is watching).
:mod:.closing — releasing what a run that never terminated left held.
:mod:.capture — the one answer to "may I record this payload?" (deny by default).
:mod:.trace — W3C trace propagation with no third-party dependency.
:mod:.cost — one accountant, delegating to the shipped pricing registry.
:mod:.callbacks, :mod:.metrics, :mod:.otel — adapters that keep the shipped public schemas working from the new source.

Importing this package pulls zero opentelemetry modules.

CallbackHandlerObserver ¶

CallbackHandlerObserver(manager: Any)

Renders observations into the shipped callback schemas.

Dispatch goes through the supplied CallbackManager, so per-handler error isolation, partial-handler tolerance and the has_hook fast path stay exactly as they are today — this observer adds a source, not a second dispatcher.

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

def __init__(self, manager: Any) -> None:
    self._manager = manager

CostAccountant ¶

Prices one run's usage, and says whether the registry actually knew.

assess ¶

assess(scope: RunScope, usage: UsageDelta, *, text: str | None = None, duration_ms: float = 0.0) -> CostObservation

Price usage for scope's model.

A zero-usage run is still priced rather than skipped: "this run cost nothing" and "nobody looked" are different facts, and only the first one is worth reporting.

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

def assess(
    self,
    scope: RunScope,
    usage: UsageDelta,
    *,
    text: str | None = None,
    duration_ms: float = 0.0,
) -> CostObservation:
    """Price ``usage`` for ``scope``'s model.

    A zero-usage run is still priced rather than skipped: "this run cost
    nothing" and "nobody looked" are different facts, and only the first
    one is worth reporting.
    """
    typed = TokenUsage.from_dict(usage_dict(usage), scope.model)
    return CostObservation(
        scope=scope,
        model=scope.model,
        usage=usage,
        cost_usd=typed.cost_usd,
        pricing_unknown=typed.pricing_unknown,
        breakdown=MappingProxyType(
            {
                "input_tokens": typed.input_tokens,
                "output_tokens": typed.output_tokens,
                "cached_tokens": typed.cached_tokens,
                "cache_creation_tokens": typed.cache_creation_tokens,
                "reasoning_tokens": typed.reasoning_tokens,
            }
        ),
        text=text,
        duration_ms=duration_ms,
    )

CostObservation `dataclass` ¶

CostObservation(scope: RunScope, model: str, usage: UsageDelta, cost_usd: float, pricing_unknown: bool, breakdown: Mapping[str, int] = (lambda: MappingProxyType({}))(), text: str | None = None, duration_ms: float = 0.0)

What the run cost, and whether the registry could actually say.

pricing_unknown is the discriminator that keeps "a $0 run" apart from "a run whose model the pricing registry did not recognise" — two states that look identical in a dashboard and mean opposite things on a bill.

CostObserver ¶

Bases: Protocol

What the run cost. Fed by the one accountant, never computed per observer.

DropObserver ¶

Bases: Protocol

Shed notices from a bounded adapter (BP-14).

ErrorObserver ¶

Bases: Protocol

The run's error terminal, already split into type and message.

EventsDropped `dataclass` ¶

EventsDropped(scope: RunScope, index: int, dropped_kind: str, dropped_count: int, first_dropped_index: int | None, last_dropped_index: int | None)

A shed notice: what the adapter could not deliver, and over what range.

MetricsObserver ¶

MetricsObserver(collector: Any)

Binds a run to its conversation row, then feeds the shipped collector.

It satisfies three ports — RunObserver, CostObserver and, since TA8.20, ErrorObserver — because those are the three the shipped callback rendering has anything to say about. Which ports an observer satisfies is resolved once at wiring time from the methods it defines (:func:~symfonic.services.observability.ports.resolve_ports), so a method absent here is a hook the collector never receives, with nothing raised and nothing logged.

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

def __init__(self, collector: Any) -> None:
    self._collector = collector
    self._delegate = CallbackHandlerObserver(CallbackManager([collector]))

on_run_failed `async` ¶

on_run_failed(observation: RunFailed) -> None

The error terminal, rendered as on_node_error like the rest.

Added by TA8.20, and it closes a gap rather than adding a feature. CallbackHandlerObserver renders four callback hooks and this observer forwarded three of them, so a collector wired through metrics_collector= was bound to the run and cost ports and to no error port at all: a failing run reached on_agent_start and on_agent_end and never on_node_error, while the legacy body delivered all three. The shipped ConversationMetricsCollector implements on_node_error as a no-op, so nothing in-tree changes — but an adopter's collector that counts failures counted none, and a silent hook is exactly the failure the observability envelope guard existed to prevent.

Failure is not a terminal replacement: the bridge emits this and then on_run_finished, which is the order the legacy body used too (on_node_error before on_agent_end).

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

async def on_run_failed(self, observation: RunFailed) -> None:
    """The error terminal, rendered as ``on_node_error`` like the rest.

    Added by TA8.20, and it closes a gap rather than adding a feature.
    ``CallbackHandlerObserver`` renders four callback hooks and this
    observer forwarded three of them, so a collector wired through
    ``metrics_collector=`` was bound to the run and cost ports and to no
    error port at all: a failing run reached ``on_agent_start`` and
    ``on_agent_end`` and never ``on_node_error``, while the legacy body
    delivered all three. The shipped ``ConversationMetricsCollector``
    implements ``on_node_error`` as a no-op, so nothing in-tree changes —
    but an adopter's collector that counts failures counted none, and a
    silent hook is exactly the failure the observability envelope guard
    existed to prevent.

    Failure is not a terminal *replacement*: the bridge emits this and then
    ``on_run_finished``, which is the order the legacy body used too
    (``on_node_error`` before ``on_agent_end``).
    """
    await self._delegate.on_run_failed(observation)

OTelObserver ¶

OTelObserver(handles: Any)

Drives the shipped OTEL span lifecycle from the kernel event stream.

Safe to build per run, and shareable across runs that are driven from their own asyncio task. What one run holds — its root span and its carrier token — belongs to :class:RunSpanTable, which owns the resource lifetime so this class can own the event lifetime: which kernel observation opens or closes which span.

Sharing is bounded by contextvars, not by that table: span nesting comes from the OTEL current-span context stack that start_run_span attaches, and only per-task context copies keep two runs' stacks apart. Runs interleaved inside one task get correct span identities and durations but cross-parented trees. See the module docstring.

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

def __init__(self, handles: Any) -> None:
    self.handles = handles
    self._spans = RunSpanTable(handles)
    self._manager = CallbackManager([handles.callback_bridge])
    self._delegate = CallbackHandlerObserver(self._manager)

aclose `async` ¶

aclose() -> None

Release every run this observer still holds. Never raises.

Process shutdown, not run cleanup: only runs whose terminal never arrived are left here — the stream guarantees one terminal per run, but a killed process does not. A caller that owns a single run wants :meth:release_run; calling this instead ends the spans of every other run in flight. It deliberately does not shut the exporter down: the handles are process-scoped and shared (see :func:build_otel_observer), so releasing them is :func:shutdown_otel's job, not one observer's.

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

async def aclose(self) -> None:
    """Release *every* run this observer still holds. Never raises.

    Process shutdown, not run cleanup: only runs whose terminal never
    arrived are left here — the stream guarantees one terminal per run, but
    a killed process does not. A caller that owns a single run wants
    :meth:`release_run`; calling this instead ends the spans of every other
    run in flight. It deliberately does **not** shut the exporter down: the
    handles are process-scoped and shared (see :func:`build_otel_observer`),
    so releasing them is :func:`shutdown_otel`'s job, not one observer's.
    """
    self._spans.release_all()

on_cost `async` ¶

on_cost(observation: CostObservation) -> None

Open the LLM span, then let the delegate's LLMEndEvent close it.

CallbackBridge.on_llm_end is a close: it pops the span keyed on (run_id, node_name) and returns when the key is absent. Without the paired start there is no span to attach the usage to, and the cost of the run reaches no exporter.

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

async def on_cost(self, observation: CostObservation) -> None:
    """Open the LLM span, then let the delegate's ``LLMEndEvent`` close it.

    ``CallbackBridge.on_llm_end`` is a *close*: it pops the span keyed on
    ``(run_id, node_name)`` and returns when the key is absent. Without the
    paired start there is no span to attach the usage to, and the cost of
    the run reaches no exporter.
    """
    scope = observation.scope
    await self._manager.on_llm_start(
        LLMStartEvent(
            model=observation.model,
            messages=[],
            run_id=scope.run_id,
            system_prompt="",
            node_name=INVOCATION_NODE,
        )
    )
    await self._delegate.on_cost(observation)

on_run_failed `async` ¶

on_run_failed(observation: RunFailed) -> None

Open the node span the delegate's NodeErrorEvent will close.

Same asymmetry as the LLM span: on_node_error calls end_node, which is tolerant of a missing start and therefore silent about it.

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

async def on_run_failed(self, observation: RunFailed) -> None:
    """Open the node span the delegate's ``NodeErrorEvent`` will close.

    Same asymmetry as the LLM span: ``on_node_error`` calls ``end_node``,
    which is tolerant of a missing start and therefore silent about it.
    """
    await self._manager.on_node_start(
        NodeStartEvent(
            node_name=INVOCATION_NODE, run_id=observation.scope.run_id
        )
    )
    await self._delegate.on_run_failed(observation)

release_run `async` ¶

release_run(run_id: str) -> None

Release one run's span and carrier. Never raises.

The call a run-scoped owner makes when its run ends without a terminal — see ObservabilityBridge.aclose. Scoped rather than wholesale because this observer may be shared by any number of overlapping runs, and one cancelled run must not end the others' spans.

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

async def release_run(self, run_id: str) -> None:
    """Release one run's span and carrier. Never raises.

    The call a run-scoped owner makes when its run ends without a terminal
    — see ``ObservabilityBridge.aclose``. Scoped rather than wholesale
    because this observer may be shared by any number of overlapping runs,
    and one cancelled run must not end the others' spans.
    """
    self._spans.release(run_id)

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)

PromptCapturePolicy `dataclass` ¶

PromptCapturePolicy(capture_prompts: bool = False, capture_outputs: bool = False, capture_tool_arguments: bool = False)

Three independent gates over the three payload classes.

They are independent because the risks are: an adopter may want the model's output for quality review while never recording the user's prompt, and a single flag cannot express that. from_config still maps the shipped single flag onto all three, so the current configuration surface is unchanged — the split is available, not imposed.

argument_names `staticmethod` ¶

argument_names(arguments: Mapping[str, Any] | None) -> tuple[str, ...]

Sorted argument names, always available.

Names are structure; values are payload. Keeping the names when the values are denied is what lets an operator see that a tool was called with a customer_id without seeing which customer.

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

@staticmethod
def argument_names(arguments: Mapping[str, Any] | None) -> tuple[str, ...]:
    """Sorted argument *names*, always available.

    Names are structure; values are payload. Keeping the names when the
    values are denied is what lets an operator see that a tool was called
    with a ``customer_id`` without seeing which customer.
    """
    if not arguments:
        return ()
    return tuple(sorted(str(key) for key in arguments))

from_config `classmethod` ¶

from_config(config: Any) -> PromptCapturePolicy

The policy for in-process fan-out, which is not an egress boundary.

Deliberately does not read otel_capture_prompts: that flag gates span attributes and is applied by the shipped SpanRecorder at the exporter, where the data actually leaves the process. Reusing it here would broaden its meaning from "do not put prompts in span attributes" to "do not give any consumer the text", silently emptying callback payloads that shipped code has always populated.

FrameworkConfig.observability_redact_payloads is the explicit opt-in for adopters who want the bridge itself to withhold. It is read through getattr so a duck-typed config still works, but it is a declared field precisely because a getattr-only control on an extra="ignore" model would be silently dropped and fail open.

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

@classmethod
def from_config(cls, config: Any) -> PromptCapturePolicy:
    """The policy for *in-process fan-out*, which is not an egress boundary.

    Deliberately does **not** read ``otel_capture_prompts``: that flag
    gates span attributes and is applied by the shipped ``SpanRecorder`` at
    the exporter, where the data actually leaves the process. Reusing it
    here would broaden its meaning from "do not put prompts in span
    attributes" to "do not give any consumer the text", silently emptying
    callback payloads that shipped code has always populated.

    ``FrameworkConfig.observability_redact_payloads`` is the explicit
    opt-in for adopters who want the bridge itself to withhold. It is read
    through ``getattr`` so a duck-typed config still works, but it is a
    declared field precisely because a ``getattr``-only control on an
    ``extra="ignore"`` model would be silently dropped and fail open.
    """
    if bool(getattr(config, "observability_redact_payloads", False)):
        return DENY_ALL
    return CAPTURE_ALL

output_text ¶

output_text(text: str | None) -> str | None

Model-produced text, or None when withheld.

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

def output_text(self, text: str | None) -> str | None:
    """Model-produced text, or ``None`` when withheld."""
    return text if self.capture_outputs else None

prompt_text ¶

prompt_text(text: str | None) -> str | None

The user-supplied prompt, or None when withheld.

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

def prompt_text(self, text: str | None) -> str | None:
    """The user-supplied prompt, or ``None`` when withheld."""
    return text if self.capture_prompts else None

tool_arguments ¶

tool_arguments(arguments: Mapping[str, Any] | None) -> Mapping[str, Any] | None

Tool call arguments, or None when withheld.

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

def tool_arguments(
    self, arguments: Mapping[str, Any] | None
) -> Mapping[str, Any] | None:
    """Tool call arguments, or ``None`` when withheld."""
    return arguments if self.capture_tool_arguments else None

RunFailed `dataclass` ¶

RunFailed(scope: RunScope, index: int, error_type: str, error_message: str)

The run emitted its error terminal.

The kernel formats the error as "TypeName: message". Splitting it here once means every downstream schema — span status, NodeErrorEvent, a log line — gets the same two fields instead of re-parsing the string.

RunFinished `dataclass` ¶

RunFinished(scope: RunScope, index: int, reason: str, text: str | None, usage: UsageDelta, duration_ms: float, stop_reason: str | None)

The run reached exactly one terminal event (EVT-1).

RunObserver ¶

Bases: Protocol

Run boundaries: the smallest port that can build a root span.

RunScope `dataclass` ¶

RunScope(run_id: str, tenant_id: str | None, session_id: str | None, entry_point: str, model: str, provider_family: str = '', prompt: str = '', root_run_id: str = '', parent_run_id: str | None = None)

The identity every observation is attributed to.

Built once per invocation by whoever composes the sink, because the kernel event stream carries a run_id and nothing else: tenant, session, model and entry point are plan facts, and re-deriving them per event is how two observers end up disagreeing about which tenant paid for a run.

RunStarted `dataclass` ¶

RunStarted(scope: RunScope, query: str | None)

The run has produced its first event.

TextEmitted `dataclass` ¶

TextEmitted(scope: RunScope, index: int, kind: str, text: str | None, length: int)

One text-bearing event from the stream, in stream order.

TextObserver ¶

Bases: Protocol

Model text, in stream order. The highest-volume port, deliberately alone.

ToolCompleted `dataclass` ¶

ToolCompleted(scope: RunScope, index: int, outcome: Any = None)

A tool returned. The outcome is opaque, exactly as it is to the kernel.

ToolInvoked `dataclass` ¶

ToolInvoked(scope: RunScope, index: int, call_id: str, name: str, arguments: Mapping[str, Any] | None, argument_names: tuple[str, ...] = ())

The loop dispatched an allowlisted tool call.

ToolObserver ¶

Bases: Protocol

Tool dispatch and completion.

TraceMode ¶

Bases: StrEnum

How much diagnostic information may leave the process.

TracePolicy `dataclass` ¶

TracePolicy(mode: TraceMode = TraceMode.METADATA, sample_rate: float = 1.0, failed_turns_always: bool = True, content_ttl_hours: int = 24, max_artifact_bytes: int = 262144)

One deployment's tracing, sampling and content-retention contract.

from_config `classmethod` ¶

from_config(config: Any) -> TracePolicy

Build from modern trace_* fields, then legacy OTEL flags.

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

@classmethod
def from_config(cls, config: Any) -> TracePolicy:
    """Build from modern ``trace_*`` fields, then legacy OTEL flags."""
    raw_mode = getattr(config, "trace_mode", None)
    if raw_mode is None:
        raw_mode = getattr(config, "symfonic_trace_mode", None)
    if raw_mode is None:
        if not bool(getattr(config, "otel_enabled", False)):
            mode = TraceMode.OFF
        elif bool(getattr(config, "otel_capture_prompts", False)):
            mode = TraceMode.CONTENT
        else:
            mode = TraceMode.METADATA
    else:
        mode = TraceMode(str(raw_mode).lower())
    def configured(name: str, default: Any) -> Any:
        value = getattr(config, f"trace_{name}", None)
        if value is None:
            value = getattr(config, f"symfonic_trace_{name}", default)
        return value

    return cls(
        mode=mode,
        sample_rate=float(configured("sample_rate", 1.0)),
        failed_turns_always=bool(configured("failed_turns_always", True)),
        content_ttl_hours=int(configured("content_ttl_hours", 24)),
        max_artifact_bytes=int(configured("max_artifact_bytes", 262_144)),
    )

TraceScope `dataclass` ¶

TraceScope(trace_id: str, span_id: str, sampled: bool = True)

A W3C trace context, without a W3C library.

traceparent ¶

traceparent() -> str

Render the traceparent header value.

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

def traceparent(self) -> str:
    """Render the ``traceparent`` header value."""
    return f"{_VERSION}-{self.trace_id}-{self.span_id}-{'01' if self.sampled else '00'}"

bind_run ¶

bind_run(scope: RunScope, trace: TraceScope) -> tuple[object, object]

Bind both carriers; returns the tokens :func:unbind_run needs.

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

def bind_run(scope: RunScope, trace: TraceScope) -> tuple[object, object]:
    """Bind both carriers; returns the tokens :func:`unbind_run` needs."""
    return _run_scope.set(scope), _trace_scope.set(trace)

build_otel_observer ¶

build_otel_observer(config: Any, *, builder: Callable[[Any], Any] | None = None) -> OTelObserver | None

Return the OTEL observer, or None when OTEL is off or unavailable.

None is a first-class answer here, not a failure: compose_event_sink drops absent observers, so a disabled or un-installed OTEL simply removes itself from the wiring without any call site learning it exists.

Lifetime. The returned observer is cheap and may be built per run; the exporter behind it is not, and is therefore cached per configuration for the life of the process and released by :func:shutdown_otel. An explicit builder bypasses that cache entirely — the caller supplying the exporter owns its lifetime, which is what test doubles want.

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

def build_otel_observer(
    config: Any,
    *,
    builder: Callable[[Any], Any] | None = None,
) -> OTelObserver | None:
    """Return the OTEL observer, or ``None`` when OTEL is off or unavailable.

    ``None`` is a first-class answer here, not a failure: ``compose_event_sink``
    drops absent observers, so a disabled or un-installed OTEL simply removes
    itself from the wiring without any call site learning it exists.

    **Lifetime.** The returned observer is cheap and may be built per run; the
    exporter behind it is not, and is therefore cached per configuration for
    the life of the process and released by :func:`shutdown_otel`. An explicit
    ``builder`` bypasses that cache entirely — the caller supplying the
    exporter owns its lifetime, which is what test doubles want.
    """
    if not getattr(config, "otel_enabled", False):
        return None
    handles = builder(config) if builder is not None else resolve_handles(config)
    if handles is None:
        logger.warning(
            "otel_enabled is set but the exporter could not be built; "
            "continuing without OpenTelemetry. Install the extra: "
            "pip install symfonic-core[otel]",
        )
        return None
    return OTelObserver(handles)

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
    )

current_run_scope ¶

current_run_scope() -> RunScope | None

The run being observed in this context, or None outside a run.

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

def current_run_scope() -> RunScope | None:
    """The run being observed in this context, or ``None`` outside a run."""
    return _run_scope.get()

current_trace ¶

current_trace() -> TraceScope | None

The trace scope of the run being observed, or None.

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

def current_trace() -> TraceScope | None:
    """The trace scope of the run being observed, or ``None``."""
    return _trace_scope.get()

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)

parse_traceparent ¶

parse_traceparent(header: str) -> TraceScope | None

Parse a traceparent header, or return None.

Refusing is deliberate. A malformed header that is "best-effort repaired" produces a trace id that correlates with nothing, which is worse than no correlation at all because it looks like one.

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

def parse_traceparent(header: str) -> TraceScope | None:
    """Parse a ``traceparent`` header, or return ``None``.

    Refusing is deliberate. A malformed header that is "best-effort repaired"
    produces a trace id that correlates with nothing, which is worse than no
    correlation at all because it looks like one.
    """
    if not header:
        return None
    parts = header.split("-")
    if len(parts) != 4:
        return None
    version, trace_id, span_id, flags = parts
    if version != _VERSION:
        return None
    if len(trace_id) != 32 or len(span_id) != 16 or len(flags) != 2:
        return None
    if trace_id == _ZERO_TRACE or span_id == _ZERO_SPAN:
        return None
    try:
        int(trace_id, 16), int(span_id, 16), int(flags, 16)
    except ValueError:
        return None
    return TraceScope(
        trace_id=trace_id, span_id=span_id, sampled=bool(int(flags, 16) & 0x01)
    )

redact_trace_value ¶

redact_trace_value(value: Any, *, max_string_bytes: int = 32768) -> Any

Recursively redact credentials and bound strings before export.

The result is JSON-shaped. Unknown objects are represented by their type, never by repr: a repr is allowed to contain precisely the secret the redactor is meant to keep out of the collector.

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

def redact_trace_value(value: Any, *, max_string_bytes: int = 32_768) -> Any:
    """Recursively redact credentials and bound strings before export.

    The result is JSON-shaped. Unknown objects are represented by their type,
    never by ``repr``: a repr is allowed to contain precisely the secret the
    redactor is meant to keep out of the collector.
    """
    if isinstance(value, Mapping):
        return {
            str(key): (
                "[REDACTED]"
                if _SECRET_KEY.search(str(key))
                else redact_trace_value(item, max_string_bytes=max_string_bytes)
            )
            for key, item in value.items()
        }
    if isinstance(value, (list, tuple, set, frozenset)):
        return [
            redact_trace_value(item, max_string_bytes=max_string_bytes)
            for item in value
        ]
    if isinstance(value, bytes):
        return f"[bytes:{len(value)}]"
    if isinstance(value, str):
        clean = _PRIVATE_KEY.sub("[REDACTED PRIVATE KEY]", value)
        clean = _BEARER.sub("Bearer [REDACTED]", clean)
        encoded = clean.encode("utf-8")
        if len(encoded) <= max_string_bytes:
            return clean
        suffix = "...[TRUNCATED]"
        room = max_string_bytes - len(suffix.encode("utf-8"))
        return encoded[:room].decode("utf-8", errors="ignore") + suffix
    if value is None or isinstance(value, (bool, int, float)):
        return value
    return f"[{type(value).__name__}]"

shutdown_otel ¶

shutdown_otel() -> None

Flush and release every cached exporter. Safe to call more than once.

Registered with atexit the first time handles are cached, because a BatchSpanProcessor holds spans that only shutdown flushes: a process that exits without it exports nothing from its final batch. Exposed publicly so a host that tears an application down deterministically — a test, a worker that reconfigures — need not wait for interpreter exit.

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

def shutdown_otel() -> None:
    """Flush and release every cached exporter. Safe to call more than once.

    Registered with ``atexit`` the first time handles are cached, because a
    ``BatchSpanProcessor`` holds spans that only ``shutdown`` flushes: a process
    that exits without it exports nothing from its final batch. Exposed
    publicly so a host that tears an application down deterministically — a
    test, a worker that reconfigures — need not wait for interpreter exit.
    """
    with _LOCK:
        handles = tuple(_HANDLES.values())
        _HANDLES.clear()
    for handle in handles:
        shutdown = getattr(getattr(handle, "tracer", None), "shutdown", None)
        if shutdown is None:
            continue
        try:
            shutdown()
        except Exception:  # noqa: BLE001 - telemetry never breaks a shutdown
            logger.warning("OTel tracer shutdown raised; suppressing", exc_info=True)

trace_for_run ¶

trace_for_run(run_id: str) -> TraceScope

Derive a trace scope for run_id.

The trace id is derived from the run id rather than drawn at random so the same run always lands in the same trace: a log line that only recorded a run id can still be joined to its trace after the fact, which is exactly the case where an operator needs the join and no longer has the process.

The span id is random, because two observers of the same run are two spans, not one.

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

def trace_for_run(run_id: str) -> TraceScope:
    """Derive a trace scope for ``run_id``.

    The trace id is derived from the run id rather than drawn at random so the
    same run always lands in the same trace: a log line that only recorded a
    run id can still be joined to its trace after the fact, which is exactly
    the case where an operator needs the join and no longer has the process.

    The span id is random, because two observers of the same run are two spans,
    not one.
    """
    digest = hashlib.sha256(run_id.encode("utf-8", errors="ignore")).hexdigest()[:32]
    if digest == _ZERO_TRACE:  # pragma: no cover - unreachable for sha256
        digest = f"{_ZERO_TRACE[:-1]}1"
    return TraceScope(trace_id=digest, span_id=secrets.token_hex(8))

unbind_run ¶

unbind_run(tokens: tuple[object, object]) -> None

Release the carriers bound by :func:bind_run.

A token minted in another context cannot be reset, and that happens for real: a terminal event delivered from a drain worker runs in a different task than the one that opened the run. Falling back to an explicit None keeps the leak bounded to that context instead of raising inside teardown.

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

def unbind_run(tokens: tuple[object, object]) -> None:
    """Release the carriers bound by :func:`bind_run`.

    A token minted in another context cannot be reset, and that happens for
    real: a terminal event delivered from a drain worker runs in a different
    task than the one that opened the run. Falling back to an explicit ``None``
    keeps the leak bounded to that context instead of raising inside teardown.
    """
    for var, token in zip((_run_scope, _trace_scope), tokens, strict=True):
        try:
            var.reset(token)  # type: ignore[arg-type]
        except ValueError:
            var.set(None)  # type: ignore[arg-type]

usage_dict ¶

usage_dict(usage: UsageDelta) -> dict[str, Any]

Kernel usage in the dict shape every shipped callback already parses.

total_tokens is deliberately absent: the shipped TokenUsage derives totals itself, and handing it a third number invites the two to drift.

The cache and reasoning dimensions are carried through because the registry prices them: it subtracts cache reads and cache writes from the billable input and charges each at its own rate. Rendering only the two totals is not a simplification — it bills a cache-heavy run at roughly ten times what it cost.

cache_ttl rides along for the same reason one dimension over: it is the rate selector for the cache writes, and TokenUsage.from_dict reads it to pick the 1h write rate over the 5-minute default. Dropping it while keeping the write count reports a number that looks right and bills wrong.

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

def usage_dict(usage: UsageDelta) -> dict[str, Any]:
    """Kernel usage in the dict shape every shipped callback already parses.

    ``total_tokens`` is deliberately absent: the shipped ``TokenUsage`` derives
    totals itself, and handing it a third number invites the two to drift.

    The cache and reasoning dimensions are carried through because the registry
    prices them: it subtracts cache reads and cache writes from the billable
    input and charges each at its own rate. Rendering only the two totals is
    not a simplification — it bills a cache-heavy run at roughly ten times what
    it cost.

    ``cache_ttl`` rides along for the same reason one dimension over: it is the
    *rate selector* for the cache writes, and ``TokenUsage.from_dict`` reads it
    to pick the 1h write rate over the 5-minute default. Dropping it while
    keeping the write count reports a number that looks right and bills wrong.
    """
    rendered: dict[str, Any] = {
        "input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
    }
    for field_name, key in _DIMENSIONS:
        value = getattr(usage, field_name, 0)
        if value:
            rendered[key] = int(value)
    cache_ttl = getattr(usage, "cache_ttl", None)
    if cache_ttl:
        rendered["cache_ttl"] = cache_ttl
    return rendered

symfonic.services.observability¶

observability ¶

CallbackHandlerObserver ¶

CostAccountant ¶

assess ¶

CostObservation dataclass ¶

CostObserver ¶

DropObserver ¶

ErrorObserver ¶

EventsDropped dataclass ¶

MetricsObserver ¶

on_run_failed async ¶

OTelObserver ¶

aclose async ¶

on_cost async ¶

on_run_failed async ¶

release_run async ¶

ObservabilityBridge ¶

aclose async ¶

PromptCapturePolicy dataclass ¶

argument_names staticmethod ¶

from_config classmethod ¶

output_text ¶

prompt_text ¶

tool_arguments ¶

RunFailed dataclass ¶

RunFinished dataclass ¶

RunObserver ¶

RunScope dataclass ¶

RunStarted dataclass ¶

TextEmitted dataclass ¶

TextObserver ¶

ToolCompleted dataclass ¶

ToolInvoked dataclass ¶

ToolObserver ¶

TraceMode ¶

TracePolicy dataclass ¶

from_config classmethod ¶

TraceScope dataclass ¶

traceparent ¶

bind_run ¶

build_otel_observer ¶

compose_event_sink ¶

current_run_scope ¶

current_trace ¶

observers_from_config ¶

parse_traceparent ¶

redact_trace_value ¶

shutdown_otel ¶

trace_for_run ¶

unbind_run ¶

usage_dict ¶

`symfonic.services.observability`¶

CostObservation `dataclass` ¶

EventsDropped `dataclass` ¶

on_run_failed `async` ¶

aclose `async` ¶

on_cost `async` ¶

on_run_failed `async` ¶

release_run `async` ¶

aclose `async` ¶

PromptCapturePolicy `dataclass` ¶

argument_names `staticmethod` ¶

from_config `classmethod` ¶

RunFailed `dataclass` ¶

RunFinished `dataclass` ¶

RunScope `dataclass` ¶

RunStarted `dataclass` ¶

TextEmitted `dataclass` ¶

ToolCompleted `dataclass` ¶

ToolInvoked `dataclass` ¶

TracePolicy `dataclass` ¶

from_config `classmethod` ¶

TraceScope `dataclass` ¶