symfonic.capabilities.memory.bridge

bridge

The HMS capability bridge: the only place memory touches an invocation.

Three seams, one per phase of the ladder that memory has any business in:

• prompt/input — :meth:HmsBridge.hydrate calls the retrieval port and turns what it returns into one declared contribution. It registers as a stage in prompt-assembly that must precede the prompt compiler's own, because a recall that arrives after the compile is a recall that is not in the prompt.
• post-response — :meth:HmsBridge.record calls the write port in post-model, once the turn has something to remember.
• finalize — :meth:HmsBridge.close flushes in finalize. Not teardown: that phase is kernel-owned (a capability that could inject into it could outlive a run it does not own), and the ladder already guarantees post-model precedes finalize without anyone declaring an edge.

The kernel never imports any of this. The bridge declares :class:~symfonic.kernel.contracts.stages.StageDescriptor values and holds ports; the concrete HMS arrives pre-bound from a composition root. That is the whole "no kernel imports of memory implementations" requirement, and it is checked by the architecture gate rather than asserted here.

Transport degrades, contracts propagate. A store that is down produces an empty, degraded result and the turn continues; a store that is misconfigured raises. The one exception is :class:~.errors.ScopeViolation, which propagates from both paths — a tenant boundary crossing is not a degraded turn.

HmsBridge

HmsBridge(*, retrieval: MemoryRetrievalPort, writes: MemoryWritePort, lifecycle: MemoryLifecyclePort, contribution_id: str = 'memory.recall', order: int = 0)

Binds the three memory ports to the three invocation seams.

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

def __init__(
    self,
    *,
    retrieval: MemoryRetrievalPort,
    writes: MemoryWritePort,
    lifecycle: MemoryLifecyclePort,
    contribution_id: str = "memory.recall",
    order: int = 0,
) -> None:
    if not contribution_id or not _ID_CHARSET.match(contribution_id):
        raise MemoryContractError(
            f"contribution id {contribution_id!r} is outside the permitted charset "
            "[A-Za-z0-9_.-]; the bridge refuses it here so a deployment fails at wiring "
            "time rather than on its first compile."
        )
    self._retrieval = retrieval
    self._writes = writes
    self._lifecycle = lifecycle
    self._contribution_id = contribution_id
    self._order = order

close async

close(scope: MemoryScope) -> LifecycleReceipt

Commit this invocation's pending memories.

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

async def close(self, scope: MemoryScope) -> LifecycleReceipt:
    """Commit this invocation's pending memories."""
    try:
        return await self._lifecycle.flush(scope)
    except MemoryUnavailable:
        return LifecycleReceipt(scope_path=scope.path, degraded=True)

forget async

forget(scope: MemoryScope) -> LifecycleReceipt

Erase a subtree. The privacy seam (SEC-PRIV), not an invocation stage.

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

async def forget(self, scope: MemoryScope) -> LifecycleReceipt:
    """Erase a subtree. The privacy seam (SEC-PRIV), not an invocation stage."""
    try:
        return await self._lifecycle.forget(scope)
    except MemoryUnavailable:
        return LifecycleReceipt(scope_path=scope.path, degraded=True)

hydrate async

hydrate(query: MemoryQuery) -> Hydration

Retrieve for query and declare the result as one contribution.

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

async def hydrate(self, query: MemoryQuery) -> Hydration:
    """Retrieve for ``query`` and declare the result as one contribution."""
    try:
        result = await self._retrieval.retrieve(query)
    except MemoryUnavailable as exc:
        result = RetrievalResult(
            dropped=((self._contribution_id, f"memory store unreachable: {exc}"),),
            degraded=True,
        )
    else:
        self._check_visibility(result, query.scope)

    scope_path = query.scope.path
    source = HydratedMemorySource(result=result, scope_path=scope_path)
    return Hydration(
        query=query,
        result=result,
        contribution=MemoryContribution(
            contribution_id=self._contribution_id,
            source=source,
            order=self._order,
        ),
        request=MemoryRequest(
            contribution_id=self._contribution_id, scope_path=scope_path, turn=query.turn
        ),
    )

record async

record(request: WriteRequest) -> WriteReceipt

Write what the turn produced, reporting failure rather than hiding it.

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

async def record(self, request: WriteRequest) -> WriteReceipt:
    """Write what the turn produced, reporting failure rather than hiding it."""
    try:
        return await self._writes.write(request)
    except MemoryUnavailable:
        return WriteReceipt(
            rejected=tuple(
                (record.record_id, "memory store unreachable") for record in request.records
            ),
            degraded=True,
        )

stages

stages(hydration: Hydration | None = None) -> tuple[StageDescriptor, ...]

The three stages this capability contributes, in ladder order.

hydration is optional because the ladder is knowable before a turn runs — a plan can be compiled and inspected without retrieving anything. When it is supplied, the retrieval stage carries what was hydrated in its frozen config, which is how "this plan recalled these memories" stays checkable from the plan alone.

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

def stages(self, hydration: Hydration | None = None) -> tuple[StageDescriptor, ...]:
    """The three stages this capability contributes, in ladder order.

    ``hydration`` is optional because the ladder is knowable before a turn
    runs — a plan can be compiled and inspected without retrieving anything.
    When it is supplied, the retrieval stage carries what was hydrated in its
    frozen config, which is how "this plan recalled these memories" stays
    checkable from the plan alone.
    """
    return (
        StageDescriptor(
            stage_id=RETRIEVAL_STAGE,
            phase=Phase.PROMPT_ASSEMBLY,
            capability=HMS_CAPABILITY,
            priority=_RETRIEVAL_PRIORITY,
            optional_before=(PROMPT_COMPILER_STAGE,),
            effects=frozenset({"memory-read"}),
            # The stage STG-7's split was reformulated for: it reaches the
            # store, once per invocation, and contributes to the snapshot
            # the prompt compiler then reads. It never writes the assembly.
            kind=StageKind.RESOLUTION,
            emits=frozenset({"memory.retrieved"}),
            config=self._retrieval_config(hydration),
        ),
        StageDescriptor(
            stage_id=WRITE_STAGE,
            phase=Phase.POST_MODEL,
            capability=HMS_CAPABILITY,
            priority=100,
            effects=frozenset({"memory-write"}),
            emits=frozenset({"memory.written"}),
            config={"contribution": self._contribution_id},
        ),
        StageDescriptor(
            stage_id=LIFECYCLE_STAGE,
            phase=Phase.FINALIZE,
            capability=HMS_CAPABILITY,
            priority=900,
            effects=frozenset({"memory-flush"}),
            emits=frozenset({"memory.flushed"}),
            config={"contribution": self._contribution_id},
        ),
    )

Hydration dataclass

Hydration(query: MemoryQuery, result: RetrievalResult, contribution: MemoryContribution, request: MemoryRequest)

One completed prompt/input pass: what was asked, found, and declared.