Skip to content

symfonic.services.observability.contracts

contracts

The declared port of the observability service (LAY-ADR §2, port).

Same cell and same rule as :mod:symfonic.services.shadow.ports: facade-compiler → runtime-service is port, so :mod:symfonic.agent.cutover.observability may reach this service only through a declared narrow interface. TA8.20 is the first facade importer this package has ever had — before it, compose_event_sink had no production caller outside its own package at all — so the port is created by the same task that creates the edge, rather than the edge being registered as an exception.

Why this module is not called ports. That name is taken, and taken by a different idea: :mod:symfonic.services.observability.ports holds the CON-P segregated observer protocols the bridge resolves against, and it is imported by :mod:~symfonic.services.observability.bridge, which :mod:~symfonic.services.observability.suite imports in turn. Re-exporting the composition seam from there would close the loop ports → suite → bridge → ports and turn a layering fix into an import cycle. contracts is the other basename archcheck.toml's port_module_names licenses, and nothing in this package imports it, so the seam stays acyclic.

The three names below are exactly what a composition root needs to answer "what is watching this run, and what identity is it watching under" — the observer set, the sink that fans a kernel event stream onto it, and the scope every observation is attributed to. Nothing here is defined locally.

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.

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 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
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)