symfonic.memory.telemetry.episodic_sink

episodic_sink

Episodic-write telemetry sink.

Captures a structured record for every successful episodic write so the future LLM-based Phase 12 pattern extractor (deferred to v6.2) has a real corpus to evaluate against.

Design constraints (see docs/releases/v6.1-plan.md T01):

1. Zero cost when disabled. The default None sink path does not serialise, does not await, does not allocate.
2. Serialisation happens off the caller's critical path. When the sink is enabled we still build the record synchronously (it is a handful of primitive fields), but the actual write (logger call, file append) is quick and idempotent.
3. Additive only. Adding fields to :class:EpisodicTelemetryRecord in v6.2 must not break existing consumers reading JSONL output.
4. Purely best-effort. A failing sink MUST NOT break the episodic write — failures log at DEBUG level and move on.

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.

build_record

build_record(
    *,
    tenant_id: str,
    scope: str,
    episode_id: str,
    source_phase: str,
    node_count: int = 1,
    action_type: str | None = None,
    extras: dict[str, object] | None = None,
    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,
) -> EpisodicTelemetryRecord

Construct a record with an ISO-8601 UTC timestamp.

Factored out of :class:EpisodicLayer so tests can build records directly without the layer stack.

v6.2 T05 exposes optional token / cost / embedding-latency fields. See :class:EpisodicTelemetryRecord for the field semantics.

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

def build_record(
    *,
    tenant_id: str,
    scope: str,
    episode_id: str,
    source_phase: str,
    node_count: int = 1,
    action_type: str | None = None,
    extras: dict[str, object] | None = None,
    # v6.2 T05 additive kwargs. All default None -- callers that do
    # not source the signal produce a record with the v6.1 shape.
    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,
) -> EpisodicTelemetryRecord:
    """Construct a record with an ISO-8601 UTC timestamp.

    Factored out of :class:`EpisodicLayer` so tests can build records
    directly without the layer stack.

    v6.2 T05 exposes optional token / cost / embedding-latency fields.
    See :class:`EpisodicTelemetryRecord` for the field semantics.
    """
    return EpisodicTelemetryRecord(
        timestamp=datetime.now(UTC).isoformat(),
        tenant_id=tenant_id,
        scope=scope,
        episode_id=episode_id,
        node_count=node_count,
        source_phase=source_phase,
        action_type=action_type,
        extras=dict(extras or {}),
        tokens_in=tokens_in,
        tokens_out=tokens_out,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        cost_usd=cost_usd,
        model=model,
        embedding_latency_ms=embedding_latency_ms,
    )

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)