symfonic.memory.layers.episodic_events

episodic_events

The episodic layer's event API: store, list, count, query.

Split out of :mod:symfonic.memory.layers.episodic (320 lines against the 300-line budget). What stayed behind is the BaseMemoryStore contract -- the five uniform methods every layer implements, plus prompt rendering. What moved here is the layer-specific domain API the file itself already separated under a Layer-specific domain methods heading: the four calls that speak in events and timestamps rather than in entries.

EpisodicLayer mixes this in, so every method is reached exactly as before.

EpisodicEventsMixin

Event storage and time-ordered reads for :class:EpisodicLayer.

The three attributes below are supplied by the host layer; they are declared here so the mixin's reads are typed rather than implicit.

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_events.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

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 adopter.

Source code in src/symfonic/memory/layers/episodic_events.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 adopter.
    """
    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_events.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

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_events.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,
        )