symfonic.agent.cutover.observability

observability

The migrated path's observability: one sink, bound at the compile seam.

TA8.20 implements the MIGRATE verdict for the observability envelope guard. TA8.15's decision-callbacks.md calls the implementing task TA8.21, which in the ledger is different work entirely; see RET-PREP/observability-migration.md §0 for the mapping.

Before it, an agent constructed with a metrics_collector or with otel_enabled was refused by the envelope and every one of its turns ran the legacy body: the migrated path emitted no callback events, so admitting the turn would have left the collector silent, and silence is invisible to cutover.fallbacks().

The metrics half of that is what TA8.20 changed. The OTEL half stayed wired and unreachable: an otel_enabled agent was still refused, one frame later, at the config guard, because the field was not on ALLOWED_FRAMEWORK_FIELDS. TA8.40 closed that gap and paid what it owed. All four otel_* fields are on the allowlist, so the observer this module composes now runs on an admitted turn — on run, stream and stream_typed alike — and the span reduction that admission asserts is enumerated in tests/agent/cutover/test_otel_fields_reach_the_kernel.py rather than read off the wiring. The distinction between "bound" and "migrated" is what that file measures; policy.py's allowlist comment states the adopter-visible change.

What replaces the refusal is a consumer, not a tolerance. The kernel already carries the whole run as KernelEvents and already reserves ServiceBindings.event_sink for a consumer of them; symfonic.services.observability already ships the projection from that stream onto the same two objects the constructor injected — ObservabilityBridge -> MetricsObserver / OTelObserver -> CallbackHandlerObserver -> the shipped CallbackHandler dataclasses. Both halves were built, tested, and joined to nothing: before this module compose_event_sink had no production caller outside its own package. This module is the join, and it is deliberately the only thing here — the projection is not re-implemented, and no observer is written.

Why the sink is per run and the suite is per agent. RunScope is the identity every observation is attributed to, and three of its five fields (run_id, session_id, tenant_id) are facts about one turn. An agent that built one sink at construction would attribute every later run to the first one's ids. The inputs to that decision — the config, the collector, the exporter handles — are agent-lifetime, so they live on :class:ObservabilitySuite and the per-run half is :meth:ObservabilitySuite.for_run.

Why the exporter handles are passed in rather than resolved here. The engine already builds them once (build_if_enabled at construction) and already registers their framework bridge into its hook list. Resolving a second set through :func:~symfonic.services.observability.otel.build_otel_observer's default path would give one agent two TracerProviders, two batch workers and two channels, and spans opened against one would be flushed by the other's shutdown. Passing the engine's handles in as the builder is what keeps "one agent, one exporter" true across both routes.

What this module does not do. It does not pass callback_handlers= to observers_from_config. That parameter exists and would expose a four-hook subscription surface to adopter handlers; TA8.15 §6.3 rejected it by name, because an eight-hook adopter handler admitted through it would silently lose four of its hooks — the exact failure the retired callbacks argument is refused to prevent, merely relocated. The parameter stays unreachable from the agent path.

ObservabilitySuite

ObservabilitySuite(*, config: Any = None, collector: Any = None, otel_handles: Any = None)

The agent-lifetime half: what this configuration buys, decided once.

inactive is the ordinary case and costs nothing: no collector, no OTEL, so :meth:compose returns None before it imports anything, the plan's event_sink stays unbound, and InvocationRunner never constructs a CallbackEventAdapter. That is the same zero-overhead guarantee compose_event_sink expresses in its return type, kept one level up so a disabled configuration does not pay for the import either.

Source code in src/symfonic/agent/cutover/observability.py

def __init__(
    self,
    *,
    config: Any = None,
    collector: Any = None,
    otel_handles: Any = None,
) -> None:
    self._config = config
    self._collector = collector
    self._otel_handles = otel_handles

active property

active: bool

Whether anything is watching. Mirrors the envelope's observability.

The two must agree: the envelope reads SymfonicAgent._with_metrics_callbacks(None), which is truthy when a collector or the OTEL bridge is present, and admits the turn on the strength of this suite serving them. An agent admitted on a truthy observability whose suite was inactive would be the exact silence the refusal used to prevent, reached through the admission instead.

compose

compose(*, run_id: str, session_id: str | None, tenant_id: str | None, entry_point: str, model: Any, prompt: str = '', lineage: tuple[str, str | None] = ('', None)) -> Any

This run's sink, or None when nobody is watching. Never raises.

session_id=None is correct here, not merely tolerated, and it is checked rather than inherited. TA8.15 recorded the claim as an INFERENCE from services/observability/metrics.py:39-47 and asked the implementing task to confirm it. Confirmed, and the evidence is stronger than the inference: MetricsObserver.on_run_started calls set_conversation(scope.run_id, scope.session_id or scope.run_id), and ConversationMetricsCollector.on_llm_end independently defaults the same lookup to the run id (core/observability/metrics.py:203/:235 — self._current_conversation.get(event.run_id, event.run_id)). So a session-less run is filed under its own run id by both halves rather than being bucketed with unrelated runs, which is what the legacy body does for the same turn. The premise the inference was drawn under has since moved — session_id is an admitted argument as of TA8.19, so an admitted scoped turn now carries a resolved id and reaches this argument with a value — and None survives exactly where the legacy body also has none: an unscoped turn whose caller supplied no id, for which SymfonicAgent._resolve_session returns the caller's None unchanged.

model is the plan's ModelResolution. RunScope.model is a str, and an unnamed model resolves to "" rather than to the string "None": provider_family is carried alongside it because the composer already knows it and re-deriving it downstream would create a second answer.

Failure to build the sink is not failure to run. Telemetry that cannot be composed is telemetry that reports nothing — the same rule ObservabilityBridge applies per event, applied once to the wiring.

Source code in src/symfonic/agent/cutover/observability.py

def compose(
    self,
    *,
    run_id: str,
    session_id: str | None,
    tenant_id: str | None,
    entry_point: str,
    model: Any,
    prompt: str = "",
    lineage: tuple[str, str | None] = ("", None),
) -> Any:
    """This run's sink, or ``None`` when nobody is watching. Never raises.

    **``session_id=None`` is correct here, not merely tolerated**, and it is
    checked rather than inherited. TA8.15 recorded the claim as an
    ``INFERENCE`` from ``services/observability/metrics.py:39-47`` and asked
    the implementing task to confirm it. Confirmed, and the evidence is
    stronger than the inference: ``MetricsObserver.on_run_started`` calls
    ``set_conversation(scope.run_id, scope.session_id or scope.run_id)``,
    and ``ConversationMetricsCollector.on_llm_end`` independently defaults
    the same lookup to the run id
    (``core/observability/metrics.py:203``/``:235`` —
    ``self._current_conversation.get(event.run_id, event.run_id)``). So a
    session-less run is filed under its own run id by *both* halves rather
    than being bucketed with unrelated runs, which is what the legacy body
    does for the same turn. The premise the inference was drawn under has
    since moved — ``session_id`` is an admitted argument as of TA8.19, so an
    admitted *scoped* turn now carries a resolved id and reaches this
    argument with a value — and ``None`` survives exactly where the legacy
    body also has none: an unscoped turn whose caller supplied no id, for
    which ``SymfonicAgent._resolve_session`` returns the caller's ``None``
    unchanged.

    ``model`` is the plan's ``ModelResolution``. ``RunScope.model`` is a
    ``str``, and an unnamed model resolves to ``""`` rather than to the
    string ``"None"``: ``provider_family`` is carried alongside it because
    the composer already knows it and re-deriving it downstream would create
    a second answer.

    Failure to build the sink is not failure to run. Telemetry that cannot
    be composed is telemetry that reports nothing — the same rule
    ``ObservabilityBridge`` applies per event, applied once to the wiring.
    """
    if not self.active:
        return None
    try:
        # Through the service's declared port, not its internals:
        # facade-compiler -> runtime-service is ``port`` in the dependency
        # matrix, and ``symfonic.services.observability.contracts`` is the
        # narrow interface TA8.20 declared for this edge.
        from symfonic.services.observability.contracts import (
            RunScope,
            compose_event_sink,
            observers_from_config,
        )

        observers = observers_from_config(
            self._config,
            collector=self._collector,
            # TA8.15 §6.3: deliberately no ``callback_handlers=``.
            otel_builder=self._otel_builder(),
        )
        scope = RunScope(
            run_id=run_id,
            tenant_id=tenant_id,
            session_id=session_id,
            entry_point=entry_point,
            model=getattr(model, "model_name", None) or "",
            provider_family=getattr(model, "provider_family", "") or "",
            # Payload, and therefore gated: ``compose_event_sink`` is given
            # the config below, so ``PromptCapturePolicy.from_config``
            # decides whether an observer ever sees this string. Carrying it
            # into the scope is what makes ``observability_redact_payloads``
            # a control with something to withhold rather than a flag over
            # an empty field.
            root_run_id=lineage[0],
            parent_run_id=lineage[1],
            prompt=prompt,
        )
        return compose_event_sink(scope, observers, config=self._config)
    except Exception:  # noqa: BLE001 - telemetry never breaks a run
        _logger().warning(
            "the migrated path could not compose an observability sink for "
            "run %s; the turn continues without it",
            run_id,
            exc_info=True,
        )
        return None

for_run

for_run(*, run_id: str, session_id: str | None, tenant_id: str | None, entry_point: str, prompt: str = '', request: Any = None) -> RunObservability

The handle the delegate hands to the compile seam and then releases.

session_id is str | None and None is a supported value, not a degraded one — see :meth:compose.

prompt is the turn's query and it is carried rather than defaulted, because AgentStartEvent.query is rendered from it (ObservabilityBridge builds RunStarted(query=capture.prompt_text( scope.prompt)), services/observability/bridge.py:140-141). Leaving it at "" would have made one of the four rendered hooks carry an empty field on the migrated route while the legacy body carried the query — a silent field-level divergence inside a hook this migration claims parity for, which is worse than a reduction that is written down.

Source code in src/symfonic/agent/cutover/observability.py

def for_run(
    self,
    *,
    run_id: str,
    session_id: str | None,
    tenant_id: str | None,
    entry_point: str,
    prompt: str = "",
    request: Any = None,
) -> RunObservability:
    """The handle the delegate hands to the compile seam and then releases.

    ``session_id`` is ``str | None`` and ``None`` is a supported value, not
    a degraded one — see :meth:`compose`.

    ``prompt`` is the turn's query and it is carried rather than defaulted,
    because ``AgentStartEvent.query`` is rendered from it
    (``ObservabilityBridge`` builds ``RunStarted(query=capture.prompt_text(
    scope.prompt))``, ``services/observability/bridge.py:140-141``). Leaving
    it at ``""`` would have made one of the four *rendered* hooks carry an
    empty field on the migrated route while the legacy body carried the
    query — a silent field-level divergence inside a hook this migration
    claims parity for, which is worse than a reduction that is written down.
    """
    return RunObservability(
        self,
        run_id=run_id,
        session_id=session_id,
        tenant_id=tenant_id,
        entry_point=entry_point,
        prompt=prompt,
        request=request,
    )

RunObservability

RunObservability(suite: ObservabilitySuite, *, run_id: str, session_id: str | None, tenant_id: str | None, entry_point: str, prompt: str = '', request: Any = None)

One run's event sink: built when the plan compiles, released with it.

Instances are callables so they can be handed to AgentPlanFactory.compile(..., sink_factory=...): the factory calls this object once, with the ModelResolution it just computed, and binds the return value as ServiceBindings.event_sink. Taking the resolution as an argument rather than re-deriving the model here is what keeps RunScope naming the model the plan actually runs — a second resolution is a second answer that can disagree with the first.

:attr:sink is None until the plan compiles, and stays None when nothing is watching. :meth:release is safe in either state and safe after a normal terminal, which is why the caller may put it in a bare finally.

Source code in src/symfonic/agent/cutover/run_observability.py

def __init__(
    self,
    suite: ObservabilitySuite,
    *,
    run_id: str,
    session_id: str | None,
    tenant_id: str | None,
    entry_point: str,
    prompt: str = "",
    request: Any = None,
) -> None:
    self._suite = suite
    self._lineage = (
        str(getattr(request, "root_run_id", "") or ""),
        str(getattr(request, "parent_run_id", "") or "") or None,
    )
    self._run_id = run_id
    self._session_id = session_id
    self._tenant_id = tenant_id
    self._entry_point = entry_point
    self._prompt = prompt
    self.sink: Any = None

release async

release() -> None

Release what an abandoned run left held. Idempotent, never raises.

Source code in src/symfonic/agent/cutover/run_observability.py

async def release(self) -> None:
    """Release what an abandoned run left held. Idempotent, never raises."""
    sink = self.sink
    if sink is None:
        return
    aclose = getattr(sink, "aclose", None)
    if aclose is None:
        return
    await aclose()

for_turn

for_turn(suite: ObservabilitySuite | None, *, run_id: str, session_id: str, tenant_id: str | None, entry_point: str, prompt: str, request: Any = None) -> RunObservability | None

This turn's observability handle, or None when nobody is watching.

The delegate's whole share of the migration is this call and the release that pairs with it, which is why the decision lives here rather than as a private method there: what makes a handle worth building is a fact about the suite, and a caller that re-derived it would be a second reader of active that can disagree with the first.

session_id arrives as str because that is the shape the facade echoes, and "" is normalised back to None rather than carried: RunScope.session_id is str | None and None is what its consumers test for. Passing "" would be a session id that is falsy but present, which reads as a session in every log line that renders it and as no session in every branch that checks it.

request is the TurnRequest this turn enters the kernel with, read rather than re-derived: telemetry computing its own lineage from the ambient identity is the second-reader problem this module already avoids for active. Absent, the turn reports a root run (issue #131).

Source code in src/symfonic/agent/cutover/observability.py

def for_turn(
    suite: ObservabilitySuite | None,
    *,
    run_id: str,
    session_id: str,
    tenant_id: str | None,
    entry_point: str,
    prompt: str,
    request: Any = None,
) -> RunObservability | None:
    """This turn's observability handle, or ``None`` when nobody is watching.

    The delegate's whole share of the migration is this call and the
    ``release`` that pairs with it, which is why the decision lives here rather
    than as a private method there: what makes a handle worth building is a fact
    about the *suite*, and a caller that re-derived it would be a second reader
    of ``active`` that can disagree with the first.

    ``session_id`` arrives as ``str`` because that is the shape the facade
    echoes, and ``""`` is normalised back to ``None`` rather than carried:
    ``RunScope.session_id`` is ``str | None`` and ``None`` is what its consumers
    test for. Passing ``""`` would be a session id that is falsy but present,
    which reads as a session in every log line that renders it and as no session
    in every branch that checks it.

    ``request`` is the ``TurnRequest`` this turn enters the kernel with, read
    rather than re-derived: telemetry computing its own lineage from the ambient
    identity is the second-reader problem this module already avoids for
    ``active``. Absent, the turn reports a root run (issue #131).
    """
    if suite is None or not suite.active:
        return None
    return suite.for_run(
        run_id=run_id,
        session_id=session_id or None,
        tenant_id=tenant_id,
        entry_point=entry_point,
        prompt=prompt,
        request=request,
    )