symfonic.services.observability.closing

closing

Releasing what a run that never finished left behind.

The kernel guarantees exactly one terminal event per run (EVT-1), and ObservabilityBridge._close is what acts on it: cost, finish, unbind. A cancelled run and a killed process deliver no terminal at all, and what they leave behind is invisible rather than loud.

Two things stay held. The run scope and trace scope stay bound to contextvars, so the next run driven from that task reads a stale identity and attributes its telemetry to a run that is already gone. And any observer holding per-run state keeps holding it — OTelObserver keeps an un-ended root span and an OTEL carrier token, which for an observer shared for the life of a process is unbounded, not merely per-task.

Why closing is duck-typed and not a seventh port. The six ports in :mod:.ports segregate what an observer is told; closing is not an observation, it is a lifetime. Making it a port would force every observer that holds nothing to answer a question about resources it does not have, which is the interface-segregation failure the ports exist to avoid. An observer that has something to release defines aclose; one that does not, does not.

Two lifetimes, two calls. A bridge is run-scoped and an observer need not be — OTelObserver is explicitly documented and tested as shareable across overlapping runs. So a bridge that answered its own abandonment by calling the observer's aclose released every run the observer held, not the one that was abandoned: run B's root span ended mid-run, its carrier token reset under it, and B's own terminal then found nothing to pop and no-opped in silence. Nothing raises; B's exported span is simply short. An observer that keeps per-run state therefore defines release_run(run_id) for that case and keeps aclose for process shutdown; close_observers prefers the first when it is given a run_id, and falls back to aclose for observers whose state is not keyed by run and for which the two calls mean the same thing.

Failures are counted, never raised. A telemetry adapter that throws on the way out must not take down the caller's cleanup path — the same rule the fan-out in :mod:.bridge follows, for the same reason.

abandoned_finish

abandoned_finish(scope: Any) -> RunFinished

The terminal an abandoned run still owes its observers (TA8.29).

A run that never reached a terminal still OPENED a run scope, and leaving it open is the unbalanced-pair defect one level up: the observer saw a start and will never see an end.

Closed is all this is. No cost is priced -- nothing was reported for a run that stopped mid-flight -- no text is invented, and duration_ms stays 0.0 rather than becoming a plausible number nobody measured. Closing an open scope and fabricating a finished one are opposite acts, and only the first is cleanup. index=-1 says the same thing about position: this finish corresponds to no event on the stream.

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

def abandoned_finish(scope: Any) -> RunFinished:
    """The terminal an abandoned run still owes its observers (TA8.29).

    A run that never reached a terminal still OPENED a run scope, and leaving
    it open is the unbalanced-pair defect one level up: the observer saw a
    start and will never see an end.

    Closed is all this is. No cost is priced -- nothing was reported for a run
    that stopped mid-flight -- no text is invented, and ``duration_ms`` stays
    ``0.0`` rather than becoming a plausible number nobody measured. Closing an
    open scope and fabricating a finished one are opposite acts, and only the
    first is cleanup. ``index=-1`` says the same thing about position: this
    finish corresponds to no event on the stream.
    """
    return RunFinished(
        scope=scope,
        index=-1,
        reason="abandoned",
        text=None,
        usage=UsageDelta(),
        duration_ms=0.0,
        stop_reason=None,
    )

close_observers async

close_observers(observers: Iterable[Any], *, run_id: str | None = None) -> int

Release what these observers hold. Never raises.

run_id narrows the release to one run: an observer that defines release_run is asked to drop that run alone and keeps everything it holds for the others. Omit it — process shutdown — and every observer's aclose runs, which is the wholesale release.

Deduplicated by identity: one object registered on several ports is one lifetime, so it is released once. Returns how many releases failed, for the caller's own failure tally.

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

async def close_observers(
    observers: Iterable[Any], *, run_id: str | None = None
) -> int:
    """Release what these observers hold. Never raises.

    ``run_id`` narrows the release to one run: an observer that defines
    ``release_run`` is asked to drop that run alone and keeps everything it
    holds for the others. Omit it — process shutdown — and every observer's
    ``aclose`` runs, which is the wholesale release.

    Deduplicated by identity: one object registered on several ports is one
    lifetime, so it is released once. Returns how many releases failed, for the
    caller's own failure tally.
    """
    failures = 0
    seen: set[int] = set()
    for observer in observers:
        if id(observer) in seen:
            continue
        seen.add(id(observer))
        scoped = run_id is not None and hasattr(observer, "release_run")
        method = "release_run" if scoped else "aclose"
        release = getattr(observer, method, None)
        if release is None:
            continue
        try:
            await (release(run_id) if scoped else release())
        except Exception:  # noqa: BLE001 - telemetry never breaks a run
            failures += 1
            logger.warning(
                "observability observer %s.%s raised; suppressing",
                type(observer).__name__,
                method,
                exc_info=True,
            )
    return failures