Skip to content

symfonic.services.observability.metrics

metrics

The metrics adapter — the shipped collector, fed from the event stream.

ConversationMetricsCollector is public API with a documented record shape and a durable-store fan-out behind it. Replacing it would be a breaking change for no benefit; what this task replaces is how it is fed.

The one thing this adapter must own is the run-to-conversation binding. The collector keys its rows on a conversation id supplied out of band (set_conversation / set_tenant) — which is exactly the kind of "remember to call this first" contract that quietly produces rows filed under a run id when a caller forgets. Here it is bound from RunScope on run start, every time, before any usage can arrive.

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 
45
46
47
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 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
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)