Skip to content

symfonic.services.observability.ports

ports

The segregated observability ports (CON-P, interface segregation).

The shipped CallbackHandler protocol is one interface with fifteen methods. Anything that wanted a token count had to either implement all fifteen or rely on the manager's tolerance for partial handlers — which is how "did my handler get registered?" became a support question with a dedicated introspection helper.

Here each concern is its own protocol, and an observer implements exactly the ports it has an opinion about. The bridge resolves which ports an observer satisfies once, at wiring time, so an observer that only cares about cost is never asked about a text delta and never pays a per-event attribute lookup.

PORT_METHODS is the machine-readable form of that table. The bridge reads it rather than hard-coding method names, so adding a port is one row here plus its projection in :mod:symfonic.services.observability.bridge.

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.

RunObserver

Bases: Protocol

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

TextObserver

Bases: Protocol

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

ToolObserver

Bases: Protocol

Tool dispatch and completion.

resolve_ports 

resolve_ports(observers: Iterable[Any]) -> dict[str, tuple[Any, ...]]

Bind each observer to every port it fully implements, once.

Resolution happens here, at wiring time, rather than per event: the bridge then walks a list per port instead of asking every observer whether it cares. That is the difference between interface segregation as a design claim and as a runtime property.

An observer that implements no port is refused rather than silently ignored. Silent ignoring is how "my handler never fired" becomes a support ticket instead of a stack trace at startup.

Source code in src/symfonic/services/observability/ports.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def resolve_ports(observers: Iterable[Any]) -> dict[str, tuple[Any, ...]]:
    """Bind each observer to every port it *fully* implements, once.

    Resolution happens here, at wiring time, rather than per event: the bridge
    then walks a list per port instead of asking every observer whether it
    cares. That is the difference between interface segregation as a design
    claim and as a runtime property.

    An observer that implements no port is refused rather than silently
    ignored. Silent ignoring is how "my handler never fired" becomes a support
    ticket instead of a stack trace at startup.
    """
    table: dict[str, list[Any]] = {name: [] for name in PORT_METHODS}
    for observer in observers:
        if observer is None:
            continue
        matched = False
        for port, methods in PORT_METHODS.items():
            if all(callable(getattr(observer, name, None)) for name in methods):
                table[port].append(observer)
                matched = True
        if not matched:
            raise ConfigurationError(
                f"{type(observer).__name__} implements no observability port; "
                f"expected at least one of {sorted(PORT_METHODS)} "
                "(see symfonic.services.observability.ports)."
            )
    return {port: tuple(members) for port, members in table.items() if members}