symfonic.memory.layers.episodic

episodic

EpisodicLayer -- narrative events and timestamped history.

Wraps a VectorBackend for event storage and similarity-based retrieval. Implements BaseMemoryStore with link as a no-op (episodes are vector-stored).

EpisodicLayer

EpisodicLayer(
    vector_backend: VectorBackend,
    embedding_provider: EmbeddingProvider,
    telemetry_sink: EpisodicTelemetrySink | None = None,
)

Episodic memory layer for narrative events and history.

BaseMemoryStore contract

• retrieve: queries events by vector similarity
• write: stores an event with its embedding
• summarize: returns empty string (delegates to compaction engine)
• link: no-op (episodes are vector-stored, not graph-linked)
• delete: removes an event from the vector store

Source code in src/symfonic/memory/layers/episodic.py

def __init__(
    self,
    vector_backend: VectorBackend,
    embedding_provider: EmbeddingProvider,
    telemetry_sink: EpisodicTelemetrySink | None = None,
) -> None:
    self._vector = vector_backend
    self._embedder = embedding_provider
    # When None (the default), the write path skips record construction
    # entirely -- zero cost on the hot path, which is the v6.1 contract.
    self._telemetry_sink = telemetry_sink
    # v7.22 T-7.22.8 -- emit-once-per-(tenant, sub_tenant) tracking for
    # MultiScopeIgnoredWarning.  See T-7.22.11 for the non-atomic
    # check-then-add concurrency rationale.
    self._multi_scope_warned: set[tuple[str, str | None]] = set()

count_events async

count_events(scope: TenantScope) -> int

Return the total number of stored episodic events for the tenant.

Delegates to VectorBackend.count() when available, otherwise falls back to 0 to avoid a full scan with an arbitrary embedding.

Source code in src/symfonic/memory/layers/episodic.py

async def count_events(self, scope: TenantScope) -> int:
    """Return the total number of stored episodic events for the tenant.

    Delegates to VectorBackend.count() when available, otherwise falls back
    to 0 to avoid a full scan with an arbitrary embedding.
    """
    if hasattr(self._vector, "count"):
        return await self._vector.count(scope)
    return 0

delete async

delete(scope: TenantScope, node_id: NodeId) -> None

Delete an episodic event by its ID.

Source code in src/symfonic/memory/layers/episodic.py

async def delete(self, scope: TenantScope, node_id: NodeId) -> None:
    """Delete an episodic event by its ID."""
    await self._vector.delete(scope, [str(node_id)])

link async

link(
    scope: TenantScope,
    source: NodeId,
    target: NodeId,
    relationship: str,
) -> None

No-op. Episodes are vector-stored, not graph-linked.

Source code in src/symfonic/memory/layers/episodic.py

async def link(
    self, scope: TenantScope, source: NodeId, target: NodeId, relationship: str
) -> None:
    """No-op. Episodes are vector-stored, not graph-linked."""
    return

list_events async

list_events(
    scope: TenantScope, limit: int = 1000
) -> list[MemoryEntry]

Return up to limit episodic events for the tenant.

Uses a neutral zero-vector query so that all stored events are returned in approximate descending-similarity order (deterministic for vectors with the same embedding, arbitrary but consistent otherwise). Prefer this over query_events when you need a full listing rather than a relevance-ranked subset.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
limit	int	Maximum number of events to return (default 1000).	1000

Returns:

Type	Description
list[MemoryEntry]	List of MemoryEntry instances (order: most-similar first).

Notes

v7.13.6: when the embedder rejects embed("") (the OpenAI embeddings endpoint returns 400 "input cannot be an empty string") we transparently fall back to embed(" "). The single-space probe is non-empty for OpenAI's validator while remaining distance-agnostic enough to enumerate stored events via similarity ranking. Sentence-transformers and other backends that accept empty strings continue to use the zero-string path. Fixes the v7.13.4 quick_nap phase-10 WARN reported by Jarvio.

Source code in src/symfonic/memory/layers/episodic.py

async def list_events(
    self, scope: TenantScope, limit: int = 1000
) -> list[MemoryEntry]:
    """Return up to *limit* episodic events for the tenant.

    Uses a neutral zero-vector query so that all stored events are
    returned in approximate descending-similarity order (deterministic
    for vectors with the same embedding, arbitrary but consistent
    otherwise).  Prefer this over ``query_events`` when you need a
    full listing rather than a relevance-ranked subset.

    Args:
        scope: Tenant scope for isolation.
        limit: Maximum number of events to return (default 1000).

    Returns:
        List of MemoryEntry instances (order: most-similar first).

    Notes:
        v7.13.6: when the embedder rejects ``embed("")`` (the OpenAI
        embeddings endpoint returns 400 ``"input cannot be an empty
        string"``) we transparently fall back to ``embed(" ")``.  The
        single-space probe is non-empty for OpenAI's validator while
        remaining distance-agnostic enough to enumerate stored events
        via similarity ranking.  Sentence-transformers and other
        backends that accept empty strings continue to use the
        zero-string path.  Fixes the v7.13.4 quick_nap phase-10 WARN
        reported by Jarvio.
    """
    count = await self.count_events(scope)
    if count == 0:
        return []
    top_k = min(count, limit)
    # Use a zero-vector query so results are distance-agnostic on
    # identical embeddings (stub backends) and approximately uniform
    # across real backends.  Cosine similarity of a zero vector is
    # undefined (0/0); real backends handle this gracefully by
    # returning 0.0 similarity, which still enumerates all documents.
    # v7.13.6: OpenAI embeddings rejects ``""`` with HTTP 400; retry
    # once with a single-space probe so list_events is safe against
    # any backend regardless of empty-input policy.
    try:
        zero_vec = await self._embedder.embed("")
    except Exception:
        zero_vec = await self._embedder.embed(" ")
    results = await self._vector.search(scope, zero_vec, top_k)
    entries: list[MemoryEntry] = []
    for result in results:
        entries.append(
            MemoryEntry(
                id=result["id"],
                layer=MemoryLayer.EPISODIC,
                tenant_id=scope.tenant_id,
                content=result.get("document", ""),
                score=result.get("score"),
                metadata=result.get("metadata", {}),
            )
        )
    return entries

query_events async

query_events(
    scope: TenantScope, query: str, top_k: int = 5
) -> list[MemoryEntry]

Query episodic events by vector similarity.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
query	str	Natural language query to search against. Must be a non-empty, non-whitespace string.	required
top_k	int	Maximum number of results to return.	5

Returns:

Type	Description
list[MemoryEntry]	List of MemoryEntry instances ranked by similarity.

Raises:

Type	Description
ValueError	If query is empty or whitespace-only, or if the embedder returns an empty vector.

Source code in src/symfonic/memory/layers/episodic.py

async def query_events(
    self, scope: TenantScope, query: str, top_k: int = 5
) -> list[MemoryEntry]:
    """Query episodic events by vector similarity.

    Args:
        scope: Tenant scope for isolation.
        query: Natural language query to search against.  Must be a
            non-empty, non-whitespace string.
        top_k: Maximum number of results to return.

    Returns:
        List of MemoryEntry instances ranked by similarity.

    Raises:
        ValueError: If ``query`` is empty or whitespace-only, or if the
            embedder returns an empty vector.
    """
    if not query.strip():
        raise ValueError("query_events requires a non-empty query string")
    query_embedding = await self._embedder.embed(query)
    if not query_embedding:
        raise ValueError("embedder returned an empty vector for query")
    results = await self._vector.search(scope, query_embedding, top_k)
    entries: list[MemoryEntry] = []
    for result in results:
        entries.append(
            MemoryEntry(
                id=result["id"],
                layer=MemoryLayer.EPISODIC,
                tenant_id=scope.tenant_id,
                content=result.get("document", ""),
                score=result.get("score"),
                metadata=result.get("metadata", {}),
            )
        )
    return entries

render_for_prompt

render_for_prompt(
    entry: Any,
    scrubber: Callable[[dict[str, Any]], dict[str, Any]]
    | None = None,
) -> str

Render an episodic entry for MEMORY_CONTEXT.

v7.7: when metadata['speaker'] is present (per-message row introduced in Phase B), emit [episodic] <speaker>: <content> so the LLM sees a clean speaker-prefixed line. Pre-v7.7 rows (no speaker) or rows tagged speaker == 'legacy' fall back to :func:render_legacy for byte-identical pre-v7.7 output and prompt-cache parity.

Source code in src/symfonic/memory/layers/episodic.py

def render_for_prompt(
    self,
    entry: Any,
    scrubber: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
) -> str:
    """Render an episodic entry for ``MEMORY_CONTEXT``.

    v7.7: when ``metadata['speaker']`` is present (per-message row
    introduced in Phase B), emit ``[episodic] <speaker>: <content>``
    so the LLM sees a clean speaker-prefixed line.  Pre-v7.7 rows
    (no speaker) or rows tagged ``speaker == 'legacy'`` fall back
    to :func:`render_legacy` for byte-identical pre-v7.7 output
    and prompt-cache parity.
    """
    metadata = getattr(entry, "metadata", {}) or {}
    speaker = metadata.get("speaker")
    if isinstance(speaker, str) and speaker and speaker != "legacy":
        content = str(getattr(entry, "content", ""))
        layer_name = getattr(entry, "layer", "episodic")
        return f"[{layer_name}] {speaker}: {content}"
    return render_legacy(entry, scrubber)

retrieve async

retrieve(
    scope: TenantScope, query: str, top_k: int = 5
) -> list[MemoryEntry]

Retrieve episodic events matching the query.

v7.22 T-7.22.8: emits :class:MultiScopeIgnoredWarning on first encounter of a scope with inherits_from. Episodic events are speaker-attributed timestamped records bound to a single (tenant, sub_tenant) pair; merging inherited scopes would corrupt per-tenant cost accounting and break GDPR scope guarantees (v5.5.0 Gate 3).

Source code in src/symfonic/memory/layers/episodic.py

async def retrieve(
    self, scope: TenantScope, query: str, top_k: int = 5
) -> list[MemoryEntry]:
    """Retrieve episodic events matching the query.

    v7.22 T-7.22.8: emits :class:`MultiScopeIgnoredWarning` on first
    encounter of a scope with ``inherits_from``.  Episodic events are
    speaker-attributed timestamped records bound to a single
    ``(tenant, sub_tenant)`` pair; merging inherited scopes would
    corrupt per-tenant cost accounting and break GDPR scope
    guarantees (v5.5.0 Gate 3).
    """
    from symfonic.memory._multi_scope_warn import maybe_warn_multi_scope

    maybe_warn_multi_scope(self, self._multi_scope_warned, scope)
    return await self.query_events(scope, query, top_k)

store_event async

store_event(
    scope: TenantScope,
    event: MemoryEntry,
    *,
    source_phase: str = "episodic.store_event",
) -> None

Store an event with its embedding in the vector backend.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
event	MemoryEntry	The memory entry representing the event.	required
source_phase	str	Tag identifying which caller triggered the write (e.g. "engine.consolidate", "phase10"). Recorded on the telemetry record when a sink is configured; ignored otherwise. Default identifies a direct call to this method.	'episodic.store_event'

v6.2 T05 telemetry: when the caller populates event.metadata with any of tokens_in / tokens_out / input_tokens / output_tokens / cost_usd / model / embedding_latency_ms, the values are lifted onto the :class:EpisodicTelemetryRecord for the corpus. Keys missing from metadata remain None and are stripped from the JSONL line (v6.1 shape preserved byte-for-byte). The engine can source the token / cost fields from LLMPreCallEvent / LLMEndEvent and attach them to the metadata before calling store_event; direct callers that do not consume LangChain callbacks may pass the values in metadata themselves.

Source code in src/symfonic/memory/layers/episodic.py

async def store_event(
    self,
    scope: TenantScope,
    event: MemoryEntry,
    *,
    source_phase: str = "episodic.store_event",
) -> None:
    """Store an event with its embedding in the vector backend.

    Args:
        scope: Tenant scope for isolation.
        event: The memory entry representing the event.
        source_phase: Tag identifying which caller triggered the
            write (e.g. ``"engine.consolidate"``, ``"phase10"``).
            Recorded on the telemetry record when a sink is
            configured; ignored otherwise. Default identifies a
            direct call to this method.

    v6.2 T05 telemetry: when the caller populates ``event.metadata``
    with any of ``tokens_in`` / ``tokens_out`` / ``input_tokens`` /
    ``output_tokens`` / ``cost_usd`` / ``model`` /
    ``embedding_latency_ms``, the values are lifted onto the
    :class:`EpisodicTelemetryRecord` for the corpus. Keys missing
    from metadata remain ``None`` and are stripped from the JSONL
    line (v6.1 shape preserved byte-for-byte). The engine can source
    the token / cost fields from ``LLMPreCallEvent`` /
    ``LLMEndEvent`` and attach them to the metadata before calling
    store_event; direct callers that do not consume LangChain
    callbacks may pass the values in ``metadata`` themselves.
    """
    embedding = await self._embedder.embed(event.content)
    doc_id = event.id or str(uuid.uuid4())
    await self._vector.add(
        scope,
        ids=[doc_id],
        embeddings=[embedding],
        metadatas=[{
            "layer": MemoryLayer.EPISODIC.value,
            "importance": event.importance,
            "created_at": event.created_at.isoformat(),
            **event.metadata,
        }],
        documents=[event.content],
    )

    # Telemetry emit -- gated on sink presence so the default path
    # does zero work.  A raising sink must never break the write.
    sink = self._telemetry_sink
    if sink is None:
        return
    try:
        md = event.metadata or {}
        action_raw = md.get("action_type")
        action_type = str(action_raw) if action_raw is not None else None

        # v6.2 T05: lift optional token / cost / model fields out
        # of ``event.metadata`` into first-class record fields.
        # Metadata keys are advisory -- callers that do not source
        # them yield a record with the v6.1 shape.
        def _as_int(v: object) -> int | None:
            try:
                return int(v) if v is not None else None
            except (TypeError, ValueError):
                return None

        def _as_float(v: object) -> float | None:
            try:
                return float(v) if v is not None else None
            except (TypeError, ValueError):
                return None

        # Pydantic TenantScope always has `namespace` as a field
        # (defaulting to None); getattr's fallback never fires. Coalesce
        # explicitly so the corpus always has a non-null partitioning
        # key, which the v6.2 Phase 12 extractor groups on.
        record = build_record(
            tenant_id=scope.tenant_id,
            scope=scope.namespace or "default",
            episode_id=doc_id,
            node_count=1,
            source_phase=source_phase,
            action_type=action_type,
            tokens_in=_as_int(md.get("tokens_in")),
            tokens_out=_as_int(md.get("tokens_out")),
            input_tokens=_as_int(md.get("input_tokens")),
            output_tokens=_as_int(md.get("output_tokens")),
            cost_usd=_as_float(md.get("cost_usd")),
            model=(
                str(md["model"])
                if md.get("model") is not None
                else None
            ),
            embedding_latency_ms=_as_float(md.get("embedding_latency_ms")),
        )
        await sink.emit(record)
    except Exception:
        _telemetry_logger.debug(
            "episodic telemetry emit failed for %s", doc_id, exc_info=True,
        )

summarize async

summarize(
    scope: TenantScope, entries: list[MemoryEntry]
) -> str

Not directly implemented. Delegates to compaction engine.

Source code in src/symfonic/memory/layers/episodic.py

async def summarize(self, scope: TenantScope, entries: list[MemoryEntry]) -> str:
    """Not directly implemented. Delegates to compaction engine."""
    return ""

write async

write(scope: TenantScope, entry: MemoryEntry) -> None

Write an episodic event to the vector store.

Source code in src/symfonic/memory/layers/episodic.py

async def write(self, scope: TenantScope, entry: MemoryEntry) -> None:
    """Write an episodic event to the vector store."""
    await self.store_event(scope, entry)