symfonic.capabilities.memory.factory

factory

Memory as something a host can compose, from a store and a scope.

The public counterpart to what SymfonicAgent._retrieval_bundle does privately. A generated project needs memory on an Agent and today the only way to get it is to build a SymfonicAgent, which is the dependency the kernel-native scaffold exists to remove.

One store, one scope, one capability. Capabilities are scope-bound at construction -- a memory capability recalls, records and erases for exactly one scope -- which is why a host keeps one agent per scope rather than passing a scope per turn. This factory is where that binding happens, and it is the only place it needs to.

The three grants travel together, and that is not a convenience. memory-read without memory-write is a deployment that recalls and never records; memory-write without memory-flush is worse -- the write stage stages a memory as pending, and a pending memory is not retrievable until flush, so a turn would answer, the store would fill, and nothing could ever be recalled. TA8.71 found exactly that shipped. Asking for memory means asking for all three.

Nothing here imports the engine, reads a private, or knows what a turn is.

MemoryBundleFactory

MemoryBundleFactory(store: Any, *, limit: int = 5, recall_budget: RecallBudget | None = None, extractor: MemoryExtractorPort | None = None, summary: STMSummaryPort | None = None, reflection: ReflectionPort | None = None, consolidation: Any | None = None, conversation: Any | None = None, recent_turns: int = 0, activation: Any | None = None, retrieval_policy: RetrievalPolicy | None = None, retrieval: RetrievalCoordinator | None = None, exclude_speakers: frozenset[str] = frozenset(), episodic_telemetry_sink: EpisodicTelemetrySink | None = None, operation_journal: MemoryOperationJournalPort | None = None, generations: Any = None, triage: IntentTriageCapability | None = None)

Builds the memory capability for one scope over a shared store.

The store is shared across scopes and the capability is not: a store isolates by scope on every read and write, and a capability is a scope. Handing the same store to two capabilities is how two tenants share persistence without sharing memory.

Parameters:

Name	Type	Description	Default
store	Any	an HMS satisfying the retrieval, write and lifecycle ports. One object for all three because staging, publishing and erasing are operations on one place -- and because a deployment that split them would have to answer what happens when only two are present, which the ports already refuse.	required
limit	int	how many memories a turn recalls.	5
recall_budget	RecallBudget | None	explicit UTF-8 block/item ceilings; independent of working-turn retention. None preserves legacy character caps.	None
extractor	MemoryExtractorPort | None	a :class:~.ports.MemoryExtractorPort, normally :class:~.extraction.MemoryExtractionService built over an extraction model. Given one, the capability runs it after the final model round and writes what it returns, so a turn that said something worth keeping is remembered without the deployment calling anything. None -- the default -- means this deployment does not extract: it still recalls, records the exchange and erases. Checked here rather than mid-turn; see :func:~.ports.validate_extractor.	None
summary	STMSummaryPort | None	an explicit evicted-turn summary collaborator. It is materialized only into JIT prompts; omitting it disables the contribution and makes no source call.	None
reflection	ReflectionPort | None	a completed-exchange insight collaborator. It runs after publication and persists accepted scoped insights.	None
consolidation	Any | None	a :class:~.napping.ConsolidationCoordinator. Given one, the capability naps on the cadence it carries -- after the turn's memories are published, on the run's background registry, once per scope. None -- the default -- means this deployment consolidates from its own scheduler or not at all, and one shared coordinator across scopes is correct: it counts turns per scope and locks per scope.	None

Raises:

Type	Description
ConfigurationError	if extractor, summary, or reflection cannot serve its port.

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

def __init__(
    self,
    store: Any,
    *,
    limit: int = 5,
    recall_budget: RecallBudget | None = None,
    extractor: MemoryExtractorPort | None = None,
    summary: STMSummaryPort | None = None,
    reflection: ReflectionPort | None = None,
    consolidation: Any | None = None,
    conversation: Any | None = None,
    recent_turns: int = 0,
    activation: Any | None = None,
    retrieval_policy: RetrievalPolicy | None = None,
    retrieval: RetrievalCoordinator | None = None,
    exclude_speakers: frozenset[str] = frozenset(),
    episodic_telemetry_sink: EpisodicTelemetrySink | None = None,
    operation_journal: MemoryOperationJournalPort | None = None,
    generations: Any = None,  # minted by memory_capabilities only (CORE-199)
    triage: IntentTriageCapability | None = None,
) -> None:
    """
    Args:
        store: an HMS satisfying the retrieval, write and lifecycle ports.
            One object for all three because staging, publishing and
            erasing are operations on one place -- and because a
            deployment that split them would have to answer what happens
            when only two are present, which the ports already refuse.
        limit: how many memories a turn recalls.
        recall_budget: explicit UTF-8 block/item ceilings; independent of
            working-turn retention. None preserves legacy character caps.
        extractor: a :class:`~.ports.MemoryExtractorPort`, normally
            :class:`~.extraction.MemoryExtractionService` built over an
            extraction model. Given one, the capability runs it after the
            final model round and writes what it returns, so a turn that
            said something worth keeping is remembered without the
            deployment calling anything. ``None`` -- the default -- means
            this deployment does not extract: it still recalls, records
            the exchange and erases. Checked here rather than mid-turn;
            see :func:`~.ports.validate_extractor`.
        summary: an explicit evicted-turn summary collaborator. It is
            materialized only into JIT prompts; omitting it disables the
            contribution and makes no source call.
        reflection: a completed-exchange insight collaborator. It runs
            after publication and persists accepted scoped insights.
        consolidation: a
            :class:`~.napping.ConsolidationCoordinator`. Given one, the
            capability naps on the cadence it carries -- after the turn's
            memories are published, on the run's background registry, once
            per scope. ``None`` -- the default -- means this deployment
            consolidates from its own scheduler or not at all, and one
            shared coordinator across scopes is correct: it counts turns
            per scope and locks per scope.

    Raises:
        ConfigurationError: if ``extractor``, ``summary``, or
            ``reflection`` cannot serve its port.
    """
    validate_extractor(extractor)
    validate_summary(summary)
    validate_reflector(reflection)
    if recall_budget is not None:
        from symfonic.capabilities.memory.budget import RecallBudget
        from symfonic.capabilities.memory.errors import MemoryContractError
        if not isinstance(recall_budget, RecallBudget):
            raise MemoryContractError("recall_budget must be a RecallBudget")
    self._store = store
    self._limit = limit
    self._recall_budget = recall_budget
    self._extractor = extractor
    self._summary = summary
    self._reflection = reflection
    self._consolidation = consolidation
    self._conversation = conversation
    self._recent_turns = recent_turns
    self._activation = activation
    self._retrieval_policy = retrieval_policy
    self._retrieval = retrieval
    self._exclude_speakers = exclude_speakers
    self._episodic_telemetry_sink = episodic_telemetry_sink
    self._operation_journal = operation_journal
    self._generations = generations
    self._triage = triage

for_scope

for_scope(scope: MemoryScope | Any) -> Any

Compose kind-aware paths, or an explicitly positional MemoryScope.

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

def for_scope(self, scope: MemoryScope | Any) -> Any:
    """Compose kind-aware paths, or an explicitly positional MemoryScope."""
    from symfonic.capabilities.memory.operation_journal import receipt_backed
    memory_scope = composition_scope(scope)
    if receipt_backed(self._store):
        from symfonic.capabilities.memory.receipt_composition import receipt_for_scope
        return receipt_for_scope(self, memory_scope)
    if memory_scope.identity_version:
        from symfonic.capabilities.memory.graph_native_admission import require_composition
        require_composition(self, memory_scope)
    from symfonic.capabilities.memory.turn_generation import guarded_writer
    writer = guarded_writer(self._store, self._conversation, memory_scope, self._generations)
    hydrator = hydrator_for(
        self._store, conversation=self._conversation, recent_turns=self._recent_turns,
        activation=self._activation,
        retrieval_policy=self._retrieval_policy,
        retrieval=self._retrieval,
        exclude_speakers=self._exclude_speakers,
    )
    from symfonic.capabilities.memory.port_inspection import pair_guard
    return MemoryCapability(
        hydrator,
        scope=memory_scope,
        limit=self._limit,
        recall_budget=self._recall_budget,
        writer=pair_guard(writer, hydrator, memory_scope),
        # Closed over the scope, not read from the turn. ``Agent.run``
        # takes a prompt and nothing about tenancy -- that is the published
        # facade decision -- so the turn request carries no scope and a
        # producer that looked for one filed nothing at all. The agent IS
        # the scope: it was composed for exactly this one, which is the
        # same fact that makes a host keep one agent per scope.
        records_from=_producer_for(memory_scope),
        lifecycle=writer,
        extractor=self._extractor,
        summary=self._summary,
        reflection=self._reflection,
        consolidation=self._consolidation,
        episodic_telemetry_sink=self._episodic_telemetry_sink,
        operation_journal=self._operation_journal,
        triage=self._triage,
    )

as_memory_scope

as_memory_scope(scope: Any) -> MemoryScope

Refuse paths deeper than this positional memory scope can preserve.

Existing one- to three-level kinds still use the legacy positional key. Their kind-identity migration is tracked separately under CORE-171.

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

def as_memory_scope(scope: Any) -> MemoryScope:
    """Refuse paths deeper than this positional memory scope can preserve.

    Existing one- to three-level kinds still use the legacy positional key.
    Their kind-identity migration is tracked separately under CORE-171.
    """
    if isinstance(scope, MemoryScope):
        return scope
    path_ids: tuple[str, ...] = ()
    levels = getattr(scope, "path", None)
    if levels and not isinstance(levels, (str, bytes)):
        path_levels = tuple(levels)
        if len(path_levels) > 3:
            raise MemoryContractError(
                "scope path has more than three levels and cannot be represented by MemoryScope"
            )
        path_ids = tuple(str(level.id) for level in path_levels if getattr(level, "id", None))
        if len(path_ids) != len(path_levels):
            raise MemoryContractError("scope path contains an empty level")
    segment_ids: tuple[str, ...] = ()
    segments = getattr(scope, "segments", None)
    if segments:
        raw_segments = tuple(segments)
        segment_ids = tuple(str(segment) for segment in raw_segments if segment)
        if len(segment_ids) != len(raw_segments):
            raise MemoryContractError("scope segments contain an empty level")
        if len(segment_ids) > 3:
            raise MemoryContractError(
                "scope path has more than three levels and cannot be represented by MemoryScope"
            )
    if path_ids and segment_ids and path_ids != segment_ids:
        raise MemoryContractError("scope path and segments disagree")
    ids = path_ids or segment_ids
    converted = getattr(scope, "to_memory_scope", None)
    if callable(converted):
        result = converted()
        if isinstance(result, MemoryScope):
            if ids and result.segments != ids:
                raise MemoryContractError(
                    "to_memory_scope lost a scope level before memory composition"
                )
            return result
    if ids:
        return MemoryScope(*ids)
    raise TypeError(
        f"{type(scope).__name__} is neither a MemoryScope nor convertible to "
        "one, so the capability cannot be bound to a scope"
    )