symfonic.core.observability.metrics_query

metrics_query

Read side of the conversation metrics collector.

Split out of :mod:symfonic.core.observability.metrics (409 lines against the 300-line budget). The collector has two halves that share exactly one attribute: the on_* handlers write self._conversations as callback events arrive, and the four methods here read it for dashboards and admin endpoints. The read half is the stable public API -- snapshots() and summary() are what adopters call -- so it is worth being able to read it without scrolling past the fold logic.

ConversationMetricsCollector mixes this in, so every call site is unchanged; _conversations is declared here as the contract the mixin requires of its host.

MetricsQueryMixin

The collector's read API: one conversation, many, or the aggregate.

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_query.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_query.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]

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_query.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_query.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
        ),
    }