Skip to content

symfonic.core.observability.metrics_store

metrics_store

Storage-agnostic durable metrics contracts, buffering, and factory.

The same sink persists billable model calls and metadata-only execution events.

BufferedMetricsSink 

BufferedMetricsSink(store: MetricsStore, *, batch_size: int = 25, flush_interval_s: float = 2.0, max_queue: int = 10000)

Hot-path buffer around any :class:MetricsStore.

record() is a synchronous O(1) enqueue so the collector fan-out in on_llm_end never awaits DB I/O. A background task flushes the queue every flush_interval_s or once batch_size rows are pending, calling the wrapped store's record_calls once per batch. All read / lifecycle methods delegate to the wrapped store, so get_metrics_store() returns a single object serving both write and read.

Source code in src/symfonic/core/observability/metrics_store.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def __init__(
    self,
    store: MetricsStore,
    *,
    batch_size: int = 25,
    flush_interval_s: float = 2.0,
    max_queue: int = 10_000,
) -> None:
    self._store = store
    self._batch_size = max(1, batch_size)
    self._flush_interval_s = flush_interval_s
    self._max_queue = max(1, max_queue)
    self._queue: deque[dict[str, Any]] = deque(maxlen=self._max_queue)
    self._event_queue: deque[dict[str, Any]] = deque(maxlen=self._max_queue)
    self._dropped = 0
    self._failed_call_records = 0
    self._failed_event_records = 0
    self._task: asyncio.Task[None] | None = None
    self._stopped = False
    self._wake = asyncio.Event()
    self._drain_lock = asyncio.Lock()

dropped property 

dropped: int

All rows irrevocably lost through overflow or failed persistence.

loss_counts property 

loss_counts: dict[str, int]

Losses by durable record family, safe for health reporting.

ensure_indexes 

ensure_indexes() -> None

Delegate to the wrapped store's index bootstrap, if it has one.

Source code in src/symfonic/core/observability/metrics_store.py 
265
266
267
268
269
def ensure_indexes(self) -> None:
    """Delegate to the wrapped store's index bootstrap, if it has one."""
    ensure = getattr(self._store, "ensure_indexes", None)
    if ensure is not None:
        ensure()

flush async 

flush() -> None

Drain the whole queue now (deterministic for tests / shutdown).

Source code in src/symfonic/core/observability/metrics_store.py 
214
215
216
217
218
219
220
221
async def flush(self) -> None:
    """Drain the whole queue now (deterministic for tests / shutdown)."""
    while self._queue or self._event_queue:
        await self._drain_once()
    flush = getattr(self._store, "flush", None)
    if flush is not None:
        with contextlib.suppress(Exception):
            await flush()

record 

record(row: dict[str, Any]) -> None

Synchronous, non-blocking enqueue. Drops oldest on overflow.

Source code in src/symfonic/core/observability/metrics_store.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
def record(self, row: dict[str, Any]) -> None:
    """Synchronous, non-blocking enqueue. Drops oldest on overflow."""
    if len(self._queue) >= self._max_queue:
        self._dropped += 1
        if self._dropped == 1 or self._dropped % 1000 == 0:
            logger.warning(
                "metrics buffer full (max_queue=%d); dropped %d rows so far",
                self._max_queue, self._dropped,
            )
    self._queue.append(row)  # deque(maxlen) evicts the oldest when full
    if len(self._queue) >= self._batch_size:
        self._wake.set()
    self._ensure_task()

record_event 

record_event(row: dict[str, Any]) -> None

Enqueue one metadata-only execution event.

Source code in src/symfonic/core/observability/metrics_store.py 
152
153
154
155
156
157
158
159
def record_event(self, row: dict[str, Any]) -> None:
    """Enqueue one metadata-only execution event."""
    if len(self._event_queue) >= self._max_queue:
        self._dropped += 1
    self._event_queue.append(row)
    if len(self._event_queue) >= self._batch_size:
        self._wake.set()
    self._ensure_task()

start 

start() -> None

Explicitly start the background flusher (lifespan startup).

Source code in src/symfonic/core/observability/metrics_store.py 
173
174
175
def start(self) -> None:
    """Explicitly start the background flusher (lifespan startup)."""
    self._ensure_task()

MetricsReader

Bases: Protocol

Rich, tenant-scoped read side backing the admin dashboards.

conversation_detail async 

conversation_detail(tenant_id: str, conversation_id: str) -> list[dict[str, Any]]

Return snapshot-shaped rows for one conversation, oldest-first.

The shape MUST match ConversationMetricsCollector.snapshots() so the scaffold's build_conversation_detail() is reused unchanged.

Source code in src/symfonic/core/observability/metrics_store.py 
47
48
49
50
51
52
53
54
55
async def conversation_detail(  # pragma: no cover - protocol
    self, tenant_id: str, conversation_id: str,
) -> list[dict[str, Any]]:
    """Return snapshot-shaped rows for one conversation, oldest-first.

    The shape MUST match ``ConversationMetricsCollector.snapshots()`` so
    the scaffold's ``build_conversation_detail()`` is reused unchanged.
    """
    ...

execution_detail async 

execution_detail(tenant_id: str, conversation_id: str) -> list[dict[str, Any]]

Return metadata-only execution events, oldest-first.

Source code in src/symfonic/core/observability/metrics_store.py 
75
76
77
78
79
async def execution_detail(  # pragma: no cover - protocol
    self, tenant_id: str, conversation_id: str,
) -> list[dict[str, Any]]:
    """Return metadata-only execution events, oldest-first."""
    ...

list_conversations async 

list_conversations(tenant_id: str, limit: int, offset: int) -> tuple[list[dict[str, Any]], int]

Return (rows, total) — per-conversation token rollup.

Source code in src/symfonic/core/observability/metrics_store.py 
57
58
59
60
61
async def list_conversations(  # pragma: no cover - protocol
    self, tenant_id: str, limit: int, offset: int,
) -> tuple[list[dict[str, Any]], int]:
    """Return ``(rows, total)`` — per-conversation token rollup."""
    ...

summary async 

summary(tenant_id: str, start: date, end: date) -> dict[str, Any]

Return {conversations, total_llm_calls, total_cost_usd, total_tokens}.

Source code in src/symfonic/core/observability/metrics_store.py 
63
64
65
66
67
async def summary(  # pragma: no cover - protocol
    self, tenant_id: str, start: date, end: date,
) -> dict[str, Any]:
    """Return ``{conversations, total_llm_calls, total_cost_usd, total_tokens}``."""
    ...

token_usage async 

token_usage(tenant_id: str, start: date, end: date, group_by: str) -> list[dict[str, Any]]

Return DailyUsageRow / ModelUsageRow shaped dicts (group_by in {day, model}).

Source code in src/symfonic/core/observability/metrics_store.py 
69
70
71
72
73
async def token_usage(  # pragma: no cover - protocol
    self, tenant_id: str, start: date, end: date, group_by: str,
) -> list[dict[str, Any]]:
    """Return DailyUsageRow / ModelUsageRow shaped dicts (``group_by`` in {day, model})."""
    ...

MetricsSink

Bases: Protocol

Durable write side of a metrics backend.

aclose async 

aclose() -> None

Release driver resources.

Source code in src/symfonic/core/observability/metrics_store.py 
34
35
36
async def aclose(self) -> None:  # pragma: no cover - protocol
    """Release driver resources."""
    ...

flush async 

flush() -> None

Deterministically drain any internal buffer (tests / shutdown).

Source code in src/symfonic/core/observability/metrics_store.py 
30
31
32
async def flush(self) -> None:  # pragma: no cover - protocol
    """Deterministically drain any internal buffer (tests / shutdown)."""
    ...

prune async 

prune(cutoff: datetime) -> int

Delete rows older than cutoff; return the number removed.

Source code in src/symfonic/core/observability/metrics_store.py 
38
39
40
async def prune(self, cutoff: datetime) -> int:  # pragma: no cover
    """Delete rows older than ``cutoff``; return the number removed."""
    ...

record_calls async 

record_calls(rows: list[dict[str, Any]]) -> None

Persist a batch of snapshot-shaped per-call dicts.

Source code in src/symfonic/core/observability/metrics_store.py 
22
23
24
async def record_calls(self, rows: list[dict[str, Any]]) -> None:
    """Persist a batch of snapshot-shaped per-call dicts."""
    ...

record_events async 

record_events(rows: list[dict[str, Any]]) -> None

Persist metadata-only execution events.

Source code in src/symfonic/core/observability/metrics_store.py 
26
27
28
async def record_events(self, rows: list[dict[str, Any]]) -> None:
    """Persist metadata-only execution events."""
    ...

MetricsStore

Bases: MetricsSink, MetricsReader, Protocol

A full metrics backend: durable writes + rich tenant-scoped reads.

get_metrics_store 

get_metrics_store() -> BufferedMetricsSink | MetricsStore | None

Return the registered metrics store, or None (in-memory only).

Source code in src/symfonic/core/observability/metrics_store.py 
280
281
282
def get_metrics_store() -> BufferedMetricsSink | MetricsStore | None:
    """Return the registered metrics store, or ``None`` (in-memory only)."""
    return _metrics_store

make_metrics_store 

make_metrics_store(identifier: str | None, *, postgres_session_factory: Any = None, mongo_database_factory: Any = None, mongo_retention_days: int = 90) -> MetricsStore | None

Resolve a METRICS_STORE identifier to a concrete backend.

none / "" → None (in-memory collector only, today's default). postgres → :class:PostgresMetricsStore (needs postgres_session_factory + symfonic-core[postgres]). mongo → :class:MongoMetricsStore (needs mongo_database_factory + symfonic-core[mongodb]). A dotted path (my_pkg.mod:MyStore or my_pkg.mod.MyStore) → the adopter's zero-arg-constructible MetricsStore.

Raises ValueError with a pip install hint on a missing extra and on an unknown identifier, so misconfiguration fails loudly at startup.

Source code in src/symfonic/core/observability/metrics_store_factory.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def make_metrics_store(
    identifier: str | None,
    *,
    postgres_session_factory: Any = None,
    mongo_database_factory: Any = None,
    mongo_retention_days: int = 90,
) -> MetricsStore | None:
    """Resolve a ``METRICS_STORE`` identifier to a concrete backend.

    ``none`` / ``""`` → ``None`` (in-memory collector only, today's default).
    ``postgres`` → :class:`PostgresMetricsStore` (needs ``postgres_session_factory``
    + ``symfonic-core[postgres]``). ``mongo`` → :class:`MongoMetricsStore`
    (needs ``mongo_database_factory`` + ``symfonic-core[mongodb]``). A dotted
    path (``my_pkg.mod:MyStore`` or ``my_pkg.mod.MyStore``) → the adopter's
    zero-arg-constructible ``MetricsStore``.

    Raises ``ValueError`` with a ``pip install`` hint on a missing extra and
    on an unknown identifier, so misconfiguration fails loudly at startup.
    """
    raw = (identifier or "").strip()
    if raw == "" or raw.lower() == "none":
        return None
    low = raw.lower()

    if low == "postgres":
        if postgres_session_factory is None:
            raise ValueError(
                "METRICS_STORE=postgres requires a postgres_session_factory",
            )
        try:
            from symfonic.core.observability.postgres_metrics_store import (  # noqa: PLC0415
                PostgresMetricsStore,
            )
        except ImportError as exc:  # pragma: no cover - exercised via extra
            raise ValueError(
                "METRICS_STORE=postgres requires SQLAlchemy: "
                "pip install 'symfonic-core[postgres]'",
            ) from exc
        return PostgresMetricsStore(postgres_session_factory)

    if low in ("mongo", "mongodb"):
        if mongo_database_factory is None:
            raise ValueError(
                "METRICS_STORE=mongo requires a mongo_database_factory",
            )
        try:
            from symfonic.core.observability.mongo_metrics_store import (  # noqa: PLC0415
                MongoMetricsStore,
            )
        except ImportError as exc:  # pragma: no cover - exercised via extra
            raise ValueError(
                "METRICS_STORE=mongo requires motor: "
                "pip install 'symfonic-core[mongodb]'",
            ) from exc
        return MongoMetricsStore(
            mongo_database_factory, retention_days=mongo_retention_days,
        )

    if ":" in raw or "." in raw:
        cls = _import_dotted(raw)
        return cls()

    raise ValueError(
        f"unknown METRICS_STORE {raw!r}; expected 'none', 'postgres', "
        "'mongo', or a dotted path 'pkg.module:Class'",
    )

set_metrics_store 

set_metrics_store(store: BufferedMetricsSink | MetricsStore | None) -> None

Register the process-wide metrics store (or None to disable).

Source code in src/symfonic/core/observability/metrics_store.py 
274
275
276
277
def set_metrics_store(store: BufferedMetricsSink | MetricsStore | None) -> None:
    """Register the process-wide metrics store (or ``None`` to disable)."""
    global _metrics_store
    _metrics_store = store