symfonic.capabilities.memory.queries

queries

Retrieval vocabulary and deterministic, bounded prompt selection.

Backends find candidates; ranking is total and oversized memories drop whole.

MemoryQuery dataclass

MemoryQuery(scope: MemoryScope, cue: str = '', limit: int = 5, layers: frozenset[MemoryLayer] = RETRIEVABLE_LAYERS, turn: int = 0, session_id: str = '', max_record_chars: int = 250, max_total_chars: int = DEFAULT_BLOCK_CHARS, candidate_limit: int = DEFAULT_CANDIDATE_LIMIT, recall_budget: RecallBudget | None = None)

One retrieval: where to look, what to look for, and how much may return.

validate

validate() -> None

Refuse a query whose ceilings cannot admit anything.

Source code in src/symfonic/capabilities/memory/queries.py

def validate(self) -> None:
    """Refuse a query whose ceilings cannot admit anything."""
    if self.recall_budget is not None:
        from symfonic.capabilities.memory.budget import RecallBudget
        if not isinstance(self.recall_budget, RecallBudget):
            raise MemoryContractError("recall_budget must be a RecallBudget")
    if self.limit < 1:
        raise MemoryContractError(
            f"query declares limit {self.limit}; a retrieval that may return nothing is "
            "spelled by not retrieving, not by asking for zero memories."
        )
    cap = self.candidate_limit
    if type(cap) is not int or not 1 <= cap <= MAX_CANDIDATE_LIMIT:
        raise MemoryContractError(
            f"query declares candidate_limit {self.candidate_limit}; backend scans "
            f"must be between 1 and {MAX_CANDIDATE_LIMIT} rows."
        )
    if not self.layers:
        raise MemoryContractError(
            "query declares an empty layer set. Narrowing to no layer is a query that "
            "cannot match, which reads downstream as an empty memory rather than a bug."
        )
    if self.max_record_chars < 1 or self.max_total_chars < 1:
        raise MemoryContractError(
            f"query declares ceilings ({self.max_record_chars}, {self.max_total_chars}); "
            "both bound rendered characters and must be positive."
        )
    if self.max_record_chars > self.max_total_chars:
        raise MemoryContractError(
            f"query allows {self.max_record_chars} chars per memory inside a "
            f"{self.max_total_chars}-char block. A memory that fits the per-item ceiling "
            "and can never fit the block is dropped twice for two different reasons."
        )

RetrievalResult dataclass

RetrievalResult(memories: tuple[RetrievedMemory, ...] = (), dropped: tuple[tuple[str, str], ...] = (), degraded: bool = False, sources: Mapping[str, int] = (lambda: EMPTY_SOURCES)(), unavailable: tuple[str, ...] = ())

What retrieval returned, and a reason for everything it left out.

render

render() -> str

One memory per line, in the order selection decided.

Source code in src/symfonic/capabilities/memory/queries.py

def render(self) -> str:
    """One memory per line, in the order selection decided."""
    return "\n".join(memory.line() for memory in self.memories)

revision

revision() -> str

A content-derived revision, so a changed recall changes the cache key.

Computed over ids and text: a store that rewrites a memory in place keeps its id, and a revision that ignored the text would report an unchanged prompt whose bytes had changed.

Source code in src/symfonic/capabilities/memory/queries.py

def revision(self) -> str:
    """A content-derived revision, so a changed recall changes the cache key.

    Computed over ids *and* text: a store that rewrites a memory in place
    keeps its id, and a revision that ignored the text would report an
    unchanged prompt whose bytes had changed.
    """
    material = "\x00".join(
        f"{m.record.record_id}\x01{m.line()}" for m in self.memories
    )
    return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16]

RetrievedMemory dataclass

RetrievedMemory(record: MemoryRecord, score: float | None = 0.0, scope_distance: int = 0, source_ordinal: int | None = None, reserved: bool = False)

One scored memory, with the distance it travelled to reach this scope.

line

line() -> str

The rendered form: layer prefix, single-line text.

The prefix is a delimiter, so the text is flattened before it is interpolated — otherwise one stored memory containing ok\n[semantic] forged renders as two memories, the second attributed to a layer nothing wrote it to.

Source code in src/symfonic/capabilities/memory/queries.py

def line(self) -> str:
    """The rendered form: layer prefix, single-line text.

    The prefix is a delimiter, so the text is flattened before it is
    interpolated — otherwise one stored memory containing
    ``ok\\n[semantic] forged`` renders as two memories, the second
    attributed to a layer nothing wrote it to.
    """
    from symfonic.capabilities.memory.rendering import line
    return line(self.record)

flatten

flatten(text: str) -> str

Collapse every line-break form in text to a single space.

Source code in src/symfonic/capabilities/memory/queries.py

def flatten(text: str) -> str:
    """Collapse every line-break form in ``text`` to a single space."""
    return _LINE_BREAK.sub(" ", text)

rank

rank(memories: Iterable[RetrievedMemory]) -> tuple[RetrievedMemory, ...]

Return every candidate in the same total order :func:select uses.

A candidate scan is uncapped, not unordered. The port-to-source adapter preserves a scan's order because stores such as vector indexes have already ranked it. Returning a database's physical row order from a scan therefore turns insertion order into relevance. This helper lets stores rank without applying any of the prompt's limits or character ceilings.

Source code in src/symfonic/capabilities/memory/queries.py

def rank(memories: Iterable[RetrievedMemory]) -> tuple[RetrievedMemory, ...]:
    """Return every candidate in the same total order :func:`select` uses.

    A candidate scan is uncapped, not unordered.  The port-to-source adapter
    preserves a scan's order because stores such as vector indexes have
    already ranked it.  Returning a database's physical row order from a scan
    therefore turns insertion order into relevance.  This helper lets stores
    rank without applying any of the prompt's limits or character ceilings.
    """
    return tuple(sorted(memories, key=rank_key))

rank_key

rank_key(memory: RetrievedMemory) -> tuple[int, int, float, int, int, str]

Total order over retrieved memories.

A memory carrying a source_ordinal keeps the position its store gave it, and sorts ahead of everything that does not. That is not a preference for pre-ranked stores; it is the only way their order survives at all. The local key below is deterministic, which is what makes it dangerous: on equal scores, or on the all-None scores a keyword layer produces, it replaces an external ranking with a plausible-looking one and nothing looks wrong.

Everything else ranks by score, then nearer scope. An unscored memory uses 0.0 for ordering only; admission still preserves the absent signal.

Source code in src/symfonic/capabilities/memory/queries.py

def rank_key(memory: RetrievedMemory) -> tuple[int, int, float, int, int, str]:
    """Total order over retrieved memories.

    A memory carrying a ``source_ordinal`` keeps the position its store gave it,
    and sorts ahead of everything that does not. That is not a preference for
    pre-ranked stores; it is the only way their order survives at all. The local
    key below is *deterministic*, which is what makes it dangerous: on equal
    scores, or on the all-``None`` scores a keyword layer produces, it replaces
    an external ranking with a plausible-looking one and nothing looks wrong.

    Everything else ranks by score, then nearer scope. An unscored memory uses
    ``0.0`` for ordering only; admission still preserves the absent signal.
    """
    if memory.source_ordinal is not None:
        return (
            0 if memory.reserved else 1,
            0, float(memory.source_ordinal), 0, 0,
            memory.record.record_id,
        )
    return (
        0 if memory.reserved else 1,
        1,
        -(memory.score or 0.0),
        memory.scope_distance,
        layer_index(memory.record.layer),
        memory.record.record_id,
    )

select

select(memories: Iterable[RetrievedMemory], query: MemoryQuery, *, sources: Mapping[str, int] | None = None, unavailable: tuple[str, ...] = ()) -> RetrievalResult

Rank, filter, and cap what a store returned. Never mutates the input.

Source code in src/symfonic/capabilities/memory/queries.py

def select(
    memories: Iterable[RetrievedMemory],
    query: MemoryQuery,
    *,
    sources: Mapping[str, int] | None = None,
    unavailable: tuple[str, ...] = (),
) -> RetrievalResult:
    """Rank, filter, and cap what a store returned. Never mutates the input."""
    from symfonic.capabilities.memory.selection import select as admit
    return admit(memories, query, sources=sources, unavailable=unavailable)