symfonic.core.observability.metrics

metrics

ConversationMetricsCollector — aggregates token/cost per conversation.

Implements the CallbackHandler protocol so it can be registered directly with SymfonicAgent(metrics_collector=...) or added to any CallbackManager.

Sprint 4 memory safety: bounded call_records deque, LRU-capped _conversations, run-scoped map purge on on_agent_end, and observable budget-tracker fan-out errors.

ConversationMetricsCollector

ConversationMetricsCollector()

Aggregates token usage and cost per conversation (session_id).

Satisfies the CallbackHandler protocol via structural typing.

Usage::

collector = ConversationMetricsCollector()
agent = SymfonicAgent(..., metrics_collector=collector)
# after some runs:
record = collector.get_conversation(session_id)
all_records = collector.list_conversations()

Source code in src/symfonic/core/observability/metrics.py

def __init__(self) -> None:
    self._conversations: dict[str, ConversationRecord] = {}
    # run_id → most-recent TokenCompositionEvent
    self._current_composition: dict[str, TokenCompositionEvent] = {}
    # run_id → conversation_id
    self._current_conversation: dict[str, str] = {}
    # run_id → tenant_id (required for TokenBudgetTracker fan-out)
    self._current_tenant: dict[str, str] = {}

get_conversation

get_conversation(
    conversation_id: str,
) -> ConversationRecord | None

Return the record for a single conversation, or None.

Source code in src/symfonic/core/observability/metrics.py

def get_conversation(self, conversation_id: str) -> ConversationRecord | None:
    """Return the record for a single conversation, or None."""
    return self._conversations.get(conversation_id)

list_conversations

list_conversations(
    limit: int = 50, tenant_id: str | None = None
) -> list[ConversationRecord]

Return conversations sorted by last_seen (most recent first).

When tenant_id is provided, only conversations belonging to that tenant are returned. The tenant filter is applied before the limit slice so a tenant always sees up to limit of its own conversations regardless of other tenants' activity.

Source code in src/symfonic/core/observability/metrics.py

def list_conversations(
    self, limit: int = 50, tenant_id: str | None = None
) -> list[ConversationRecord]:
    """Return conversations sorted by last_seen (most recent first).

    When ``tenant_id`` is provided, only conversations belonging to that
    tenant are returned. The tenant filter is applied *before* the
    ``limit`` slice so a tenant always sees up to ``limit`` of its own
    conversations regardless of other tenants' activity.
    """
    records = self._conversations.values()
    if tenant_id is not None:
        records = [c for c in records if c.tenant_id == tenant_id]
    return sorted(
        records,
        key=lambda c: c.last_seen,
        reverse=True,
    )[:limit]

on_agent_end async

on_agent_end(event: AgentEndEvent) -> None

Increment turn count and purge run-scoped bookkeeping.

Source code in src/symfonic/core/observability/metrics.py

async def on_agent_end(self, event: AgentEndEvent) -> None:
    """Increment turn count and purge run-scoped bookkeeping."""
    conv_id = self._current_conversation.get(event.run_id, event.run_id)
    if conv_id in self._conversations:
        self._conversations[conv_id].turns += 1
    self._current_composition.pop(event.run_id, None)
    self._current_conversation.pop(event.run_id, None)
    self._current_tenant.pop(event.run_id, None)

on_llm_end async

on_llm_end(event: LLMEndEvent) -> None

Accumulate token/cost metrics into the matching conversation record.

Also fans the same usage out to the process-wide :class:TokenBudgetTracker (if one has been registered via set_budget_tracker) so per-tenant ceilings can fire on the NEXT call. Failures in the tracker must never break the LLM flow.

Source code in src/symfonic/core/observability/metrics.py

async def on_llm_end(self, event: LLMEndEvent) -> None:
    """Accumulate token/cost metrics into the matching conversation record.

    Also fans the same usage out to the process-wide
    :class:`TokenBudgetTracker` (if one has been registered via
    ``set_budget_tracker``) so per-tenant ceilings can fire on the
    NEXT call.  Failures in the tracker must never break the LLM flow.
    """
    usage = TokenUsage.from_dict(event.usage, event.model)
    conv_id = self._current_conversation.get(event.run_id, event.run_id)
    tenant_id = self._current_tenant.get(event.run_id, "")

    conv = self._conversations.setdefault(
        conv_id,
        ConversationRecord(conversation_id=conv_id, tenant_id=tenant_id),
    )
    self._evict_conversations_if_over_cap()
    # Back-fill tenant_id if the record pre-dates set_tenant() being called.
    if tenant_id and not conv.tenant_id:
        conv.tenant_id = tenant_id
    conv.llm_calls += 1
    conv.input_tokens += usage.input_tokens
    conv.output_tokens += usage.output_tokens
    conv.cached_tokens += usage.cached_tokens
    conv.cost_usd = round(conv.cost_usd + usage.cost_usd, 6)
    conv.last_seen = datetime.now(UTC)

    # Fan usage into the budget tracker (optional / best-effort).
    # Budget tracking must never break the LLM flow, but failures
    # MUST be observable -- silent suppression previously hid real
    # tracker regressions for weeks.
    from symfonic.core.observability.budget import get_budget_tracker
    tracker = get_budget_tracker()
    if tracker is not None and conv.tenant_id:
        try:
            await tracker.record(
                conv.tenant_id,
                event.model,
                {
                    "input_tokens": usage.input_tokens,
                    "output_tokens": usage.output_tokens,
                    "cached_tokens": usage.cached_tokens,
                    "cache_write_tokens": getattr(
                        usage, "cache_creation_tokens", 0,
                    ),
                },
            )
        except Exception as exc:
            logger.warning(
                "budget tracker fan-out failed: %s", exc, exc_info=True,
            )

    comp = self._current_composition.get(event.run_id)
    conv.call_records.append(
        LLMCallRecord(
            call_id=event.run_id,
            conversation_id=conv_id,
            model=event.model,
            # Real call-site discriminator emitted since v7.4.3.  Fall
            # back to "react" only when the emitter left it empty (pre-
            # v7.4.3 payloads) so the historical default is preserved.
            node=event.node_name or "react",
            input_tokens=usage.input_tokens,
            cached_tokens=usage.cached_tokens,
            output_tokens=usage.output_tokens,
            cost_usd=usage.cost_usd,
            # iteration 0 is the turn's opening call; later React hops
            # are tool follow-ups.  The event carries no explicit
            # call_type, so derive it from the iteration index.
            call_type="initial" if event.iteration_index == 0 else "tool_followup",
            timestamp=datetime.now(UTC),
            composition={
                "system_prompt": comp.system_prompt_tokens if comp else 0,
                "tool_definitions": comp.tool_definitions_tokens if comp else 0,
                "conversation_history": comp.conversation_history_tokens if comp else 0,
                "memory_context": comp.memory_context_tokens if comp else 0,
            },
            cache_creation_tokens=usage.cache_creation_tokens,
            reasoning_tokens=usage.reasoning_tokens,
            duration_ms=event.duration_ms,
            iteration_index=event.iteration_index,
            turn_index=event.turn_index,
        )
    )

    # v8.8.0 — fan the same per-call detail into the durable metrics
    # store, if one is registered, so the admin dashboards survive a
    # worker restart.  Best-effort + synchronous O(1) enqueue (the
    # BufferedMetricsSink batches the actual DB I/O off the hot path);
    # a sink failure must NEVER break the LLM flow.
    from symfonic.core.observability.metrics_store import get_metrics_store
    sink = get_metrics_store()
    if sink is not None:
        try:
            sink.record(_record_to_snapshot(conv.call_records[-1], conv.tenant_id))
        except Exception as exc:  # noqa: BLE001 - telemetry never propagates
            logger.warning("metrics sink fan-out failed: %s", exc, exc_info=True)

on_token_composition async

on_token_composition(event: TokenCompositionEvent) -> None

Cache the composition breakdown keyed by run_id.

Source code in src/symfonic/core/observability/metrics.py

async def on_token_composition(self, event: TokenCompositionEvent) -> None:
    """Cache the composition breakdown keyed by run_id."""
    self._current_composition[event.run_id] = event

set_conversation

set_conversation(run_id: str, conversation_id: str) -> None

Associate a run_id with a conversation (session) id.

Call this before the agent run so metrics land under the right conversation bucket (e.g. use the session_id as conversation_id).

Source code in src/symfonic/core/observability/metrics.py

def set_conversation(self, run_id: str, conversation_id: str) -> None:
    """Associate a run_id with a conversation (session) id.

    Call this before the agent run so metrics land under the right
    conversation bucket (e.g. use the session_id as conversation_id).
    """
    self._current_conversation[run_id] = conversation_id

set_tenant

set_tenant(run_id: str, tenant_id: str) -> None

Associate a run_id with a tenant_id.

Call this before the agent run so on_llm_end can fan usage into the registered TokenBudgetTracker. Without this the tracker never fires because the ConversationRecord's tenant_id stays empty.

Source code in src/symfonic/core/observability/metrics.py

def set_tenant(self, run_id: str, tenant_id: str) -> None:
    """Associate a run_id with a tenant_id.

    Call this before the agent run so ``on_llm_end`` can fan usage
    into the registered ``TokenBudgetTracker``.  Without this the
    tracker never fires because the ConversationRecord's tenant_id
    stays empty.
    """
    if tenant_id:
        self._current_tenant[run_id] = tenant_id

snapshots

snapshots() -> list[dict[str, Any]]

Return per-LLM-call metric snapshots as plain dicts.

Public, stable API for admin dashboards and downstream aggregation. Each snapshot represents a single LLM call, enriched with the parent conversation's tenant_id so callers can filter by tenant without joining.

The return value is a newly constructed list of dicts — mutating it does not affect the collector's internal state.

The dict shape is intentionally permissive so dashboards can continue to work across minor releases; currently includes:

conversation_id, call_id, tenant_id, model, node,
input_tokens, output_tokens, cached_tokens, cost_usd,
call_type, timestamp, created_at, composition,
cache_creation_tokens, reasoning_tokens, duration_ms,
iteration_index, turn_index.

The last five (v8.7.11) surface the per-call detail the framework already emits on LLMEndEvent — real node names, latency, React iteration + user-turn index, and the cache-creation/reasoning token splits — so a conversation-detail drill-down can render them without an external OTel backend.

Source code in src/symfonic/core/observability/metrics.py

def snapshots(self) -> list[dict[str, Any]]:
    """Return per-LLM-call metric snapshots as plain dicts.

    Public, stable API for admin dashboards and downstream aggregation.
    Each snapshot represents a single LLM call, enriched with the
    parent conversation's ``tenant_id`` so callers can filter by tenant
    without joining.

    The return value is a newly constructed list of dicts — mutating
    it does not affect the collector's internal state.

    The dict shape is intentionally permissive so dashboards can
    continue to work across minor releases; currently includes:

        conversation_id, call_id, tenant_id, model, node,
        input_tokens, output_tokens, cached_tokens, cost_usd,
        call_type, timestamp, created_at, composition,
        cache_creation_tokens, reasoning_tokens, duration_ms,
        iteration_index, turn_index.

    The last five (v8.7.11) surface the per-call detail the framework
    already emits on ``LLMEndEvent`` — real node names, latency, React
    iteration + user-turn index, and the cache-creation/reasoning
    token splits — so a conversation-detail drill-down can render them
    without an external OTel backend.
    """
    out: list[dict[str, Any]] = []
    for conv in self._conversations.values():
        for rec in conv.call_records:
            out.append(_record_to_snapshot(rec, conv.tenant_id))
    return out

summary

summary(
    start: datetime | None = None,
    end: datetime | None = None,
    tenant_id: str | None = None,
) -> dict[str, Any]

Return aggregate totals across all conversations in the time window.

When tenant_id is provided only that tenant's conversations are aggregated, preventing cross-tenant metric leakage.

Source code in src/symfonic/core/observability/metrics.py

def summary(
    self,
    start: datetime | None = None,
    end: datetime | None = None,
    tenant_id: str | None = None,
) -> dict[str, Any]:
    """Return aggregate totals across all conversations in the time window.

    When ``tenant_id`` is provided only that tenant's conversations are
    aggregated, preventing cross-tenant metric leakage.
    """
    records = list(self._conversations.values())
    if tenant_id is not None:
        records = [r for r in records if r.tenant_id == tenant_id]
    if start is not None:
        records = [r for r in records if r.last_seen >= start]
    if end is not None:
        records = [r for r in records if r.last_seen <= end]

    return {
        "conversations": len(records),
        "total_llm_calls": sum(r.llm_calls for r in records),
        "total_cost_usd": round(sum(r.cost_usd for r in records), 6),
        "total_tokens": sum(
            r.input_tokens + r.output_tokens for r in records
        ),
    }

ConversationRecord dataclass

ConversationRecord(
    conversation_id: str,
    tenant_id: str,
    turns: int = 0,
    llm_calls: int = 0,
    input_tokens: int = 0,
    output_tokens: int = 0,
    cached_tokens: int = 0,
    cost_usd: float = 0.0,
    first_seen: datetime = (lambda: datetime.now(UTC))(),
    last_seen: datetime = (lambda: datetime.now(UTC))(),
    call_records: deque[LLMCallRecord] = (
        lambda: deque(
            maxlen=MAX_CALL_RECORDS_PER_CONVERSATION
        )
    )(),
)

Aggregated per-conversation metrics.

LLMCallRecord dataclass

LLMCallRecord(
    call_id: str,
    conversation_id: str,
    model: str,
    node: str,
    input_tokens: int,
    cached_tokens: int,
    output_tokens: int,
    cost_usd: float,
    call_type: str,
    timestamp: datetime,
    composition: dict[str, int] = dict(),
    cache_creation_tokens: int = 0,
    reasoning_tokens: int = 0,
    duration_ms: float = 0.0,
    iteration_index: int = 0,
    turn_index: int = 0,
)

Single LLM call record for dashboard display.

v8.7.11 — the record now carries the full per-call telemetry the framework already emits on :class:LLMEndEvent (real node name, call_type, iteration_index, turn_index, duration_ms, and the cache_creation/reasoning token splits) instead of hardcoding node="react" / call_type="initial" and dropping latency. All new fields default to zero/empty so pre-v8.7.11 callers that construct the record positionally with the original arg set keep working unchanged.