symfonic.memory.telemetry

telemetry

Telemetry sinks for memory-layer write observability.

The episodic telemetry sink (see :mod:episodic_sink) captures a structured record for every successful episodic write so downstream corpus-evaluation and v6.2 extractor work has ground truth to train / evaluate against.

The sink API is intentionally minimal: construct a sink via :func:resolve_episodic_sink, call await sink.emit(record), that is the whole contract. Sinks are fire-and-forget from the caller's perspective; the default None path is a no-op and adds zero cost.

EpisodicTelemetryRecord dataclass

EpisodicTelemetryRecord(
    timestamp: str,
    tenant_id: str,
    scope: str,
    episode_id: str,
    node_count: int,
    source_phase: str,
    action_type: str | None = None,
    extras: dict[str, object] = dict(),
    tokens_in: int | None = None,
    tokens_out: int | None = None,
    input_tokens: int | None = None,
    output_tokens: int | None = None,
    cost_usd: float | None = None,
    model: str | None = None,
    embedding_latency_ms: float | None = None,
)

Structured telemetry record for a single successful episodic write.

The field set is the minimum viable shape. v6.2 (T05) added six optional fields relevant to the LLM-extractor corpus: tokens_in, tokens_out, input_tokens, output_tokens, cost_usd, model, and embedding_latency_ms. All are None by default so the existing JSONL shape is unchanged for records that omit them; the to_dict / to_jsonl serialisers lift populated values into the output and drop fields that remain None to keep the corpus lean.

Do NOT remove or rename fields without a major-version bump — downstream consumers (the eventual LLM extractor corpus) depend on the shape.

to_dict

to_dict() -> dict[str, object]

Return a JSON-serialisable dict representation.

Fields that remain None (i.e. the engine did not source the signal for this write) are dropped so the JSONL stream keeps the v6.1 shape for records that carry none of the v6.2 fields.

Source code in src/symfonic/memory/telemetry/episodic_sink.py

def to_dict(self) -> dict[str, object]:
    """Return a JSON-serialisable dict representation.

    Fields that remain ``None`` (i.e. the engine did not source the
    signal for this write) are dropped so the JSONL stream keeps
    the v6.1 shape for records that carry none of the v6.2 fields.
    """
    data = asdict(self)
    # v6.2 T05: drop None-valued additive fields so v6.1 shape is
    # preserved byte-for-byte when the engine doesn't populate them.
    for key in (
        "tokens_in", "tokens_out", "input_tokens", "output_tokens",
        "cost_usd", "model", "embedding_latency_ms",
    ):
        if data.get(key) is None:
            data.pop(key, None)
    return data

to_jsonl

to_jsonl() -> str

Return a newline-terminated JSON line ready for appending.

Source code in src/symfonic/memory/telemetry/episodic_sink.py

def to_jsonl(self) -> str:
    """Return a newline-terminated JSON line ready for appending."""
    return json.dumps(self.to_dict(), default=str, sort_keys=True) + "\n"

EpisodicTelemetrySink

Bases: Protocol

Protocol implemented by all episodic telemetry sinks.

emit async

emit(record: EpisodicTelemetryRecord) -> None

Persist / forward a telemetry record.

Source code in src/symfonic/memory/telemetry/episodic_sink.py

async def emit(self, record: EpisodicTelemetryRecord) -> None:
    """Persist / forward a telemetry record."""
    ...

JsonlEpisodicSink

JsonlEpisodicSink(path: str | Path)

Append records to a JSON-lines file at the configured path.

• Path is created lazily on first emit.
• Writes are append-only; no in-process buffering (the OS buffer is sufficient for the expected traffic shape — a single write per successful episodic entry).
• File-handle management is per-call: we reopen on each emit. This is intentional: keeps the sink reentrant across event loops and avoids stale handles after log rotation.

Source code in src/symfonic/memory/telemetry/episodic_sink.py

def __init__(self, path: str | Path) -> None:
    self._path = Path(path)

LoggingEpisodicSink

LoggingEpisodicSink(
    logger_name: str = "symfonic.memory.telemetry.episodic",
)

Emit records to the stdlib logging framework at INFO level.

The log record carries the structured dict in extra so a downstream JSON log handler can lift the fields without parsing the formatted message. Intended for deployments that already pipe python logging into a structured log collector (Loki, Datadog).

Level requirement (important)

The record is emitted at INFO, but the named logger "symfonic.memory.telemetry.episodic" inherits its effective level from the root logger, which defaults to WARNING in production. Without caller configuration, INFO records are filtered BEFORE reaching any handler and the sink appears silent -- a customer setting FrameworkConfig.episodic_telemetry_sink="logger" and seeing no output is almost always a level misconfiguration, not a broken sink.

Add this once in your application's logging config::

import logging
logging.getLogger("symfonic.memory.telemetry.episodic").setLevel(
    logging.INFO,
)

This sink intentionally does NOT auto-install a handler or raise the logger level -- that would be too magic for a Protocol sink shared by every EpisodicLayer.store_event call.

Source code in src/symfonic/memory/telemetry/episodic_sink.py

def __init__(self, logger_name: str = "symfonic.memory.telemetry.episodic") -> None:
    self._logger = logging.getLogger(logger_name)

NullEpisodicSink

No-op sink. Used when telemetry is disabled.

EpisodicLayer branches on sink is None BEFORE constructing the record, so this class is only used as an explicit marker for tests and type contracts — the hot path does not instantiate it.

resolve_episodic_sink

resolve_episodic_sink(
    config_value: str | None,
) -> EpisodicTelemetrySink | None

Turn a FrameworkConfig.episodic_telemetry_sink value into a sink.

Mapping:

• None or empty string -> None (sink disabled, zero cost).
• "logger" -> :class:LoggingEpisodicSink.
• Anything else -> :class:JsonlEpisodicSink with the value treated as a filesystem path.

Source code in src/symfonic/memory/telemetry/episodic_sink.py

def resolve_episodic_sink(
    config_value: str | None,
) -> EpisodicTelemetrySink | None:
    """Turn a ``FrameworkConfig.episodic_telemetry_sink`` value into a sink.

    Mapping:

    - ``None`` or empty string -> ``None`` (sink disabled, zero cost).
    - ``"logger"`` -> :class:`LoggingEpisodicSink`.
    - Anything else -> :class:`JsonlEpisodicSink` with the value treated
      as a filesystem path.
    """
    if not config_value:
        return None
    if config_value == LOGGER_SINK_TOKEN:
        return LoggingEpisodicSink()
    return JsonlEpisodicSink(config_value)