symfonic.capabilities.memory.coordinator

coordinator

The component that decides what a turn remembers.

Everything the facade did inline between a store call and a prompt string lives here: which layers are searched, what each hit is worth, which hits are too weak to earn a slot, how far up the hierarchy a memory may have come from, and which neighbours come along with it. The facade could not be described without describing all of it; this class can be handed a different store and keep every one of those decisions unchanged, which is the whole point of the extraction.

The coordinator is a :class:~.ports.MemoryRetrievalPort — the bridge holds it and knows nothing else. :meth:RetrievalCoordinator.coordinate is the same pass with its work shown, for the hydration layer and for anything that wants the activation provenance. Nothing about a turn is stored on the instance: the log rides the return value, so two concurrent invocations cannot read each other's recall.

Transport degrades, contracts propagate — and the two halves differ. An unreachable store raises :class:~.errors.MemoryUnavailable out of :meth:retrieve, because the port's contract says an empty result means "this scope remembers nothing" and the bridge is the component licensed to collapse the two. An unreachable association source does not: by the time expansion runs the recall is in hand, and failing the turn to protect its neighbours trades the answer for the garnish.

CoordinatedRetrieval dataclass

CoordinatedRetrieval(result: RetrievalResult, activation: ActivationLog = ActivationLog(), layers: frozenset[MemoryLayer] = frozenset())

One retrieval pass with its work shown.

RetrievalCoordinator

RetrievalCoordinator(*, source: CandidateSource, policy: RetrievalPolicy | None = None, activation: SpreadingActivation | None = None)

Owns layers, scoring, gates, scope blending, and activation.

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

def __init__(
    self,
    *,
    source: CandidateSource,
    policy: RetrievalPolicy | None = None,
    activation: SpreadingActivation | None = None,
) -> None:
    self._source = source
    self._policy = policy or RetrievalPolicy()
    self._activation = activation

coordinate async

coordinate(query: MemoryQuery) -> CoordinatedRetrieval

Retrieve for query, returning the result and its provenance.

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

async def coordinate(self, query: MemoryQuery) -> CoordinatedRetrieval:
    """Retrieve for ``query``, returning the result and its provenance."""
    layers = self._policy.effective_layers(query)
    if not layers:
        return CoordinatedRetrieval(
            result=RetrievalResult(dropped=((NO_RECORD, _no_layers(query)),)),
            layers=layers,
        )

    candidates = await self._source.search(query, layers)
    # A source may refuse rows before this coordinator ever sees them --
    # an adapter over legacy storage drops what it cannot attribute to a
    # scope. Those refusals belong in the same ledger as the ones the gate
    # makes, or a refused row is indistinguishable from a scope that
    # remembers nothing. Optional because ``CandidateSource`` does not
    # require it: a source with nothing to report simply has no drain.
    drain = getattr(self._source, "drain_dropped", None)
    refused = tuple(drain()) if callable(drain) else ()
    # The same shape for the route tally: optional, because
    # ``CandidateSource`` does not require it, and drained rather than read
    # so one turn's recall is never attributed to the next.
    tally = getattr(self._source, "drain_sources", None)
    sources, unavailable = tally() if callable(tally) else ({}, ())
    kept, dropped = self._gate(tuple(candidates), query, layers)
    found, log = await self._expand(query.scope, kept)

    ranked = select((*kept, *found), query)
    return CoordinatedRetrieval(
        result=RetrievalResult(
            memories=ranked.memories,
            dropped=(*refused, *dropped, *ranked.dropped),
            sources=sources,
            unavailable=unavailable,
        ),
        activation=log,
        layers=layers,
    )

retrieve async

retrieve(query: MemoryQuery) -> RetrievalResult

The :class:~.ports.MemoryRetrievalPort surface: just the result.

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

async def retrieve(self, query: MemoryQuery) -> RetrievalResult:
    """The :class:`~.ports.MemoryRetrievalPort` surface: just the result."""
    return (await self.coordinate(query)).result

RetrievalPolicy dataclass

RetrievalPolicy(enabled_layers: frozenset[MemoryLayer] = RETRIEVABLE_LAYERS, weights: RelevanceWeights = DEFAULT_WEIGHTS, blend: ScopeBlend = ScopeBlend(), min_relevance: float = 0.0, min_salience: float = 0.0, profile_slots: int = 1)

The deployment's retrieval decisions, in one value.

Frozen and validated at construction so a deployment that mis-states a gate fails at wiring time rather than by quietly retrieving nothing on every turn — the failure mode a floor of 1.5 actually has.

effective_layers

effective_layers(query: MemoryQuery) -> frozenset[MemoryLayer]

The layers actually searched: what is enabled, narrowed by the query.

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

def effective_layers(self, query: MemoryQuery) -> frozenset[MemoryLayer]:
    """The layers actually searched: what is enabled, narrowed by the query."""
    return self.enabled_layers & query.layers