symfonic.capabilities.memory.candidates

candidates

A MemoryRetrievalPort seen as the CandidateSource the coordinator wants.

The two seams were built for different jobs and neither is wrong. A port returns a :class:RetrievalResult — ranked, capped, already selected. A candidate source returns :class:Candidate rows and lets :class:RetrievalCoordinator do the ranking, the gates and the scope blending.

Path A of #14 puts HydrationCoordinator in charge of composing the block, so the port has to arrive as a source, and this is that translation. It is deliberately a thin, honest one:

• the port has already ranked, so its order is preserved and this adds none;
• the port's score becomes the cue signal, because that is the signal the score answers — a similarity to the query — and putting it anywhere else would have the coordinator weigh it as something it is not;
• every other signal stays None. None means "this store cannot answer that question", which is exactly true of a port that never reported it, and is the distinction :class:RelevanceSignals exists to keep. Filling them with 0.0 would turn silence into an observation of absence.

PortCandidateSource

PortCandidateSource(port: MemoryRetrievalPort | CandidateScan)

Adapts one :class:MemoryRetrievalPort to the coordinator's source seam.

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

def __init__(self, port: MemoryRetrievalPort | CandidateScan) -> None:
    if not callable(getattr(port, "scan_candidates", None)):
        # A wiring error, raised where a human is reading it. The previous
        # version fell back to a widened public ``retrieve`` for a store
        # without this method, which put the cap back before the gates for
        # every store but one: the defect survived in the escape hatch and
        # nothing failed, because the fallback returned plausible results.
        raise MemoryContractError(
            f"{type(port).__name__} implements MemoryRetrievalPort but not "
            "CandidateScan, so it can only be asked for a capped result. "
            "The gates that decide what survives run after retrieval, so a "
            "cap applied here makes one rejection a lost slot. Implement "
            "scan_candidates -- everything admissible, uncapped by the "
            "query's limit and bounded by the store's own budget."
        )
    self._port: CandidateScan = port  # type: ignore[assignment]
    self._dropped: tuple[tuple[str, str], ...] = ()
    self._sources: Mapping[str, int] = {}
    self._unavailable: tuple[str, ...] = ()

drain_dropped

drain_dropped() -> tuple[tuple[str, str], ...]

What the last search refused, and clearing it as it is read.

The port reports provenance refusals in RetrievalResult.dropped; CandidateSource.search returns candidates and has nowhere to put them, so before this they died here. A scope whose every row was refused then looked exactly like a scope that remembered nothing — same empty block, same reason — and "this deployment's data cannot be attributed" is not "this user has no history".

Drained rather than accumulated because one source serves every turn of an agent's life: a ledger that grew would attribute one turn's refusals to the next, forever.

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

def drain_dropped(self) -> tuple[tuple[str, str], ...]:
    """What the last search refused, and clearing it as it is read.

    The port reports provenance refusals in ``RetrievalResult.dropped``;
    ``CandidateSource.search`` returns candidates and has nowhere to put
    them, so before this they died here. A scope whose every row was
    refused then looked exactly like a scope that remembered nothing — same
    empty block, same reason — and "this deployment's data cannot be
    attributed" is not "this user has no history".

    Drained rather than accumulated because one source serves every turn of
    an agent's life: a ledger that grew would attribute one turn's refusals
    to the next, forever.
    """
    drained, self._dropped = self._dropped, ()
    return drained

drain_sources

drain_sources() -> tuple[Mapping[str, int], tuple[str, ...]]

What each route contributed to the last search, and what was lost.

Drained for the same reason the refusals are: one source serves every turn of an agent's life, and a tally that accumulated would attribute one turn's recall to the next.

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

def drain_sources(self) -> tuple[Mapping[str, int], tuple[str, ...]]:
    """What each route contributed to the last search, and what was lost.

    Drained for the same reason the refusals are: one source serves every
    turn of an agent's life, and a tally that accumulated would attribute
    one turn's recall to the next.
    """
    counts, self._sources = self._sources, {}
    lost, self._unavailable = self._unavailable, ()
    return counts, lost

search async

search(query: MemoryQuery, layers: frozenset[MemoryLayer]) -> Sequence[Candidate]

Ask the port and hand over what it answered, unfiltered.

This used to drop rows whose layer the deployment had disabled, framed as a courtesy that saved the coordinator work. It was not a courtesy: a row filtered here leaves no entry in dropped, so a deployment that disabled a layer saw its memories vanish with no reason recorded -- indistinguishable from a scope that never held them.

The coordinator gates by layer and writes down what it refused, which is the behaviour its own docstring promises ("a source that answers outside the set is a bug this seam should surface, not one it should hide"). Surfacing it means letting it through to the component that keeps the ledger.

MemoryUnavailable is not caught. The coordinator and the hydration policy above it decide what a degraded store means for a turn; swallowing it here would hand them an empty result that reads as "nothing was remembered", which is the one thing an unreachable store must never look like.

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

async def search(
    self, query: MemoryQuery, layers: frozenset[MemoryLayer]
) -> Sequence[Candidate]:
    """Ask the port and hand over what it answered, unfiltered.

    This used to drop rows whose layer the deployment had disabled, framed
    as a courtesy that saved the coordinator work. It was not a courtesy: a
    row filtered here leaves no entry in ``dropped``, so a deployment that
    disabled a layer saw its memories vanish with no reason recorded --
    indistinguishable from a scope that never held them.

    The coordinator gates by layer *and writes down what it refused*, which
    is the behaviour its own docstring promises ("a source that answers
    outside the set is a bug this seam should surface, not one it should
    hide"). Surfacing it means letting it through to the component that
    keeps the ledger.

    ``MemoryUnavailable`` is **not** caught. The coordinator and the
    hydration policy above it decide what a degraded store means for a
    turn; swallowing it here would hand them an empty result that reads as
    "nothing was remembered", which is the one thing an unreachable store
    must never look like.
    """
    # Always the scan. Widening the public port's limit is still a cap
    # before the gates -- the same defect four times, with a bigger number
    # each time -- so there is no capped path left to fall back to. The
    # constructor refuses a store that cannot scan.
    result = await self._port.scan_candidates(query)
    self._dropped = result.dropped
    self._sources = getattr(result, "sources", {}) or {}
    self._unavailable = getattr(result, "unavailable", ()) or ()
    return tuple(
        Candidate(
            record=memory.record,
            signals=RelevanceSignals(cue=memory.score),
            # Legacy retrieval ranks before it returns, so its order is an
            # answer, not an accident of iteration. Saying so here is what
            # stops the coordinator from re-deciding it.
            pre_ranked=True,
        )
        for memory in result.memories
    )

overscan

overscan(limit: int) -> int

How many candidates to request so the gates have a choice.

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

def overscan(limit: int) -> int:
    """How many candidates to request so the gates have a choice."""
    if limit <= 0:
        return 0
    return max(OVERSCAN_MINIMUM, limit * OVERSCAN_FACTOR)