symfonic.capabilities.memory.coordinator

coordinator

The component that scores, gates, ranks, and expands a turn's recall.

It is a :class:~.ports.MemoryRetrievalPort; coordinate returns the pass with provenance for hydration. Optional local capture is single-use. The store can be substituted without changing decisions. Ordinary turn state stays on the result, so concurrent turns cannot read each other's recall.

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, exclude_speakers: frozenset[str] = frozenset(), capture: LocalCandidateCapture | 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,
    exclude_speakers: frozenset[str] = frozenset(),
    capture: LocalCandidateCapture | None = None,
) -> None:
    self._source = source
    self._policy = policy or RetrievalPolicy()
    self._activation = activation
    self._exclude_speakers = frozenset(exclude_speakers)
    self._capture = capture

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 = tuple(await self._source.search(query, layers))
    # Preserve source refusals alongside gate refusals.
    drain = getattr(self._source, "drain_dropped", None)
    refused = tuple(drain()) if callable(drain) else ()
    # Drain route counts so one turn cannot attribute them to the next.
    tally = getattr(self._source, "drain_sources", None)
    sources, unavailable = tally() if callable(tally) else ({}, ())
    observed: list[tuple[str, float | None, str]] | None = [] if self._capture else None
    kept, dropped = self._gate(candidates, query, layers, observed)
    from symfonic.capabilities.memory.retrieval_selection import select_recall
    ranked, found, log, linked, excluded = await select_recall(self, query, kept)
    if self._capture is not None:
        try:
            self._capture.observe(query, candidates, kept, ranked.memories,
                                  observed, found, log.nodes, linked=linked)
        except Exception:
            # Capture is diagnostic and cannot fail a retrieval turn.
            self._capture.invalidate()
    return CoordinatedRetrieval(
        result=RetrievalResult(
            memories=ranked.memories,
            dropped=(*refused, *dropped, *excluded, *ranked.dropped),
            sources=sources,
            unavailable=unavailable,
            linked=linked,
        ),
        activation=log,
        layers=layers,
    )

excluding_speakers

excluding_speakers(speakers: frozenset[str]) -> RetrievalCoordinator

Copy the deployment policy, keeping the supplied coordinator isolated.

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

def excluding_speakers(self, speakers: frozenset[str]) -> RetrievalCoordinator:
    """Copy the deployment policy, keeping the supplied coordinator isolated."""
    combined = self._exclude_speakers | frozenset(speakers)
    if combined == self._exclude_speakers:
        return self
    cloned = copy(self)
    cloned._exclude_speakers = combined
    return cloned

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)

Frozen retrieval policy, validated at wiring so a bad floor fails early.

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