symfonic.capabilities.memory

memory

The hierarchical memory system, as one capability instead of a dependency.

The HMS used to be reachable from everywhere: an engine called it to hydrate a prompt, a post-response path called it to extract, a scheduler called it to consolidate, and several modules imported its internals. That is why "run this agent without memory" was a fork rather than a configuration.

• The seam (T3.3.1) -- three narrow ports (:mod:.ports) and one bridge (:mod:.bridge) that declares stages instead of being called. :mod:.in_memory is the zero-dependency reference adapter; the vocabulary the ports move (:mod:.scope, :mod:.layers, :mod:.records, :mod:.queries) validates at construction.
• The services (T3.3.3) -- extraction (:mod:.extraction), the owned-write coordinator (:mod:.writes), and the consolidation runtime (:mod:.consolidation) with promotion and entity linking. :mod:.compat keeps everything they persist readable by the legacy path.

This facade re-exports the broad memory surface. Recall has its own curated surface at :mod:.recall; these hub aliases preserve established imports.

Not here, deliberately: the Postgres/Mongo/graph adapters behind the ports are T3.3.4's.

Association dataclass

Association(source_id: str, target: MemoryRecord, relationship: str = 'associated', weight: float = 1.0)

One edge out of a memory, as a backend reports it.

AssociationSource

Bases: Protocol

The graph half of the memory system, as activation needs it.

neighbours async

neighbours(scope: MemoryScope, record_ids: tuple[str, ...]) -> Sequence[Association]

Every edge out of record_ids, one round trip per frontier.

Takes the whole frontier rather than one id so a hop costs one query instead of one per seed. Raises :class:~.errors.MemoryUnavailable when the graph cannot be reached.

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

async def neighbours(
    self, scope: MemoryScope, record_ids: tuple[str, ...]
) -> Sequence[Association]:
    """Every edge out of ``record_ids``, one round trip per frontier.

    Takes the whole frontier rather than one id so a hop costs one query
    instead of one per seed. Raises :class:`~.errors.MemoryUnavailable` when
    the graph cannot be reached.
    """
    ...

BackgroundWorkPort

Bases: Protocol

Somewhere to put owned work — structurally, T2.3.4's registry.

spawn

spawn(work: Coroutine[Any, Any, Any], *, owner: str, purpose: str, deadline_seconds: float | None = None) -> Any

Register and start one unit of run-owned work.

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

def spawn(
    self,
    work: Coroutine[Any, Any, Any],
    *,
    owner: str,
    purpose: str,
    deadline_seconds: float | None = None,
) -> Any:
    """Register and start one unit of run-owned work."""
    ...

ConsolidationCoordinator

ConsolidationCoordinator(runtime: ConsolidationRuntime, *, schedule: ConsolidationSchedule | None = None, cursors: MutableMapping[str, ScheduleCursor] | None = None, background: bool = True, cycles: Sequence[ConsolidationCycle] = (ConsolidationCycle.QUICK,), leases: Any = None, lease_ttl_seconds: float = 900.0, transaction: Any = None, terminal_sink: Any = None)

Bases: BackgroundCycles

Decides when a scope consolidates, and runs at most one cycle at a time.

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

def __init__(
    self,
    runtime: ConsolidationRuntime,
    *,
    schedule: ConsolidationSchedule | None = None,
    cursors: MutableMapping[str, ScheduleCursor] | None = None,
    background: bool = True,
    cycles: Sequence[ConsolidationCycle] = (ConsolidationCycle.QUICK,),
    leases: Any = None,
    #: How long a lease stands before it lapses. Long by default because
    #: a port with no ``renew`` has only the TTL between a slow cycle and
    #: losing its scope; on a renewing port -- both shipped adapters are --
    #: it can safely be far shorter. See :mod:`.heartbeat`.
    lease_ttl_seconds: float = 900.0,
    #: The one transaction domain this cycle commits in: the pool the
    #: graph, the lease and the staged records all speak through. Required
    #: whenever the lease port has one, and checked here at composition
    #: rather than discovered from a report that said ``clean`` about half
    #: a cycle. See :func:`~.commit.require_domain_for`.
    transaction: Any = None,
    terminal_sink: Any = None,
) -> None:
    self._runtime = runtime
    self._terminal_sink = terminal_sink
    self._cadence = Cadence(schedule or ConsolidationSchedule(), cycles, cursors)
    # Required, and not defaulted to the in-process table: a default that
    # silently coordinated one process would be the exact failure this port
    # exists to prevent, arrived at by omission.
    if leases is None:
        raise MemoryContractError(
            "a consolidation coordinator needs a lease port. Two cycles "
            "over one scope corrupt each other's counters and double its "
            "model spend, and the processes that would collide -- a "
            "scheduler and a manual backfill -- do not share memory. Pass "
            "``PostgresLeases(pool)``, or "
            "``InProcessLeases(single_process=True)`` if this deployment "
            "really is one process."
        )
    self._leases = leases
    self._domain = require_domain_for(leases, runtime, transaction)
    self._lease_ttl = lease_ttl_seconds
    self._tasks = OwnedCycles()
    #: Whether the turn waits for the cycle. A deployment that wants the
    #: answer held until its memories are consolidated sets this ``False``.
    self.background = background
    for cycle in cycles:
        if not runtime.serves(cycle):
            raise MemoryContractError(
                f"this coordinator would schedule the {cycle.value} cadence and "
                f"its runtime cannot serve it (registered: "
                f"{', '.join(runtime.registered) or 'nothing'}). Build the "
                "runtime from ``quick_phases`` so the roster is complete "
                "before anything is scheduled against it."
            )

leases property

leases: Any

The lease port. Published so a composition root can prepare it.

An adapter backed by a table has schema to create, and the app that opened the pool is the thing that knows when to do it.

registered property

registered: tuple[str, ...]

The phases a cycle from this coordinator will run, by name.

Published because "which phases does my nap run?" is a question an operator asks of the thing they configured, and answering it by reaching for the runtime inside would be the private access this capability exists to remove.

after_turn

after_turn(scope: MemoryScope) -> ConsolidationCycle | None

Record a successful turn and answer which cycle is now due.

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

def after_turn(self, scope: MemoryScope) -> ConsolidationCycle | None:
    """Record a successful turn and answer which cycle is now due."""
    return self._cadence.after_turn(scope)

cursor

cursor(scope: MemoryScope) -> ScheduleCursor

This scope's counter. A scope nobody has served yet has a fresh one.

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

def cursor(self, scope: MemoryScope) -> ScheduleCursor:
    """This scope's counter. A scope nobody has served yet has a fresh one."""
    return self._cadence.cursor(scope)

run async

run(scope: MemoryScope, cycle: ConsolidationCycle, *, run_id: str = '', root_run_id: str = '') -> ConsolidationState

Run one cycle over one scope, alone.

run_id/root_run_id are the turn this cycle is attributable to. A quick nap fires from a turn and phase 13 may spend a model call, so the cost belongs to that tenant and that root run rather than to whichever turn happened to be in flight when the task was scheduled.

Bound as the ambient run identity for the cycle, not merely passed on the phase context: a provider reads the ContextVar, and a phase calls the model through the provider rather than through anything that could be handed the ids. Bound explicitly rather than inherited from the task that created this one, so a cycle a scheduler ran outside any turn carries no identity instead of quietly borrowing whichever run happened to be ambient when the scheduler ticked.

Returns a state whose status is already_running when another holder has the scope. A None would have been the smaller change and the wrong one: the caller needs a record saying why nothing happened, and an absent one reads like a cycle that ran and found nothing.

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

async def run(
    self,
    scope: MemoryScope,
    cycle: ConsolidationCycle,
    *,
    run_id: str = "",
    root_run_id: str = "",
) -> ConsolidationState:
    """Run one cycle over one scope, alone.

    ``run_id``/``root_run_id`` are the turn this cycle is attributable to.
    A quick nap fires from a turn and phase 13 may spend a model call, so
    the cost belongs to that tenant and that root run rather than to
    whichever turn happened to be in flight when the task was scheduled.

    Bound as the ambient run identity for the cycle, not merely passed on
    the phase context: a provider reads the ContextVar, and a phase calls
    the model through the provider rather than through anything that could
    be handed the ids. Bound explicitly rather than inherited from the task
    that created this one, so a cycle a scheduler ran outside any turn
    carries no identity instead of quietly borrowing whichever run happened
    to be ambient when the scheduler ticked.

    Returns a state whose ``status`` is ``already_running`` when another
    holder has the scope. A ``None`` would have been the smaller change and
    the wrong one: the caller needs a record saying *why* nothing happened,
    and an absent one reads like a cycle that ran and found nothing.
    """
    owner = new_owner_token()
    lease = await self._leases.acquire(scope, owner=owner, ttl_seconds=self._lease_ttl)
    if lease is None:
        logger.info(
            "consolidation cycle=%s scope=%s skipped: already running",
            cycle.value,
            scope.path,
        )
        return already_running(scope, cycle)
    try:
        # Renewed while the cycle runs, so the TTL can stay short enough
        # for a crash to be noticed without a slow-but-living worker
        # losing its scope to its own model call.
        async with heartbeat(self._leases, lease):
            with attributed(run_id, root_run_id):
                return await self._runtime.run(
                    scope,
                    cycle,
                    run_id=run_id,
                    root_run_id=root_run_id,
                    # The fence. Checked before every durable write, so a
                    # cycle whose lease lapsed anyway stops rather than racing
                    # the worker that took the scope over.
                    still_authorised=lambda: self._leases.holds(lease),
                    # The commit-time check, which locks the lease row and
                    # keeps it until the transaction ends, so a rival
                    # acquisition queues behind this commit rather than
                    # landing between the answer and the batch it allowed.
                    commit_authority=lambda: self._leases.hold_for_update(lease),
                    domain=self._domain,
                )
    finally:
        # On success, on failure and on cancellation. Owner-checked inside
        # the port, so a slow worker whose lease was taken over does not
        # free it for whoever holds it now.
        await self._leases.release(lease)

ConsolidationCycle

Bases: StrEnum

The three consolidation cadences, narrowest first.

ConsolidationRuntime

ConsolidationRuntime(*, phases: Sequence[ConsolidationPhase] = (), writes: MemoryWriteCoordinator | None = None, cycles: Sequence[ConsolidationCycle] = (), participants: Sequence[Any] = (), max_new_edges_per_sweep: int | None = None)

Runs one cadence's roster over one scope.

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

def __init__(
    self,
    *,
    phases: Sequence[ConsolidationPhase] = (),
    writes: MemoryWriteCoordinator | None = None,
    cycles: Sequence[ConsolidationCycle] = (),
    #: The stores this runtime's phases write through, for a composition
    #: root to check against its transaction domain *before* a cycle runs.
    #: Taken from the roster when the phases came from one of the shipped
    #: factories, which is where the graph was resolved; pass it explicitly
    #: for a hand-built roster, or the mismatch is only caught at the
    #: commit -- after seventeen phases and their model calls.
    participants: Sequence[Any] = (),
    #: Gross new edges per durable scope/cadence sweep; None is unlimited.
    #: Exhaustion defers groups, not cursor progress. Mid-sweep changes refuse.
    max_new_edges_per_sweep: int | None = None,
) -> None:
    from symfonic.capabilities.memory.growth import validate_limit
    validate_limit(max_new_edges_per_sweep)
    self._edge_limit = max_new_edges_per_sweep
    self._phases = {phase.name: phase for phase in phases}
    self._writes = writes
    from symfonic.capabilities.memory.cycle_writes import validate_writes
    if self._edge_limit is not None:
        validate_writes(writes)
    declared = getattr(phases, "graph", None)
    self._participants: tuple[Any, ...] = tuple(participants) or (
        (declared,) if declared is not None else ()
    )
    for cycle in cycles:
        self._require_complete(cycle, at="composition")

participants property

participants: tuple[Any, ...]

The stores this runtime's phases write through.

Published so a composition root can prove, before any cycle runs, that the graph the phases mutate is in the same transaction domain as the lease that authorises them. Empty for a hand-built roster that declared none, which is the one case where the check has to wait for the commit.

writes property

writes: MemoryWriteCoordinator | None

The coordinator this runtime publishes through, if it has one.

Published so a composition root can check, before any cycle runs, that the staged records and the graph mutations share one transaction domain.

run async

run(scope: MemoryScope, cycle: ConsolidationCycle, *, run_id: str = '', root_run_id: str = '', still_authorised: Any = None, commit_authority: Any = None, domain: Any = None) -> ConsolidationState

Execute cycle's roster over scope and report what happened.

run_id/root_run_id name the turn this cycle is attributable to. A quick nap fires from a turn and one of its phases can spend a model call, so the cost has an owner rather than landing on whichever run was in flight when the background task was scheduled. Empty is the honest answer for a cycle a scheduler ran outside any turn.

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

async def run(
    self,
    scope: MemoryScope,
    cycle: ConsolidationCycle,
    *,
    run_id: str = "",
    root_run_id: str = "",
    still_authorised: Any = None,
    commit_authority: Any = None,
    domain: Any = None,
) -> ConsolidationState:
    """Execute ``cycle``'s roster over ``scope`` and report what happened.

    ``run_id``/``root_run_id`` name the turn this cycle is attributable to.
    A quick nap fires from a turn and one of its phases can spend a model
    call, so the cost has an owner rather than landing on whichever run was
    in flight when the background task was scheduled. Empty is the honest
    answer for a cycle a scheduler ran outside any turn.
    """
    # Second gate, for a runtime composed without declaring its cadences.
    # A caller that asks for a cycle this runtime cannot serve gets an
    # error rather than a state, because a state is a report of work.
    self._require_complete(cycle, at="the call")

    state = ConsolidationState(
        scope_path=scope.path,
        cycle=cycle,
        registered=self.registered,
        run_id=run_id,
        root_run_id=root_run_id,
    )
    from symfonic.capabilities.memory.cycle_writes import prepare_cycle
    buffered, domain = prepare_cycle(self._writes, self._edge_limit, domain)
    context = PhaseContext(
        scope=scope,
        cycle=cycle,
        started_at=state.started_at,
        writes=buffered,
        run_id=run_id,
        root_run_id=root_run_id,
    )
    fence = (
        Fence(still_authorised, commit_authority)
        if still_authorised is not None
        else None
    )
    journal = CycleJournal(cycle=cycle.value, max_new_edges_per_sweep=self._edge_limit)
    # The phases run against the journal: their mutations are held, and
    # read back, but nothing is durable until this cycle earns it below.
    async with held(fence), journalled(journal):
        await self._roster(cycle, context, state, journal)
        await still_ours(state, fence)
        state.ledger = dict(context.ledger)
        state.counters.update(context.legacy)
        state.ledger["candidates"] = len(context.seen)
        state.committed = await publish_cycle(
            scope, state, journal, fence, domain, buffered
        )
    state.ledger["published"] = len(state.committed)
    state.ledger.update(journal.growth_counts)
    state.finished_at = datetime.now(UTC)
    return state

serves

serves(cycle: ConsolidationCycle) -> bool

Whether every phase on cycle's roster has an implementation.

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

def serves(self, cycle: ConsolidationCycle) -> bool:
    """Whether every phase on ``cycle``'s roster has an implementation."""
    return not self._missing(cycle)

ConsolidationSchedule dataclass

ConsolidationSchedule(quick_every_turns: int = 5, nightly_after_seconds: float = 24 * 60 * 60, deep_after_seconds: float = 7 * 24 * 60 * 60)

The cadences, and the rule that picks one.

due

due(cursor: ScheduleCursor, *, now: datetime | None = None, among: Sequence[ConsolidationCycle] | None = None) -> ConsolidationCycle | None

The widest cycle due at now, or None.

among narrows the answer to cadences the caller can actually run. It is not a convenience: a cadence that has never run reads as infinitely overdue, so on a fresh deployment every cadence is due on turn one — and a turn-driven nap asked for the widest would be told to run Deep Sleep, which is the roster it was never composed for and the cost nobody scheduled. A caller that runs one cadence asks about one.

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

def due(
    self,
    cursor: ScheduleCursor,
    *,
    now: datetime | None = None,
    among: Sequence[ConsolidationCycle] | None = None,
) -> ConsolidationCycle | None:
    """The widest cycle due at ``now``, or ``None``.

    ``among`` narrows the answer to cadences the caller can actually run.
    It is not a convenience: a cadence that has never run reads as
    infinitely overdue, so on a fresh deployment *every* cadence is due on
    turn one — and a turn-driven nap asked for the widest would be told to
    run Deep Sleep, which is the roster it was never composed for and the
    cost nobody scheduled. A caller that runs one cadence asks about one.
    """
    moment = now if now is not None else datetime.now(UTC)
    allowed = _WIDEST_FIRST if among is None else tuple(among)
    for cycle in _WIDEST_FIRST:
        if cycle in allowed and self._is_due(cycle, cursor, moment):
            return cycle
    return None

validate

validate() -> None

Refuse a schedule under which a cadence can never fire.

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

def validate(self) -> None:
    """Refuse a schedule under which a cadence can never fire."""
    if self.quick_every_turns < 1:
        raise MemoryContractError(
            f"quick_every_turns is {self.quick_every_turns}; a cadence of zero turns "
            "either never fires or fires on every turn, and the two read identically "
            "from the call site. Disable a cadence by not scheduling it."
        )
    if self.nightly_after_seconds <= 0 or self.deep_after_seconds <= 0:
        raise MemoryContractError(
            "nightly and deep cadences are periods in seconds and must be positive."
        )
    if self.deep_after_seconds <= self.nightly_after_seconds:
        raise MemoryContractError(
            f"deep_after_seconds ({self.deep_after_seconds}) is not wider than "
            f"nightly_after_seconds ({self.nightly_after_seconds}). Deep Sleep is the "
            "wider roster; at this configuration every due nightly run is also a due "
            "deep run, so the nightly cadence never fires on its own."
        )

ConsolidationState dataclass

ConsolidationState(scope_path: str, cycle: ConsolidationCycle, tenant_id: str = '', started_at: datetime = (lambda: datetime.now(UTC))(), finished_at: datetime | None = None, run_id: str = '', root_run_id: str = '', registered: tuple[str, ...] = (), phases_run: tuple[str, ...] = (), skipped: tuple[str, ...] = (), failed: tuple[str, ...] = (), deferred: tuple[str, ...] = (), mutations: dict[str, int] = dict(), ledger: dict[str, int] = dict(), counters: dict[str, int] = dict(), errors: tuple[str, ...] = (), committed: tuple[str, ...] = (), already_running: bool = False, lease_lost: bool = False)

What one cycle did — in this capability's terms and in legacy's.

clean property

clean: bool

Whether every phase that ran, ran without error.

duration_ms property

duration_ms: int

How long the cycle took, or 0 while it is still running.

status property

status: str

The cycle's final word.

running, clean, degraded -- and two more that are none of those and must not be reported as any of them.

already_running: another holder had the scope, so this cycle never started. Saying clean would make "somebody else is consolidating this" indistinguishable from "there was nothing to consolidate", and a scheduler reading a dashboard would conclude the cycle had run.

lease_lost: this cycle started, then lost the scope partway. Not degraded either, though it is a kind of failure, because the two want opposite responses: degraded is a phase that broke and wants looking at, while lease_lost is a worker that correctly stood down so another one could do the work properly. Alerting on the second is alerting on the mechanism working. Ranked above degraded because a cycle that loses its lease also collects the fence's error, and the specific fact is the useful one.

telemetry

telemetry() -> dict[str, Any]

The safe record of this cycle: integers, roster names, and a status.

Everything here is either a framework constant or a count. The scope is not: tenant_id identifies whose consolidation this was, which is what makes the model cost a phase spends attributable, and it is already the key every other metric in the system carries.

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

def telemetry(self) -> dict[str, Any]:
    """The safe record of this cycle: integers, roster names, and a status.

    Everything here is either a framework constant or a count. The scope
    is *not*: ``tenant_id`` identifies whose consolidation this was, which
    is what makes the model cost a phase spends attributable, and it is
    already the key every other metric in the system carries.
    """
    return {
        "cycle_kind": self.cycle.value,
        "tenant_id": self.tenant_id or scope_from_path(self.scope_path).tenant,
        "scope_path": self.scope_path,
        # What a phase's model call is charged against. A nap fires from a
        # turn, so its cost belongs to that turn's root run rather than to
        # whichever run was in flight when the task was scheduled.
        "run_id": self.run_id,
        "root_run_id": self.root_run_id,
        "phases_registered": list(self.registered),
        "phases_run": list(self.phases_run),
        "phases_skipped": list(self.skipped),
        "phases_failed": list(self.failed),
        "phases_deferred": list(self.deferred),
        "registered": len(self.registered),
        "ran": len(self.phases_run),
        "skipped": len(self.skipped),
        "failed": len(self.failed),
        **{name: self.ledger.get(name, 0) for name in CYCLE_LEDGER},
        "duration_ms": self.duration_ms,
        "status": self.status,
    }

to_legacy_dict

to_legacy_dict() -> dict[str, Any]

The shipped ConsolidationReport.to_dict() shape, plus this cycle.

Every legacy key is present with its legacy type, so a reader written against the old report needs no change. The additions (cycle, scope_path, skipped, committed) are new keys, which a dict consumer ignores.

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

def to_legacy_dict(self) -> dict[str, Any]:
    """The shipped ``ConsolidationReport.to_dict()`` shape, plus this cycle.

    Every legacy key is present with its legacy type, so a reader written
    against the old report needs no change. The additions
    (``cycle``, ``scope_path``, ``skipped``, ``committed``) are new keys,
    which a dict consumer ignores.
    """
    payload: dict[str, Any] = {key: 0 for key in LEGACY_REPORT_KEYS}
    for phase, count in self.mutations.items():
        for counter in PHASE_COUNTERS.get(phase, ()):
            payload[counter] = payload.get(counter, 0) + count
    # Accumulated, not assigned: legacy's ``run`` does ``+=`` into these
    # fields too, and a counter fed from both sides must not lose one.
    for counter, count in self.counters.items():
        payload[counter] = payload.get(counter, 0) + count
    payload.update(
        {
            "tenant_id": self.tenant_id or scope_from_path(self.scope_path).tenant,
            "started_at": self.started_at.isoformat(),
            "finished_at": (
                self.finished_at.isoformat() if self.finished_at else None
            ),
            "errors": list(self.errors),
            "phases_run": list(self.phases_run),
            "cycle": self.cycle.value,
            "scope_path": self.scope_path,
            "skipped": list(self.skipped),
            "committed": list(self.committed),
        }
    )
    return payload

ContributionLayer

Bases: StrEnum

The stratigraphic layer a recall renders on.

ContributionScope

Bases: StrEnum

How widely one contribution's content is shared.

ContributionTier

Bases: StrEnum

Authority tiers, mirroring the prompt contract's vocabulary.

ConversationSource

Bases: Protocol

The working layer, as hydration needs it.

recent async

recent(scope: MemoryScope, limit: int) -> Sequence[ConversationTurn]

The last limit turns at scope, oldest first.

Raises :class:~.errors.MemoryUnavailable when the working store cannot be reached; :class:WorkingWindow degrades rather than failing the turn.

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

async def recent(self, scope: MemoryScope, limit: int) -> Sequence[ConversationTurn]:
    """The last ``limit`` turns at ``scope``, oldest first.

    Raises :class:`~.errors.MemoryUnavailable` when the working store cannot
    be reached; :class:`WorkingWindow` degrades rather than failing the turn.
    """
    ...

ConversationTurn dataclass

ConversationTurn(turn_id: str, speaker: str = '', text: str = '', turn: int = 0)

One thing that was said, as the working layer holds it.

line

line() -> str

The rendered form: layer prefix, speaker, single-line text.

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

def line(self) -> str:
    """The rendered form: layer prefix, speaker, single-line text."""
    body = flatten(self.text).strip()
    prefix = f"[{MemoryLayer.WORKING.value}]"
    return f"{prefix} {self.speaker}: {body}" if self.speaker else f"{prefix} {body}"

CredentialScrubber

CredentialScrubber(*, value_patterns: Iterable[tuple[str, Pattern[str]]] | None = None, key_parts: Iterable[str] | None = None)

Removes credentials from memory text and from memory properties.

Build a scrubber.

None selects the built-in set; an empty iterable disables that scan. The distinction is deliberate and matches the shipped hygiene contract: switching a scrubber off is something a deployment must say, not something it can fall into by passing an empty config.

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

def __init__(
    self,
    *,
    value_patterns: Iterable[tuple[str, re.Pattern[str]]] | None = None,
    key_parts: Iterable[str] | None = None,
) -> None:
    """Build a scrubber.

    ``None`` selects the built-in set; an **empty** iterable disables that
    scan. The distinction is deliberate and matches the shipped hygiene
    contract: switching a scrubber off is something a deployment must say,
    not something it can fall into by passing an empty config.
    """
    patterns = (
        CREDENTIAL_VALUE_PATTERNS if value_patterns is None else tuple(value_patterns)
    )
    self._value_patterns = patterns
    parts = (
        DEFAULT_CREDENTIAL_KEY_PARTS if key_parts is None else tuple(key_parts)
    )
    self._key_pattern = (
        re.compile("(?i)(" + "|".join(f"(?:{part})" for part in parts) + ")")
        if parts
        else None
    )

scrub_properties

scrub_properties(properties: Mapping[str, object]) -> tuple[dict[str, object], tuple[str, ...]]

Drop credential-named keys, and scrub the string values that remain.

Shallow, like the shipped scrubber: graph properties are persisted flat, so a nested dict is not a shape any backend writes.

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

def scrub_properties(
    self, properties: Mapping[str, object]
) -> tuple[dict[str, object], tuple[str, ...]]:
    """Drop credential-named keys, and scrub the string values that remain.

    Shallow, like the shipped scrubber: graph properties are persisted
    flat, so a nested dict is not a shape any backend writes.
    """
    clean: dict[str, object] = {}
    dropped: list[str] = []
    for key, value in properties.items():
        if self._key_pattern is not None and self._key_pattern.search(str(key)):
            dropped.append(key)
            continue
        clean[key] = (
            self.scrub_text(value).text if isinstance(value, str) else value
        )
    return clean, tuple(dropped)

scrub_text

scrub_text(text: str) -> ScrubResult

Replace every credential-shaped value in text.

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

def scrub_text(self, text: str) -> ScrubResult:
    """Replace every credential-shaped value in ``text``."""
    redactions: list[str] = []
    scrubbed = text
    for label, pattern in self._value_patterns:
        replacement = REDACTION_TEMPLATE.format(label=label)
        scrubbed, count = pattern.subn(replacement, scrubbed)
        redactions.extend([label] * count)
    return ScrubResult(text=scrubbed, redactions=tuple(redactions))

ExtractionRequest dataclass

ExtractionRequest(scope: MemoryScope, user_message: str = '', assistant_message: str = '', turn: int = 0, max_records: int = 50, min_importance: float = 3.0)

One turn, and the ceilings the extraction of it must respect.

FencedGraph

FencedGraph(graph: Any)

The graph a cycle writes through while it still holds the scope.

Wraps rather than subclasses: the phases take "a graph" duck-typed, the real one has a surface this module has no business restating, and anything not named in :data:FENCED_MUTATIONS is forwarded exactly as it was.

Parameters:

Name	Type	Description	Default
graph	Any	the store the phases would otherwise have been handed.	required

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

def __init__(self, graph: Any) -> None:
    """
    Args:
        graph: the store the phases would otherwise have been handed.
    """
    self._graph = graph

unfenced property

unfenced: Any

The wrapped store, for the runtime's own reads and for equality.

Named rather than private: a caller that legitimately needs the real object -- a test asserting on rows, a phase factory rewrapping -- should say so, instead of reaching for _graph and coupling to the field.

GraphAdminService

GraphAdminService(graph: Any, records: Any = None)

Relationship reads for one deployment's graph backend.

Parameters:

Name	Type	Description	Default
graph	Any	the GraphBackend the memory store persists through.	required
records	Any	a MemoryAdminService, needed only by :meth:export. Absent, an export would contain relationships and no memories, which is a worse answer to a portability request than an error.	None

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

def __init__(self, graph: Any, records: Any = None) -> None:
    """
    Args:
        graph: the ``GraphBackend`` the memory store persists through.
        records: a ``MemoryAdminService``, needed only by :meth:`export`.
            Absent, an export would contain relationships and no memories,
            which is a worse answer to a portability request than an error.
    """
    self._graph = graph
    self._records = records

edges async

edges(scope: MemoryScope, *, limit: int = 500, offset: int = 0) -> list[dict[str, Any]]

One bounded page of relationships, in newest backend order.

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

async def edges(
    self, scope: MemoryScope, *, limit: int = 500, offset: int = 0
) -> list[dict[str, Any]]:
    """One bounded page of relationships, in newest backend order."""
    found = await self._graph.query_edges(
        tenant_scope(scope), {}, limit=limit, offset=offset
    )
    return [edge_body(edge) for edge in found]

export async

export(scope: MemoryScope) -> dict[str, Any]

Everything this scope owns, as a portable document (GDPR Art. 20).

complete is part of the payload rather than an exception, because a partial export is still owed to the subject -- and a request that silently returned nine layers of ten would be a portability failure nobody could see. What could not be read is named.

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

async def export(self, scope: MemoryScope) -> dict[str, Any]:
    """Everything this scope owns, as a portable document (GDPR Art. 20).

    ``complete`` is part of the payload rather than an exception, because a
    partial export is still owed to the subject -- and a request that
    silently returned nine layers of ten would be a portability failure
    nobody could see. What could not be read is named.
    """
    if self._records is None:
        raise ValueError(
            "this GraphAdminService was built without a records service, so "
            "an export would carry relationships and no memories. Pass the "
            "MemoryAdminService that owns the same store."
        )
    failed: dict[str, str] = {}
    try:
        page = await self._records.record_page(scope, limit=MAX_CANDIDATE_LIMIT)
        memories = [_record_body(item.record) for item in page.memories]
        incomplete = page.dropped or page.degraded or page.unavailable
        if incomplete or len(memories) >= MAX_CANDIDATE_LIMIT:
            failed["memories"] = "bounded export is incomplete; use a paginated inventory"
    except Exception as unreadable:  # noqa: BLE001 - named, not swallowed
        memories, failed["memories"] = [], str(unreadable)[:200]
    try:
        relationships = await self.edges(scope, limit=100_000)
    except Exception as unreadable:  # noqa: BLE001
        relationships, failed["edges"] = [], str(unreadable)[:200]

    return {
        "scope_path": scope.path,
        "schema_version": EXPORT_SCHEMA_VERSION,
        "memories": memories,
        "edges": relationships,
        "complete": not failed,
        "failed": failed,
    }

neighborhood async

neighborhood(scope: MemoryScope, node_id: str, *, depth: int = 1) -> dict[str, Any]

One node's neighbours, out to depth hops.

Depth 1 is the immediate neighbours, which is what a graph view expands on a click. Deeper traversals are the backend's job -- doing it here with repeated neighbour calls would issue a query per node and call it a traversal.

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

async def neighborhood(
    self, scope: MemoryScope, node_id: str, *, depth: int = 1
) -> dict[str, Any]:
    """One node's neighbours, out to ``depth`` hops.

    Depth 1 is the immediate neighbours, which is what a graph view expands
    on a click. Deeper traversals are the backend's job -- doing it here
    with repeated neighbour calls would issue a query per node and call it
    a traversal.
    """
    if depth <= 1:
        found = await self._graph.get_neighbors(tenant_scope(scope), node_id)
    else:
        found = await self._graph.traverse(tenant_scope(scope), node_id, depth)
    return {
        "node_id": node_id,
        "depth": depth,
        "neighbors": [_node_body(node) for node in found],
    }

nodes async

nodes(scope: MemoryScope, *, limit: int = 500, offset: int = 0) -> list[dict[str, Any]]

Drawable nodes owned by scope, including its descendants.

A graph browser and a recall answer are different views. Recall is ranked and ancestor-facing; a graph needs the endpoints of the edges it was given, including conversation descendants, or it silently drops almost every relationship as dangling.

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

async def nodes(
    self, scope: MemoryScope, *, limit: int = 500, offset: int = 0
) -> list[dict[str, Any]]:
    """Drawable nodes owned by ``scope``, including its descendants.

    A graph browser and a recall answer are different views.  Recall is
    ranked and ancestor-facing; a graph needs the endpoints of the edges
    it was given, including conversation descendants, or it silently drops
    almost every relationship as dangling.
    """
    legacy = tenant_scope(scope)
    query = getattr(self._graph, "query_subtree", None)
    if limit <= 0:
        return []
    native_page = getattr(self._graph, "query_subtree_page", None)
    if callable(native_page):
        found = await native_page(legacy, {}, limit=limit, offset=offset)
        return [_node_body(node) for node in found]
    # GraphBackend's historical node contract has no offset. Fetch only
    # through the requested page and slice here; this bounds process memory
    # while keeping custom backends compatible. Native cursor pagination
    # can replace this without changing the public admin/API contract.
    through = offset + limit
    found = await (
        query(legacy, {}, limit=through)
        if callable(query)
        else self._graph.query_nodes(legacy, {}, limit=through)
    )
    return [_node_body(node) for node in found[offset:through]]

GraphBackedHms

GraphBackedHms(graph: Any, *, vectors: Any = None, embedder: Any = None)

All three memory ports, over a graph backend and optionally a vector one.

Composed with vectors and embedder, recall runs both routes and merges them: the graph answers what matches the cue's words, the vector index what matches its meaning, and a query sharing no vocabulary with the memory it needs is answered by the second. Composed without them, nothing changes -- which is why they are one optional pair rather than a second store class.

A pair because neither is useful alone: an embedder with nowhere to put a vector writes nothing, and a vector backend with no embedder cannot be queried. One without the other is refused at construction rather than discovered later as recall that was silently lexical.

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

def __init__(
    self, graph: Any, *, vectors: Any = None, embedder: Any = None
) -> None:
    if (vectors is None) != (embedder is None):
        raise ConfigurationError(
            "semantic recall needs a vector backend and an embedder "
            f"together; got vectors={type(vectors).__name__} and "
            f"embedder={type(embedder).__name__}. One without the other "
            "recalls nothing and looks composed."
        )
    self._graph = graph
    self._transaction_participants = (graph,) + ((vectors,) if vectors is not None else ())
    self._recall = (
        VectorRecall(vectors, embedder) if vectors is not None else None
    )
    from symfonic.capabilities.memory.mutations import mutation_fence_for

    self._mutations = mutation_fence_for(graph, vectors)

graph property

graph: Any

The store or backend this HMS writes through.

Published for the consolidation commit, which must establish that the records it publishes land in the same transaction domain as the graph mutations it applies beside them.

transaction_participants property

transaction_participants: tuple[Any, ...]

Every backend a write/flush can mutate, including vector publication.

discard async

discard(scope: MemoryScope) -> LifecycleReceipt

Delete the pending rows under scope; committed history survives.

Row-by-row rather than through delete_subtree, and deliberately: the subtree sweep is the privacy verb and takes everything. A turn taking back its own writes may not take the previous turns' with them, so the pending predicate has to be part of the selection.

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

async def discard(self, scope: MemoryScope) -> LifecycleReceipt:
    """Delete the pending rows under ``scope``; committed history survives.

    Row-by-row rather than through ``delete_subtree``, and deliberately:
    the subtree sweep is the *privacy* verb and takes everything. A turn
    taking back its own writes may not take the previous turns' with them,
    so the pending predicate has to be part of the selection.
    """
    discarded: list[str] = []
    async with self._mutations.hold(tenant_scope(scope)):
        for node in await self._pending_under(scope):
            if self._recall is not None:
                await self._recall.forget(
                    tenant_scope(node_scope(node)), (str(node.id),)
                )
            await self._graph.delete_node(
                tenant_scope(node_scope(node)), NodeId(str(node.id)), cascade=True
            )
            discarded.append(str(node.id))
    return LifecycleReceipt(scope_path=scope.path, discarded=tuple(sorted(discarded)))

forget async

forget(scope: MemoryScope) -> LifecycleReceipt

Erase scope and everything below it, pending or not.

The ids are read before the sweep because the receipt names them, not because the sweep needs them: delete_subtree is one backend-native statement over the whole subtree, so a row written into a descendant scope between the read and the delete is still erased.

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

async def forget(self, scope: MemoryScope) -> LifecycleReceipt:
    """Erase ``scope`` and everything below it, pending or not.

    The ids are read before the sweep because the receipt names them, not
    because the sweep needs them: ``delete_subtree`` is one backend-native
    statement over the whole subtree, so a row written into a descendant
    scope between the read and the delete is still erased.
    """
    legacy = tenant_scope(scope)
    async with self._mutations.hold(legacy):
        doomed_nodes = await self._graph.query_subtree(legacy, {}, limit=None)
        doomed = sorted(str(node.id) for node in doomed_nodes)
        await self._graph.delete_subtree(legacy)
        if self._recall is not None:
            await self._recall.forget_subtree(legacy)
    return LifecycleReceipt(scope_path=scope.path, discarded=tuple(doomed))

retrieve async

retrieve(query: MemoryQuery) -> RetrievalResult

Both routes, merged by record_id and ranked once.

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

async def retrieve(self, query: MemoryQuery) -> RetrievalResult:
    """Both routes, merged by ``record_id`` and ranked once."""
    candidates, sources, unavailable = await self._gather(query)
    return select(candidates, query, sources=sources, unavailable=unavailable)

scan_candidates async

scan_candidates(query: MemoryQuery) -> RetrievalResult

Every visible candidate, uncapped by query.limit (CandidateScan).

Not a call to :meth:retrieve: that one ends in select, which applies the limit and the character ceilings, and those are exactly the decisions a scan must leave to its caller.

Both routes, because this is the method a turn's hydration actually calls -- a scan that asked only the lexical route would leave the vector index composed, written to, and never consulted, which is the shape this pair exists to remove.

The vector half is bounded by its own top-k, which is the "store's own budget" this contract allows: a similarity search has no unbounded form, and asking for every vector in the scope would be a scan of the index rather than a search of it.

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

async def scan_candidates(self, query: MemoryQuery) -> RetrievalResult:
    """Every visible candidate, uncapped by ``query.limit`` (CandidateScan).

    Not a call to :meth:`retrieve`: that one ends in ``select``, which
    applies the limit and the character ceilings, and those are exactly
    the decisions a scan must leave to its caller.

    Both routes, because this is the method a turn's hydration actually
    calls -- a scan that asked only the lexical route would leave the
    vector index composed, written to, and never consulted, which is the
    shape this pair exists to remove.

    The vector half is bounded by its own top-k, which is the "store's own
    budget" this contract allows: a similarity search has no unbounded
    form, and asking for every vector in the scope would be a scan of the
    index rather than a search of it.
    """
    candidates, sources, unavailable = await self._gather(query)
    return RetrievalResult(
        memories=rank(candidates), sources=sources, unavailable=unavailable
    )

HeuristicEntityExtractor

Capitalised-token extraction with a stoplist. The zero-dependency default.

extract

extract(text: str) -> tuple[EntityMention, ...]

Every capitalised surface in text that survives the stoplist.

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

def extract(self, text: str) -> tuple[EntityMention, ...]:
    """Every capitalised surface in ``text`` that survives the stoplist."""
    mentions: list[EntityMention] = []
    seen: set[str] = set()
    for match in self._TOKEN.finditer(text):
        surface = match.group(0).strip()
        surface = _strip_stopwords(surface, self.STOPLIST)
        if not surface or surface in self.STOPLIST or surface in seen:
            continue
        seen.add(surface)
        mentions.append(EntityMention(surface=surface))
    return tuple(mentions)

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},
        ),
    )

HydratedMemorySource dataclass

HydratedMemorySource(result: RetrievalResult, scope_path: str, scope_aware: bool = True, offline_safe: bool = True)

A source over an already-hydrated retrieval, bound to the scope it used.

The binding is the security property. A compiled prompt is built from whatever contributions the caller passed; without the check below, a hydration performed for one tenant would render into another tenant's compile if a composition root reused the object — which is exactly the kind of reuse an object pool or a cached request makes easy.

InMemoryHms

InMemoryHms(*, records: Iterable[MemoryRecord] = (), capacity: int = 1024)

A complete HMS held in one process's memory.

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

def __init__(
    self,
    *,
    records: Iterable[MemoryRecord] = (),
    capacity: int = 1024,
) -> None:
    #: scope path -> record id -> record, for retrievable memories.
    self._committed: dict[str, dict[str, MemoryRecord]] = {}
    #: scope path -> record id -> record, for written-but-unflushed memories.
    self._pending: dict[str, dict[str, MemoryRecord]] = {}
    self._capacity = capacity
    for record in records:
        self._committed.setdefault(record.scope_path, {})[record.record_id] = record

discard async

discard(scope: MemoryScope) -> LifecycleReceipt

Drop the pending buffer for scope and below, committing nothing.

The optional rollback capability (MemoryDiscardPort). forget cannot serve as one: it erases the committed memories too, so a turn taking back its own writes would take every earlier turn's with them.

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

async def discard(self, scope: MemoryScope) -> LifecycleReceipt:
    """Drop the *pending* buffer for ``scope`` and below, committing nothing.

    The optional rollback capability (``MemoryDiscardPort``). ``forget``
    cannot serve as one: it erases the committed memories too, so a turn
    taking back its own writes would take every earlier turn's with them.
    """
    discarded: list[str] = []
    for scope_path in self._covered(self._pending, scope):
        discarded.extend(self._pending.pop(scope_path))
    return LifecycleReceipt(scope_path=scope.path, discarded=tuple(sorted(discarded)))

scan_candidates async

scan_candidates(query: MemoryQuery) -> RetrievalResult

Rank eligible rows before capping, retaining O(candidate_limit) rows.

The in-process reference scans its existing store, not a copied list. The ceiling bounds retained candidates, not CPU spent scoring rows.

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

async def scan_candidates(self, query: MemoryQuery) -> RetrievalResult:
    """Rank eligible rows before capping, retaining O(candidate_limit) rows.

    The in-process reference scans its existing store, not a copied list.
    The ceiling bounds retained candidates, not CPU spent scoring rows.
    """
    from .queries import rank_key
    from .record_admission import conversation_local, is_profile

    def eligible():
        for item in self._candidates(query):
            if (query.session_id and conversation_local(item.record)
                    and item.record.metadata.get("session_id") != query.session_id):
                continue
            yield item

    # Reserve a human profile before ordinary cue ranking, as the
    # coordinator does. Otherwise a crowded store can starve that gate.
    profiles = nsmallest(
        1, (item for item in eligible() if is_profile(item.record)), key=rank_key,
    )
    reserved = {item.record.record_id for item in profiles}
    count = 0

    def ordinary():
        nonlocal count
        for item in eligible():
            count += 1
            if item.record.record_id not in reserved:
                yield item

    candidates = nsmallest(query.candidate_limit, ordinary(), key=rank_key)
    candidates = profiles + candidates[:query.candidate_limit - len(profiles)]
    return RetrievalResult(
        memories=rank(candidates),
        unavailable=("memory_scan_incomplete",) if count > query.candidate_limit else (),
    )

InProcessLeases

InProcessLeases(*, single_process: bool, now: Any = None)

A lease table in this process's memory. Single-process use only.

Correct for a dev server, a test, and a deployment that genuinely runs one worker -- and silently wrong for every other, which is why it takes single_process=True rather than defaulting to convenient. The scaffold composes the Postgres adapter; this one exists so a laptop does not need a database to run a cycle, and so the port has an implementation whose behaviour a test can pin without one.

now is injected for the same reason the schedule's is: a test that had to sleep through a TTL would be a slow test asserting a timeout.

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

def __init__(self, *, single_process: bool, now: Any = None) -> None:
    if not single_process:
        raise MemoryContractError(
            "InProcessLeases coordinates one process and this deployment "
            "did not declare itself to be one. Celery beat and a manual "
            "backfill are different processes, and an in-memory table "
            "excludes neither from the other -- so two Deep Sleeps would "
            "run over one scope while every test passed. Pass "
            "single_process=True to say you mean it, or compose the "
            "Postgres adapter."
        )
    import time

    self._now = now or time.monotonic
    self._held: dict[str, tuple[str, float]] = {}

hold_for_update async

hold_for_update(lease: Lease) -> bool

The same answer as :meth:holds, and correctly so.

One process, one event loop, and a batch this table's owner applies without awaiting anything that yields: there is no instant between the check and the write for a rival to occupy. The lock PostgresLeases needs exists because there the rival is another process.

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

async def hold_for_update(self, lease: Lease) -> bool:
    """The same answer as :meth:`holds`, and correctly so.

    One process, one event loop, and a batch this table's owner applies
    without awaiting anything that yields: there is no instant between the
    check and the write for a rival to occupy. The lock ``PostgresLeases``
    needs exists because *there* the rival is another process.
    """
    return await self.holds(lease)

JournalledGraph

JournalledGraph(durable: Any)

The graph backend a deployment composes once, for everything.

Wraps rather than subclasses: the backends are protocol implementations with surfaces this module has no business restating, and anything outside :data:GRAPH_MUTATIONS and :data:GRAPH_READS is forwarded exactly as it was.

Composed at the root rather than around the phases, so that everything a cycle writes through goes into the same journal -- including the layers. ProceduralLayer.store_skill writes through a GraphMemoryStore, and a journal wrapped around only the phase factory's graph= argument would have left the one mutation an adopter is most likely to notice, the promoted draft skill, durable on its own.

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

def __init__(self, durable: Any) -> None:
    # Never nested. A journal whose "durable" store is another journal
    # replays into that one instead of the database, so the batch defers
    # itself forever and nothing is ever written.
    self._durable = (
        durable.durable if isinstance(durable, JournalledGraph) else durable
    )

durable property

durable: Any

The store underneath. Named, so nobody has to reach for a private.

LearningPolicy dataclass

LearningPolicy(lookback_hours: int = 24, promotion_min_pattern_count: int = 3, promotion_recency_days: int = 30, promotion_max_drafts_per_run: int = 5, promotion_use_tool_calls_fallback: bool = False, promotion_promote_assistant_content: bool = False, phase_12_use_llm_extractor: bool = False, phase_12_llm_model: str = 'claude-haiku-4-5', phase_12_llm_max_episodes_per_run: int = 100, phase_12_llm_max_drafts_per_run: int = 5, episodic_summarization_max_entries: int = 100, episodic_summarization_batch_size: int = 50, phase1_spreading_weight: float = 0.5, synthetic_link_min_co_count: int = 2, enable_entity_linker: bool = False, entity_linker_extractor_kind: str = 'regex', entity_linker_min_mention_count: int = 2, entity_linker_max_episodics_per_run: int = 200, entity_linker_confidence_threshold: float = 0.5)

What a consolidation run should promote, summarise and link.

Frozen: a run reads its policy once, and one that could change mid-pass would produce a result nobody can reproduce.

as_kwargs

as_kwargs() -> dict[str, Any]

The keyword arguments a consolidation run takes.

One call site instead of nineteen. Every getattr(config, name, default) it replaces was a place the framework's default and the template's copy could drift apart with nothing to notice.

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

def as_kwargs(self) -> dict[str, Any]:
    """The keyword arguments a consolidation run takes.

    One call site instead of nineteen. Every
    ``getattr(config, name, default)`` it replaces was a place the
    framework's default and the template's copy could drift apart with
    nothing to notice.
    """
    return asdict(self)

as_phase_kwargs

as_phase_kwargs() -> dict[str, Any]

The same knobs, named as the phase factories name them.

A translation table rather than a rename, because the two vocabularies are genuinely different: the shipped consolidator takes nineteen keyword arguments on one constructor, and the factories take them where the phase that reads each one is built. Written once, here, next to the fields it maps -- a worker doing this inline would be a second copy of every default in the place this class exists to remove them from.

enable_entity_linker and phase_12_use_llm_extractor are absent on purpose: in the factories a phase runs when it was given the collaborator it needs, so "enabled" is not a flag but the presence of an extractor. A deployment that set the flag and composed nothing would otherwise have a phase that reports zero rather than declining.

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

def as_phase_kwargs(self) -> dict[str, Any]:
    """The same knobs, named as the phase factories name them.

    A translation table rather than a rename, because the two vocabularies
    are genuinely different: the shipped consolidator takes nineteen
    keyword arguments on one constructor, and the factories take them where
    the phase that reads each one is built. Written once, here, next to the
    fields it maps -- a worker doing this inline would be a second copy of
    every default in the place this class exists to remove them from.

    ``enable_entity_linker`` and ``phase_12_use_llm_extractor`` are absent
    on purpose: in the factories a phase runs when it was given the
    collaborator it needs, so "enabled" is not a flag but the presence of
    an extractor. A deployment that set the flag and composed nothing would
    otherwise have a phase that reports zero rather than declining.
    """
    return {
        "lookback_hours": float(self.lookback_hours),
        "spreading_weight": self.phase1_spreading_weight,
        "episodic_max_entries": self.episodic_summarization_max_entries,
        "episodic_summarize_batch": self.episodic_summarization_batch_size,
        "entity_min_mention_count": self.entity_linker_min_mention_count,
        "entity_max_episodics_per_run": self.entity_linker_max_episodics_per_run,
        "entity_confidence_threshold": self.entity_linker_confidence_threshold,
        "procedural_model_name": self.phase_12_llm_model,
    }

from_settings classmethod

from_settings(settings: Any) -> LearningPolicy

Read a deployment's own settings object, falling back per field.

Absent names take the shipped default rather than raising: a settings object carries what that deployment chose to configure and nothing else, and requiring all nineteen would put a second copy of every default back in the place this removes it from.

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

@classmethod
def from_settings(cls, settings: Any) -> LearningPolicy:
    """Read a deployment's own settings object, falling back per field.

    Absent names take the shipped default rather than raising: a settings
    object carries what that deployment chose to configure and nothing
    else, and requiring all nineteen would put a second copy of every
    default back in the place this removes it from.
    """
    supplied = {
        field.name: getattr(settings, field.name)
        for field in fields(cls)
        if hasattr(settings, field.name)
        # Pydantic settings commonly use ``None`` to mean "not
        # configured".  Passing it through replaces this value object's
        # typed default and fails later in validation (or arithmetic),
        # which made a stock generated worker unable to start DEEP.
        and getattr(settings, field.name) is not None
    }
    # The scaffold's environment spelling predates this value object's
    # explicit ``_kind`` suffix.  It is the same choice, not a second
    # setting, so translate it at the boundary.
    if "entity_linker_extractor_kind" not in supplied:
        alias = getattr(settings, "entity_linker_extractor", None)
        if alias is not None:
            supplied["entity_linker_extractor_kind"] = alias
    return cls(**supplied)

Lease dataclass

Lease(scope_path: str, owner: str, ttl_seconds: float)

One scope, held by one owner, until one deadline.

LeaseLost

Bases: MemoryContractError

This cycle no longer holds its scope, and stopped before writing.

Its own type because the three answers a cycle can end with are three different facts. clean says the work happened. A phase error says the work was attempted and broke. This says the work was abandoned -- another holder has the scope, is doing it properly, and everything this cycle already wrote is that holder's problem rather than a partial result anybody should read.

LeasePort

Bases: Protocol

Exclusion for one scope, across whatever processes serve it.

acquire async

acquire(scope: MemoryScope, *, owner: str, ttl_seconds: float) -> Lease | None

Take the lease for scope, or None if someone else holds it.

None is the loser's answer and it is not an error: the scope is being consolidated right now, by somebody, which is the outcome asked for. An expired lease is available -- that is what makes a dead worker a delay rather than an outage.

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

async def acquire(self, scope: MemoryScope, *, owner: str, ttl_seconds: float) -> Lease | None:
    """Take the lease for ``scope``, or ``None`` if someone else holds it.

    ``None`` is the loser's answer and it is not an error: the scope is
    being consolidated right now, by somebody, which is the outcome asked
    for. An expired lease is available -- that is what makes a dead worker
    a delay rather than an outage.
    """
    ...

hold_for_update async

hold_for_update(lease: Lease) -> bool

Whether lease is ours, and keep it ours until the transaction ends.

The commit-time fence, and a different question from :meth:holds. holds answers about the instant it ran: a caller that then writes has a window in which the lease can lapse and a rival can take it, and running both on one connection does not close it -- the window is between the check and the commit. This one takes a lock the rival's acquisition must wait for, so the answer is still true when the batch lands.

Only meaningful inside a transaction. An implementation with no transactions to speak of may return :meth:holds, but a deployment composed on one is refused before a cycle runs rather than told afterwards that "atomic" meant something weaker.

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

async def hold_for_update(self, lease: Lease) -> bool:
    """Whether ``lease`` is ours, and keep it ours until the transaction ends.

    The commit-time fence, and a different question from :meth:`holds`.
    ``holds`` answers about the instant it ran: a caller that then writes
    has a window in which the lease can lapse and a rival can take it, and
    running both on one connection does not close it -- the window is
    between the check and the commit. This one takes a lock the rival's
    acquisition must wait for, so the answer is still true when the batch
    lands.

    Only meaningful inside a transaction. An implementation with no
    transactions to speak of may return :meth:`holds`, but a deployment
    composed on one is refused before a cycle runs rather than told
    afterwards that "atomic" meant something weaker.
    """
    ...

holds async

holds(lease: Lease) -> bool

Whether lease is still this owner's, right now.

The fence. A holder calls it before a mutation it cannot take back, so a worker whose lease expired mid-cycle stops rather than writing under an authority it lost.

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

async def holds(self, lease: Lease) -> bool:
    """Whether ``lease`` is still this owner's, right now.

    The fence. A holder calls it before a mutation it cannot take back, so
    a worker whose lease expired mid-cycle stops rather than writing under
    an authority it lost.
    """
    ...

release async

release(lease: Lease) -> bool

Give up lease. False when it was not this owner's to give.

Refused rather than applied, because by the time a slow worker reaches its finally the lease may belong to whoever took over -- and releasing it there would hand a third worker a scope two are already writing to.

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

async def release(self, lease: Lease) -> bool:
    """Give up ``lease``. ``False`` when it was not this owner's to give.

    Refused rather than applied, because by the time a slow worker reaches
    its ``finally`` the lease may belong to whoever took over -- and
    releasing it there would hand a third worker a scope two are already
    writing to.
    """
    ...

renew async

renew(lease: Lease) -> bool

Push lease's deadline out by its own TTL. False if it lapsed.

Owner-checked like :meth:release: a worker whose scope was taken over must not extend the deadline of whoever holds it now. False is the answer that says "you no longer have this", and a heartbeat that gets it should stop rather than retry -- the scope is somebody else's.

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

async def renew(self, lease: Lease) -> bool:
    """Push ``lease``'s deadline out by its own TTL. ``False`` if it lapsed.

    Owner-checked like :meth:`release`: a worker whose scope was taken over
    must not extend the deadline of whoever holds it now. ``False`` is the
    answer that says "you no longer have this", and a heartbeat that gets
    it should stop rather than retry -- the scope is somebody else's.
    """
    ...

LifecycleReceipt dataclass

LifecycleReceipt(scope_path: str, committed: tuple[str, ...] = (), discarded: tuple[str, ...] = (), degraded: bool = False)

What a flush or a forget did, and to which scope.

MemoryAdminService

MemoryAdminService(store: Any, *, procedural: Any = None)

Read, resolve and erase one scope's memories.

One service per process, not per scope: the scope is an argument to every method, because an administrative caller acts on scopes it is authorised for rather than on the one it was built with. That is the opposite of :class:MemoryCapability, which IS a scope -- and the difference is real: a capability serves one agent's turns, this serves an operator's request.

Parameters:

Name	Type	Description	Default
store	Any	an HMS satisfying the retrieval, write and lifecycle ports. Held privately and never returned: a service that exposed it would be a slower way to reach the layers.	required

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

def __init__(self, store: Any, *, procedural: Any = None) -> None:
    """
    Args:
        store: an HMS satisfying the retrieval, write and lifecycle ports.
            Held privately and never returned: a service that exposed it
            would be a slower way to reach the layers.
    """
    self._store = store
    self._procedural = procedural

procedural_review_available property

procedural_review_available: bool

Whether this service was composed with the procedural review door.

approve_procedure async

approve_procedure(scope: MemoryScope, procedure: Any, *, action_tool: str | None = None, precondition: Any = None) -> Any

Approve one draft, and record what a reviewer decided it governs.

Approval is the review, so this is where a tool and a precondition are named. That is not a convenience: the offline extractor reads what a scope did and has no way to know which registered tool a narrated action corresponds to, or what state must hold before repeating it. A draft it wrote governs nothing until a person says what it governs -- which is what makes "draft" the right status for it and human review the quality gate rather than a formality.

Both arguments are optional, so approving a procedure that only informs the prompt stays one call.

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

async def approve_procedure(
    self,
    scope: MemoryScope,
    procedure: Any,
    *,
    action_tool: str | None = None,
    precondition: Any = None,
) -> Any:
    """Approve one draft, and record what a reviewer decided it governs.

    Approval is the review, so this is where a tool and a precondition are
    named. That is not a convenience: the offline extractor reads *what a
    scope did* and has no way to know which registered tool a narrated
    action corresponds to, or what state must hold before repeating it. A
    draft it wrote governs nothing until a person says what it governs --
    which is what makes "draft" the right status for it and human review
    the quality gate rather than a formality.

    Both arguments are optional, so approving a procedure that only informs
    the prompt stays one call.
    """
    layer = self._require_procedural("approve a procedure")
    node_id = _node_id(procedure)
    if action_tool is not None or precondition is not None:
        await self._amend(
            scope,
            node_id,
            action_tool=action_tool,
            precondition=precondition,
        )
    return await layer.approve_skill(_legacy(scope), node_id)

correction async

correction(scope: MemoryScope, record_id: str, text: str, fields: Mapping[str, Any], *, salience: float = 0.9) -> MemoryRecord

Record that a person corrected their own profile. Staged, not published.

The named door for user_manual_edit. Phase 5 promotes a memory carrying that authority onto the scope's profile on the next nap, so the value is a grant rather than a description -- and a grant an extractor could mint by putting a string in its metadata bag would be no grant at all. MemoryRecord refuses it from metadata and accepts it only on its own field; this is where a deployment sets that field without writing the vocabulary out by hand.

Staged rather than published for the same reason every other write here is: a correction that failed halfway should leave nothing, and publish is the step that makes it retrievable.

Returns the record so a caller can name it in a receipt or a log.

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

async def correction(
    self,
    scope: MemoryScope,
    record_id: str,
    text: str,
    fields: Mapping[str, Any],
    *,
    salience: float = 0.9,
) -> MemoryRecord:
    """Record that a person corrected their own profile. Staged, not published.

    The named door for ``user_manual_edit``. Phase 5 promotes a memory
    carrying that authority onto the scope's profile on the next nap, so
    the value is a grant rather than a description -- and a grant an
    extractor could mint by putting a string in its metadata bag would be
    no grant at all. ``MemoryRecord`` refuses it from ``metadata`` and
    accepts it only on its own field; this is where a deployment sets that
    field without writing the vocabulary out by hand.

    Staged rather than published for the same reason every other write
    here is: a correction that failed halfway should leave nothing, and
    ``publish`` is the step that makes it retrievable.

    Returns the record so a caller can name it in a receipt or a log.
    """
    record = MemoryRecord(
        record_id=record_id,
        layer=MemoryLayer.SEMANTIC,
        text=text,
        scope_path=scope.path,
        salience=salience,
        origin="user-correction",
        # The producer's own vocabulary -- which fields, and what they now
        # say. Nested where every producer's metadata goes; the authority
        # above is the only thing that travels at the top.
        metadata=dict(fields),
        edited_by="user_manual_edit",
    )
    await self.stage(scope, (record,))
    return record

delete_record async

delete_record(scope: MemoryScope, record_id: str)

Exact-owner storage operation; platform callers must audit first.

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

async def delete_record(self, scope: MemoryScope, record_id: str):
    """Exact-owner storage operation; platform callers must audit first."""
    from symfonic.capabilities.memory.record_access import operation
    return await operation(self._store, "delete_record")(scope, record_id)

discard async

discard(scope: MemoryScope) -> Any

Drop staging without committing any of it.

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

async def discard(self, scope: MemoryScope) -> Any:
    """Drop staging without committing any of it."""
    return await self._store.discard(scope)

forget async

forget(scope: MemoryScope) -> Any

Erase scope and everything below it, staged and committed alike.

The operation a deletion request needs, and deliberately not discard: a privacy request that had to go through discard would depend on whether a turn happened to have finished. Idempotent -- an empty scope answers with an empty receipt rather than raising, because "there was nothing to erase" and "the erasure failed" must not look the same to a caller acting on a deletion request.

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

async def forget(self, scope: MemoryScope) -> Any:
    """Erase ``scope`` and everything below it, staged and committed alike.

    The operation a deletion request needs, and deliberately not
    ``discard``: a privacy request that had to go through ``discard`` would
    depend on whether a turn happened to have finished. Idempotent -- an
    empty scope answers with an empty receipt rather than raising, because
    "there was nothing to erase" and "the erasure failed" must not look the
    same to a caller acting on a deletion request.
    """
    return await self._store.forget(scope)

get_record async

get_record(scope: MemoryScope, record_id: str)

Direct published-record lookup, without a retrieval-page ceiling.

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

async def get_record(self, scope: MemoryScope, record_id: str):
    """Direct published-record lookup, without a retrieval-page ceiling."""
    from symfonic.capabilities.memory.record_access import operation
    return await operation(self._store, "get_record")(scope, record_id)

inventory_page async

inventory_page(scope: MemoryScope, *, layer=None, limit=200, cursor=None)

Published records in ID order; continuation is independent of recall.

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

async def inventory_page(self, scope: MemoryScope, *, layer=None, limit=200, cursor=None):
    """Published records in ID order; continuation is independent of recall."""
    from symfonic.capabilities.memory.inventory import page
    return await page(self._store, scope, layer=layer, limit=limit, cursor=cursor)

procedures async

procedures(scope: MemoryScope, *, drafts: bool = True) -> tuple[Any, ...]

What this scope has learned, for a reviewer to read.

Drafts included by default: this is the review queue, and a queue that hid what was waiting on review would be a queue with nothing in it.

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

async def procedures(
    self, scope: MemoryScope, *, drafts: bool = True
) -> tuple[Any, ...]:
    """What this scope has learned, for a reviewer to read.

    Drafts included by default: this is the review queue, and a queue that
    hid what was waiting on review would be a queue with nothing in it.
    """
    layer = self._require_procedural("list this scope's procedures")
    return tuple(
        await layer.query_skills(
            _legacy(scope), "", top_k=200, include_drafts=drafts
        )
    )

publish async

publish(scope: MemoryScope) -> Any

Commit what scope staged, and what anything below it staged.

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

async def publish(self, scope: MemoryScope) -> Any:
    """Commit what ``scope`` staged, and what anything below it staged."""
    return await self._store.flush(scope)

record_page async

record_page(scope: MemoryScope, *, layer: MemoryLayer | None = None, limit: int = DEFAULT_PAGE) -> RetrievalResult

Bounded admin records with explicit drops and incomplete-scan signals.

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

async def record_page(
    self, scope: MemoryScope, *, layer: MemoryLayer | None = None, limit: int = DEFAULT_PAGE
) -> RetrievalResult:
    """Bounded admin records with explicit drops and incomplete-scan signals."""
    layers = RETRIEVABLE_LAYERS if layer is None else frozenset({layer})
    query = MemoryQuery(
        scope=scope,
        layers=layers,
        limit=limit,
        candidate_limit=limit,
        max_record_chars=_ADMIN_RECORD_CHARS,
        max_total_chars=(_ADMIN_RECORD_CHARS + 32) * limit,
    )
    result: RetrievalResult = await self._store.retrieve(query)
    return result

records async

records(scope: MemoryScope, *, layer: MemoryLayer | None = None, limit: int = DEFAULT_PAGE) -> tuple[MemoryRecord, ...]

Committed memories scope may read: its own and its ANCESTORS'.

Reading widens upward, never downward; erasure walks the opposite direction. layer narrows that visibility and cannot grant access. Use record_page when omission/completeness evidence is required.

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

async def records(
    self,
    scope: MemoryScope,
    *,
    layer: MemoryLayer | None = None,
    limit: int = DEFAULT_PAGE,
) -> tuple[MemoryRecord, ...]:
    """Committed memories ``scope`` may read: its own and its ANCESTORS'.

    Reading widens upward, never downward; erasure walks the opposite
    direction. ``layer`` narrows that visibility and cannot grant access.
    Use ``record_page`` when omission/completeness evidence is required.
    """
    result = await self.record_page(scope, layer=layer, limit=limit)
    return tuple(memory.record for memory in result.memories)

reject_procedure async

reject_procedure(scope: MemoryScope, procedure: Any) -> Any

Reject one draft. It stays readable and stops being active.

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

async def reject_procedure(self, scope: MemoryScope, procedure: Any) -> Any:
    """Reject one draft. It stays readable and stops being active."""
    layer = self._require_procedural("reject a procedure")
    return await layer.reject_skill(_legacy(scope), _node_id(procedure))

stage async

stage(scope: MemoryScope, records: tuple[MemoryRecord, ...]) -> Any

Prepare records under scope. Nothing is durable until publish.

Exposed because an administrative import is a real operation and the alternative is a caller writing straight to the store, which is the access this service exists to replace.

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

async def stage(
    self, scope: MemoryScope, records: tuple[MemoryRecord, ...]
) -> Any:
    """Prepare ``records`` under ``scope``. Nothing is durable until publish.

    Exposed because an administrative import is a real operation and the
    alternative is a caller writing straight to the store, which is the
    access this service exists to replace.
    """
    return await self._store.write(WriteRequest(scope=scope, records=records))

MemoryBundleFactory

MemoryBundleFactory(store: Any, *, limit: int = 5, recall_budget: RecallBudget | None = None, extractor: MemoryExtractorPort | None = None, consolidation: Any | None = None, conversation: Any | None = None, recent_turns: int = 0, activation: Any | 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
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 cannot serve the 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,
    consolidation: Any | None = None,
    conversation: Any | None = None,
    recent_turns: int = 0,
    activation: Any | 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`.
        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`` cannot serve the port.
    """
    validate_extractor(extractor)
    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._consolidation = consolidation
    self._conversation = conversation
    self._recent_turns = recent_turns
    self._activation = activation

for_scope

for_scope(scope: MemoryScope | Any) -> MemoryCapability

The capability that recalls, records and erases for scope.

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

def for_scope(self, scope: MemoryScope | Any) -> MemoryCapability:
    """The capability that recalls, records and erases for ``scope``."""
    memory_scope = as_memory_scope(scope)
    hydrator = hydrator_for(
        self._store,
        conversation=self._conversation,
        recent_turns=self._recent_turns,
        activation=self._activation,
    )
    return MemoryCapability(
        hydrator,
        scope=memory_scope,
        limit=self._limit,
        recall_budget=self._recall_budget,
        writer=self._store,
        # 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=self._store,
        extractor=self._extractor,
        consolidation=self._consolidation,
    )

MemoryContractError

Bases: MemoryCapabilityError

A declaration or a call is impossible as stated.

Raised at declaration time wherever the shape is knowable then — a scope, a record, and a query all validate at construction — so a misconfigured deployment fails before its first turn rather than mid-invocation.

MemoryContribution dataclass

MemoryContribution(contribution_id: str, source: MemoryContextSource, capability: str = 'memory', layer: ContributionLayer = ContributionLayer.L2, tier: ContributionTier = ContributionTier.SESSION, scope: ContributionScope = ContributionScope.DEPLOYMENT, order: int = 0, inherit: bool = True, pinned: bool = False, requires_hydration: bool = True)

The bridge's declaration of the recall block it contributes.

validate

validate() -> None

Refuse every declaration this capability is not allowed to make.

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

def validate(self) -> None:
    """Refuse every declaration this capability is not allowed to make."""
    if not self.contribution_id:
        raise MemoryContractError(
            "a memory contribution must declare a non-empty contribution_id."
        )
    if not _ID_CHARSET.match(self.contribution_id):
        raise MemoryContractError(
            f"contribution id {self.contribution_id!r} is outside the permitted charset "
            "[A-Za-z0-9_.-]; ids appear in isolation keys and in the rendered untrusted "
            "delimiter, where a separator or an angle bracket forges a boundary."
        )
    if not callable(getattr(self.source, "read", None)):
        raise MemoryContractError(
            f"contribution {self.contribution_id!r} declares a source that cannot read: "
            f"{type(self.source).__name__} has no callable read()."
        )
    if self.tier in AUTHORED_TIERS:
        raise MemoryContractError(
            f"contribution {self.contribution_id!r} declares the authored tier "
            f"{self.tier.value!r}. A recall is aggregated from what a user said; an "
            "authored tier renders it verbatim beside the operator's own instructions."
        )
    if self.layer is ContributionLayer.L0:
        raise MemoryContractError(
            f"contribution {self.contribution_id!r} declares layer L0. L0 is the cached "
            "authored prefix: a recall placed there is served back to the model on every "
            "later turn of the session, so one poisoned memory outlives the turn that "
            "retrieved it. Declare L1 (standing) or L2 (per turn)."
        )
    if self.pinned:
        raise MemoryContractError(
            f"contribution {self.contribution_id!r} declares itself pinned. Pinned content "
            "fails the compile rather than being dropped for budget; a recall is context, "
            "not instruction, and a turn without it is still a correct turn."
        )
    if self.scope is not ContributionScope.DEPLOYMENT and not getattr(
        self.source, "scope_aware", False
    ):
        raise MemoryContractError(
            f"contribution {self.contribution_id!r} declares scope={self.scope.value!r} "
            f"but its source {type(self.source).__name__} is not scope_aware: it serves "
            "one value for every tenant."
        )

MemoryExtractionService

MemoryExtractionService(model: ExtractionModelPort | None = None, *, scrubber: CredentialScrubber | None = None)

Extracts candidate memories from a finished turn.

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

def __init__(
    self,
    model: ExtractionModelPort | None = None,
    *,
    scrubber: CredentialScrubber | None = None,
) -> None:
    self._model = model
    self._scrubber = scrubber if scrubber is not None else CredentialScrubber()

extract async

extract(request: ExtractionRequest) -> ExtractionResult

Ask the model, read the reply, and mint what survives the policy.

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

async def extract(self, request: ExtractionRequest) -> ExtractionResult:
    """Ask the model, read the reply, and mint what survives the policy."""
    prompt, redactions = self.prompt(request)
    if self._model is None:
        return ExtractionResult(
            scope=request.scope,
            turn=request.turn,
            redactions=redactions,
            reason="no extraction model is bound; this deployment does not extract",
        )
    try:
        reply = read_reply(await self._model.complete(prompt))
    except MemoryCapabilityError:
        raise
    except Exception as exc:  # noqa: BLE001 - a failed extraction is not a failed turn
        return ExtractionResult(
            scope=request.scope,
            turn=request.turn,
            redactions=redactions,
            degraded=True,
            reason=f"extraction model failed: {type(exc).__name__}",
        )
    return self.parse(reply, request, redactions=redactions)

parse

parse(reply: ProviderReply, request: ExtractionRequest, *, redactions: tuple[str, ...] = ()) -> ExtractionResult

Turn a read reply into records. Pure, so a corpus can replay it.

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

def parse(
    self,
    reply: ProviderReply,
    request: ExtractionRequest,
    *,
    redactions: tuple[str, ...] = (),
) -> ExtractionResult:
    """Turn a read reply into records. Pure, so a corpus can replay it."""
    if not reply.readable:
        return ExtractionResult(
            scope=request.scope,
            turn=request.turn,
            family=reply.family,
            redactions=redactions,
            reason=(
                "the model reply is unreadable: no known provider family "
                "matched its shape, and stringifying it would mine the transport"
            ),
        )
    payload = json_payload(reply.text)
    if payload is None:
        return ExtractionResult(
            scope=request.scope,
            turn=request.turn,
            family=reply.family,
            redactions=redactions,
            reason="the model reply carried no JSON object",
        )
    return self._mint(payload, request, reply.family, redactions)

prompt

prompt(request: ExtractionRequest) -> tuple[str, tuple[str, ...]]

The extraction prompt for request, with credentials removed.

Scrubbing here is not belt-and-braces for the store: a secret handed to a provider is disclosed whether or not anyone ever writes it down.

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

def prompt(self, request: ExtractionRequest) -> tuple[str, tuple[str, ...]]:
    """The extraction prompt for ``request``, with credentials removed.

    Scrubbing here is not belt-and-braces for the store: a secret handed to
    a provider is disclosed whether or not anyone ever writes it down.
    """
    user = self._scrubber.scrub_text(request.user_message)
    assistant = self._scrubber.scrub_text(request.assistant_message)
    text = EXTRACTION_PROMPT.format(
        user_message=user.text, assistant_message=assistant.text
    )
    return text, user.redactions + assistant.redactions

MemoryExtractorPort

Bases: Protocol

Turns one finished exchange into memories worth keeping.

The capability calls this after the final model round and writes whatever comes back. :class:~.extraction.MemoryExtractionService is the shipped implementation; a deployment with its own extraction policy implements this instead.

A failed extraction is not a failed turn. An implementation that cannot reach its model returns a result with degraded=True and a reason rather than raising: the exchange already happened and the user already has an answer, and losing it because a side effect failed is the worse outcome. Raising is reserved for being asked something impossible.

extract async

extract(request: ExtractionRequest) -> ExtractionResult

Read request's exchange and mint the memories it justifies.

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

async def extract(self, request: ExtractionRequest) -> ExtractionResult:
    """Read ``request``'s exchange and mint the memories it justifies."""
    ...

MemoryLayer

Bases: StrEnum

The five layers of the Pentad memory model.

EPISODIC class-attribute instance-attribute

EPISODIC = 'episodic'

Narrative events, scenarios, and timestamps (When/Where).

PROCEDURAL class-attribute instance-attribute

PROCEDURAL = 'procedural'

Skills, code snippets, and workflows (How).

PROSPECTIVE class-attribute instance-attribute

PROSPECTIVE = 'prospective'

Commitments, reminders, and pending tasks (Future).

SEMANTIC class-attribute instance-attribute

SEMANTIC = 'semantic'

Permanent facts and graph entities (What).

WORKING class-attribute instance-attribute

WORKING = 'working'

Active conversation context, session-scoped (Now).

MemoryLifecyclePort

Bases: Protocol

Commits and erases. The finalize seam, and the privacy seam.

flush async

flush(scope: MemoryScope) -> LifecycleReceipt

Commit every pending memory in scope and its descendants.

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

async def flush(self, scope: MemoryScope) -> LifecycleReceipt:
    """Commit every pending memory in ``scope`` and its descendants."""
    ...

forget async

forget(scope: MemoryScope) -> LifecycleReceipt

Erase every memory in scope and its descendants, pending or not.

Scoped by construction, so a deletion request can never reach outside the subtree it named (SEC-PRIV). Idempotent: forgetting an already-empty scope reports an empty receipt rather than failing.

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

async def forget(self, scope: MemoryScope) -> LifecycleReceipt:
    """Erase every memory in ``scope`` and its descendants, pending or not.

    Scoped by construction, so a deletion request can never reach outside
    the subtree it named (SEC-PRIV). Idempotent: forgetting an already-empty
    scope reports an empty receipt rather than failing.
    """
    ...

MemoryQuery dataclass

MemoryQuery(scope: MemoryScope, cue: str = '', limit: int = 5, layers: frozenset[MemoryLayer] = RETRIEVABLE_LAYERS, turn: int = 0, session_id: str = '', max_record_chars: int = 250, max_total_chars: int = DEFAULT_BLOCK_CHARS, candidate_limit: int = DEFAULT_CANDIDATE_LIMIT, recall_budget: RecallBudget | None = None)

One retrieval: where to look, what to look for, and how much may return.

validate

validate() -> None

Refuse a query whose ceilings cannot admit anything.

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

def validate(self) -> None:
    """Refuse a query whose ceilings cannot admit anything."""
    if self.recall_budget is not None:
        from symfonic.capabilities.memory.budget import RecallBudget
        if not isinstance(self.recall_budget, RecallBudget):
            raise MemoryContractError("recall_budget must be a RecallBudget")
    if self.limit < 1:
        raise MemoryContractError(
            f"query declares limit {self.limit}; a retrieval that may return nothing is "
            "spelled by not retrieving, not by asking for zero memories."
        )
    cap = self.candidate_limit
    if type(cap) is not int or not 1 <= cap <= MAX_CANDIDATE_LIMIT:
        raise MemoryContractError(
            f"query declares candidate_limit {self.candidate_limit}; backend scans "
            f"must be between 1 and {MAX_CANDIDATE_LIMIT} rows."
        )
    if not self.layers:
        raise MemoryContractError(
            "query declares an empty layer set. Narrowing to no layer is a query that "
            "cannot match, which reads downstream as an empty memory rather than a bug."
        )
    if self.max_record_chars < 1 or self.max_total_chars < 1:
        raise MemoryContractError(
            f"query declares ceilings ({self.max_record_chars}, {self.max_total_chars}); "
            "both bound rendered characters and must be positive."
        )
    if self.max_record_chars > self.max_total_chars:
        raise MemoryContractError(
            f"query allows {self.max_record_chars} chars per memory inside a "
            f"{self.max_total_chars}-char block. A memory that fits the per-item ceiling "
            "and can never fit the block is dropped twice for two different reasons."
        )

MemoryRead dataclass

MemoryRead(text: str, revision: str = '', untrusted: bool = True)

What a memory source answered.

untrusted defaults to True — the inverse of the general prompt contract's default, and the whole point of a separate value type. A recall is aggregated from what a user said; a source here would have to remember to declare it trusted, which nothing in this capability ever does.

MemoryRecord dataclass

MemoryRecord(record_id: str, layer: MemoryLayer, text: str, scope_path: str, salience: float = 0.5, origin: str = '', revision: str = '', metadata: Mapping[str, Any] = dict(), edited_by: str = '')

One memory, as every port in this capability moves it.

scope property

scope: MemoryScope

The scope this memory was written at.

validate

validate() -> None

Refuse a record no store should be asked to hold.

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

def validate(self) -> None:
    """Refuse a record no store should be asked to hold."""
    if not self.record_id:
        raise MemoryContractError("a memory record must declare a non-empty record_id.")
    if not RECORD_ID_CHARSET.match(self.record_id):
        raise MemoryContractError(
            f"record id {self.record_id!r} is outside the permitted charset "
            "[A-Za-z0-9_.:-]; ids appear in receipts and drop reasons, which are rendered."
        )
    if not self.text.strip():
        raise MemoryContractError(
            f"memory {self.record_id!r} carries no text. An empty memory costs a retrieval "
            "slot and renders as nothing, which is indistinguishable from a lost one."
        )
    if not 0.0 <= self.salience <= 1.0:
        raise MemoryContractError(
            f"memory {self.record_id!r} declares salience {self.salience}; salience is a "
            "weight in [0, 1], and a value outside it silently dominates every ranking."
        )
    # Raises MemoryContractError on a malformed path — the record's scope is
    # the isolation key, so an unparseable one is refused at construction.
    scope_from_path(self.scope_path)
    validate_edit_authority(self.edited_by)

MemoryRequest dataclass

MemoryRequest(contribution_id: str, scope_path: str = '', turn: int = 0)

What a source is asked for: one contribution, one scope, one turn.

MemoryRetrievalPort

Bases: Protocol

Reads memories visible from one scope. The prompt/input seam.

retrieve async

retrieve(query: MemoryQuery) -> RetrievalResult

Return the memories visible at query.scope, ranked and capped.

Visibility is the adapter's obligation, not a courtesy: a memory whose scope does not cover query.scope must not appear in the result. The bridge re-checks it (:class:~.errors.ScopeViolation) because the adapter is exactly the component that might be wrong.

Raises :class:~.errors.MemoryUnavailable when the store is unreachable. Returning an empty result instead would be indistinguishable from a scope that genuinely remembers nothing.

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

async def retrieve(self, query: MemoryQuery) -> RetrievalResult:
    """Return the memories visible at ``query.scope``, ranked and capped.

    Visibility is the adapter's obligation, not a courtesy: a memory whose
    scope does not cover ``query.scope`` must not appear in the result. The
    bridge re-checks it (:class:`~.errors.ScopeViolation`) because the
    adapter is exactly the component that might be wrong.

    Raises :class:`~.errors.MemoryUnavailable` when the store is unreachable.
    Returning an empty result instead would be indistinguishable from a
    scope that genuinely remembers nothing.
    """
    ...

MemoryScope dataclass

MemoryScope(tenant: str, principal: str = '', session: str = '')

Where a memory lives: tenant, then principal, then session.

The levels are positional and gapless. A session without a principal is a hole in the hierarchy — it would compare as a child of the tenant while naming something the tenant cannot enumerate — so it is refused at construction rather than normalised into something plausible.

path property

path: str

The canonical string form. Stable, and safe to use as a store key.

segments property

segments: tuple[str, ...]

The populated levels, outermost first.

covers

covers(other: MemoryScope) -> bool

Whether a memory written at self is visible at other.

Segment-wise, never string-prefix: acme does not cover acmecorp, and a rule written with :meth:str.startswith would say it does.

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

def covers(self, other: MemoryScope) -> bool:
    """Whether a memory written at ``self`` is visible at ``other``.

    Segment-wise, never string-prefix: ``acme`` does not cover ``acmecorp``,
    and a rule written with :meth:`str.startswith` would say it does.
    """
    mine = self.segments
    theirs = other.segments
    return len(mine) <= len(theirs) and theirs[: len(mine)] == mine

distance

distance(other: MemoryScope) -> int

Levels from self down to other, or -1 when not covered.

-1 rather than an exception: distance is asked once per candidate during ranking, and "not visible from here" is an ordinary answer there.

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

def distance(self, other: MemoryScope) -> int:
    """Levels from ``self`` down to ``other``, or ``-1`` when not covered.

    ``-1`` rather than an exception: distance is asked once per candidate
    during ranking, and "not visible from here" is an ordinary answer there.
    """
    if not self.covers(other):
        return -1
    return len(other.segments) - len(self.segments)

validate

validate() -> None

Refuse every scope this capability cannot compare.

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

def validate(self) -> None:
    """Refuse every scope this capability cannot compare."""
    if not self.tenant:
        raise MemoryContractError(
            "a memory scope must name a tenant; an unscoped memory is a memory no "
            "backend can isolate (SEC-TEN-5)."
        )
    if self.session and not self.principal:
        raise MemoryContractError(
            f"scope declares session {self.session!r} with no principal. The levels are "
            "positional: a session under an anonymous principal would compare as a direct "
            "child of the tenant, making it visible to every other principal in it."
        )
    for level, value in (
        ("tenant", self.tenant),
        ("principal", self.principal),
        ("session", self.session),
    ):
        if value and not _SEGMENT_CHARSET.match(value):
            raise MemoryContractError(
                f"{level} segment {value!r} is outside the permitted charset "
                f"[A-Za-z0-9_.:@-]. {SCOPE_SEPARATOR!r} separates levels, so a segment "
                "containing one forges a level of the hierarchy."
            )

MemoryUnavailable

Bases: MemoryCapabilityError

The memory store could not be reached for this operation.

Adapters raise it; the bridge catches it and degrades. It is deliberately not a subclass of :class:MemoryContractError: the bridge's whole degradation rule is "transport degrades, contracts propagate", and a shared base would collapse the two.

MemoryWriteCoordinator

MemoryWriteCoordinator(*, writes: MemoryWritePort, lifecycle: MemoryLifecyclePort, background: BackgroundWorkPort | None = None, deadline_seconds: float | None = None)

Owns the write side of a turn: foreground, background, and flush.

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

def __init__(
    self,
    *,
    writes: MemoryWritePort,
    lifecycle: MemoryLifecyclePort,
    background: BackgroundWorkPort | None = None,
    deadline_seconds: float | None = None,
) -> None:
    self._writes = writes
    self._lifecycle = lifecycle
    self._background = background
    self._deadline = deadline_seconds
    self._pending: dict[str, _Pending] = {}
    self._tasks: list[tuple[WriteRequest, Any]] = []

in_flight property

in_flight: int

Background writes spawned and not yet joined.

lifecycle property

lifecycle: Any

What publishes a staged record, and so its transaction domain.

Published so the consolidation commit can establish that the records it flushes land in the same domain as the graph mutations applied beside them; without that the two halves could not be one commit.

pending_ids property

pending_ids: tuple[str, ...]

Every uncommitted record id this coordinator wrote, sorted.

pending_scopes property

pending_scopes: tuple[str, ...]

Scope paths holding written-but-uncommitted memories, sorted.

write_port property

write_port: MemoryWritePort

The participant that stages records, for transaction-domain validation.

abandon async

abandon(scope: MemoryScope) -> LifecycleReceipt

Give up on scope's uncommitted memories without committing them.

The turn's rollback. Nothing committed is touched — a forget would take the previous turns' memories along with this one's.

When the bound lifecycle port implements :class:MemoryDiscardPort the pending buffer is dropped at the store and the rollback is real. When it does not, the receipt comes back degraded: this coordinator will not commit those memories, but nothing stops another flush of the same scope from doing so, and saying otherwise would be a rollback that only exists in the caller's head.

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

async def abandon(self, scope: MemoryScope) -> LifecycleReceipt:
    """Give up on ``scope``'s uncommitted memories without committing them.

    The turn's rollback. Nothing *committed* is touched — a ``forget``
    would take the previous turns' memories along with this one's.

    When the bound lifecycle port implements :class:`MemoryDiscardPort` the
    pending buffer is dropped at the store and the rollback is real. When
    it does not, the receipt comes back ``degraded``: this coordinator will
    not commit those memories, but nothing stops another flush of the same
    scope from doing so, and saying otherwise would be a rollback that only
    exists in the caller's head.
    """
    await self.join()
    mine = tuple(
        sorted(
            record_id
            for path in self._covered_paths(scope)
            for record_id in self._pending[path].ids
        )
    )
    self._forget_covered(scope)
    if not isinstance(self._lifecycle, MemoryDiscardPort):
        return LifecycleReceipt(
            scope_path=scope.path, discarded=mine, degraded=True
        )
    try:
        return await self._lifecycle.discard(scope)
    except MemoryUnavailable:
        return LifecycleReceipt(
            scope_path=scope.path, discarded=mine, degraded=True
        )

flush async

flush(scope: MemoryScope, *, join: bool = True, required_ids: tuple[str, ...] = ()) -> LifecycleReceipt

Commit scope and everything below it.

join awaits this coordinator's in-flight background writes first, because a flush that overtakes its own write commits half a turn. required_ids rejects incomplete publication before forgetting pending bookkeeping; an atomic caller can then roll back the whole transaction.

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

async def flush(self, scope: MemoryScope, *, join: bool = True,
                required_ids: tuple[str, ...] = ()) -> LifecycleReceipt:
    """Commit ``scope`` and everything below it.

    ``join`` awaits this coordinator's in-flight background writes first,
    because a flush that overtakes its own write commits half a turn.
    ``required_ids`` rejects incomplete publication before forgetting pending
    bookkeeping; an atomic caller can then roll back the whole transaction.
    """
    if join:
        await self.join()
    try:
        receipt = await self._lifecycle.flush(scope)
    except MemoryUnavailable:
        return LifecycleReceipt(scope_path=scope.path, degraded=True)
    if not receipt.degraded:
        if not set(required_ids).issubset(receipt.committed):
            raise MemoryContractError("required buffered records were not published")
        self._forget_covered(scope)
    return receipt

join async

join() -> tuple[WriteOutcome, ...]

Await every background write spawned since the last join.

A write that raised is reported, not re-raised: the failure belongs to the write, and a flush that exploded because a memory did not land would end the turn over the thing that was supposed to be optional.

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

async def join(self) -> tuple[WriteOutcome, ...]:
    """Await every background write spawned since the last join.

    A write that raised is *reported*, not re-raised: the failure belongs
    to the write, and a flush that exploded because a memory did not land
    would end the turn over the thing that was supposed to be optional.
    """
    outcomes: list[WriteOutcome] = []
    spawned, self._tasks = self._tasks, []
    for request, task in spawned:
        outcomes.append(await self._settle(request, task))
    return tuple(outcomes)

write async

write(request: WriteRequest) -> WriteReceipt

Write request now, degrading rather than failing the turn.

A ScopeViolation is not caught: a tenant boundary crossing is not a degraded turn, and a write that quietly reported degraded for one would hide the single failure isolation exists to surface.

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

async def write(self, request: WriteRequest) -> WriteReceipt:
    """Write ``request`` now, degrading rather than failing the turn.

    A ``ScopeViolation`` is *not* caught: a tenant boundary crossing is not
    a degraded turn, and a write that quietly reported ``degraded`` for one
    would hide the single failure isolation exists to surface.
    """
    try:
        receipt = await self._writes.write(request)
    except MemoryUnavailable as exc:
        return WriteReceipt(
            rejected=tuple(
                (record.record_id, f"memory store unreachable: {exc}")
                for record in request.records
            ),
            degraded=True,
        )
    self._pending.setdefault(request.scope.path, _Pending()).add(receipt.accepted)
    return receipt

write_in_background

write_in_background(request: WriteRequest) -> Any

Spawn request as run-owned work, or refuse to spawn it at all.

There is no third path. A coordinator with no registry that fell back to asyncio.create_task would recreate the detached-set problem the registry exists to end, and it would do it invisibly.

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

def write_in_background(self, request: WriteRequest) -> Any:
    """Spawn ``request`` as run-owned work, or refuse to spawn it at all.

    There is no third path. A coordinator with no registry that fell back
    to ``asyncio.create_task`` would recreate the detached-set problem the
    registry exists to end, and it would do it invisibly.
    """
    if self._background is None:
        raise MemoryContractError(
            "a background memory write needs a background-work registry to own it; "
            "none is bound. Work with no owner is work no run waits for and no "
            "teardown reports (RCX-1/RCX-8) — bind one, or write in the foreground."
        )
    task = self._background.spawn(
        self._writing(request),
        owner=WRITE_OWNER,
        purpose=f"memory-write:{request.scope.path}",
        deadline_seconds=self._deadline,
    )
    self._tasks.append((request, task))
    return task

MemoryWritePort

Bases: Protocol

Records what a turn produced. The post-response seam.

write async

write(request: WriteRequest) -> WriteReceipt

Record request's memories as pending, and report per memory.

Idempotent on record_id within a scope: writing the same id twice upserts rather than duplicating, so a retried post-response stage does not double a memory.

Pending memories are not retrievable until :meth:MemoryLifecyclePort.flush.

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

async def write(self, request: WriteRequest) -> WriteReceipt:
    """Record ``request``'s memories as pending, and report per memory.

    Idempotent on ``record_id`` within a scope: writing the same id twice
    upserts rather than duplicating, so a retried post-response stage does
    not double a memory.

    Pending memories are not retrievable until :meth:`MemoryLifecyclePort.flush`.
    """
    ...

PhaseContext dataclass

PhaseContext(scope: MemoryScope, cycle: ConsolidationCycle, started_at: datetime, writes: MemoryWriteCoordinator | None = None, run_id: str = '', root_run_id: str = '', ledger: dict[str, int] = dict(), legacy: dict[str, int] = dict(), seen: set[str] = set())

What a phase is told about the cycle it is running inside.

Frozen, and the two mutable fields are the cycle's own accumulators rather than state a phase can rewrite: a phase adds to the ledger and names what it read, and cannot reach anything another phase decided.

contributed

contributed(counter: str, delta: int) -> None

Add to a legacy report counter this phase owns but cannot headline.

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

def contributed(self, counter: str, delta: int) -> None:
    """Add to a legacy report counter this phase owns but cannot headline."""
    self.legacy[counter] = self.legacy.get(counter, 0) + int(delta)

counted

counted(name: str, delta: int = 1) -> None

Add to a cycle counter. Integers and closed names only.

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

def counted(self, name: str, delta: int = 1) -> None:
    """Add to a cycle counter. Integers and closed names only."""
    merge_ledger(self.ledger, {name: delta})

examined

examined(record_ids: Iterable[Any]) -> None

Name what this phase read, for the cycle's candidates count.

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

def examined(self, record_ids: Iterable[Any]) -> None:
    """Name what this phase read, for the cycle's ``candidates`` count."""
    self.seen.update(str(record_id) for record_id in record_ids)

PromotionCandidate dataclass

PromotionCandidate(record: MemoryRecord, confidence: float = 0.0, durability: str = _PROMOTABLE_DURABILITY, conversation_id: str = '')

One memory considered for promotion, with the signals that decide it.

ProviderFamily

Bases: StrEnum

The reply shapes this capability knows how to read.

RetrievalResult dataclass

RetrievalResult(memories: tuple[RetrievedMemory, ...] = (), dropped: tuple[tuple[str, str], ...] = (), degraded: bool = False, sources: Mapping[str, int] = (lambda: EMPTY_SOURCES)(), unavailable: tuple[str, ...] = ())

What retrieval returned, and a reason for everything it left out.

render

render() -> str

One memory per line, in the order selection decided.

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

def render(self) -> str:
    """One memory per line, in the order selection decided."""
    return "\n".join(memory.line() for memory in self.memories)

revision

revision() -> str

A content-derived revision, so a changed recall changes the cache key.

Computed over ids and text: a store that rewrites a memory in place keeps its id, and a revision that ignored the text would report an unchanged prompt whose bytes had changed.

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

def revision(self) -> str:
    """A content-derived revision, so a changed recall changes the cache key.

    Computed over ids *and* text: a store that rewrites a memory in place
    keeps its id, and a revision that ignored the text would report an
    unchanged prompt whose bytes had changed.
    """
    material = "\x00".join(
        f"{m.record.record_id}\x01{m.line()}" for m in self.memories
    )
    return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16]

RetrievedMemory dataclass

RetrievedMemory(record: MemoryRecord, score: float | None = 0.0, scope_distance: int = 0, source_ordinal: int | None = None, reserved: bool = False)

One scored memory, with the distance it travelled to reach this scope.

line

line() -> str

The rendered form: layer prefix, single-line text.

The prefix is a delimiter, so the text is flattened before it is interpolated — otherwise one stored memory containing ok\n[semantic] forged renders as two memories, the second attributed to a layer nothing wrote it to.

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

def line(self) -> str:
    """The rendered form: layer prefix, single-line text.

    The prefix is a delimiter, so the text is flattened before it is
    interpolated — otherwise one stored memory containing
    ``ok\\n[semantic] forged`` renders as two memories, the second
    attributed to a layer nothing wrote it to.
    """
    from symfonic.capabilities.memory.rendering import line
    return line(self.record)

ScheduleCursor dataclass

ScheduleCursor(turns_since_quick: int = 0, last_quick_at: datetime | None = None, last_nightly_at: datetime | None = None, last_deep_at: datetime | None = None)

What has run so far. The persisted half of the schedule.

completed

completed(cycle: ConsolidationCycle, *, at: datetime) -> ScheduleCursor

Record cycle as run, satisfying every narrower cadence too.

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

def completed(self, cycle: ConsolidationCycle, *, at: datetime) -> ScheduleCursor:
    """Record ``cycle`` as run, satisfying every narrower cadence too."""
    updates: dict[str, Any] = {"turns_since_quick": 0, "last_quick_at": at}
    if cycle in (ConsolidationCycle.NIGHTLY, ConsolidationCycle.DEEP):
        updates["last_nightly_at"] = at
    if cycle is ConsolidationCycle.DEEP:
        updates["last_deep_at"] = at
    return replace(self, **updates)

from_state classmethod

from_state(state: dict[str, Any]) -> ScheduleCursor

Read a persisted cursor, tolerating one written by an older build.

Missing keys default rather than raise: a cursor written before Deep Sleep existed is a valid cursor with no deep run behind it, and refusing it would make a rollback forward-incompatible.

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

@classmethod
def from_state(cls, state: dict[str, Any]) -> ScheduleCursor:
    """Read a persisted cursor, tolerating one written by an older build.

    Missing keys default rather than raise: a cursor written before Deep
    Sleep existed is a valid cursor with no deep run behind it, and
    refusing it would make a rollback forward-incompatible.
    """
    return cls(
        turns_since_quick=int(state.get("turns_since_quick", 0) or 0),
        last_quick_at=_parse(state.get("last_quick_at")),
        last_nightly_at=_parse(state.get("last_nightly_at")),
        last_deep_at=_parse(state.get("last_deep_at")),
    )

to_state

to_state() -> dict[str, Any]

The persisted form: ISO-8601 timestamps, like every other record.

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

def to_state(self) -> dict[str, Any]:
    """The persisted form: ISO-8601 timestamps, like every other record."""
    return {
        "turns_since_quick": self.turns_since_quick,
        "last_quick_at": _iso(self.last_quick_at),
        "last_nightly_at": _iso(self.last_nightly_at),
        "last_deep_at": _iso(self.last_deep_at),
    }

turn

turn() -> ScheduleCursor

Advance by one completed turn.

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

def turn(self) -> ScheduleCursor:
    """Advance by one completed turn."""
    return replace(self, turns_since_quick=self.turns_since_quick + 1)

ScopeViolation

Bases: MemoryCapabilityError

A memory crossed a scope boundary it is not visible across.

Raised at the port boundary — on what an adapter returned, not only on what a caller asked for. A backend enforces isolation (SEC-TEN-5); the bridge verifies it, because a bridge that trusts the backend has no answer when the backend is the thing that is wrong.

ScrubResult dataclass

ScrubResult(text: str, redactions: tuple[str, ...] = ())

Text with its credentials replaced, and what was replaced.

clean property

clean: bool

Whether the input carried no credential-shaped value.

SpreadingActivation dataclass

SpreadingActivation(source: AssociationSource, max_hops: int = 1, decay: float = 0.5, max_nodes: int = 10)

Expands a recall through the association graph, with decay and a cap.

expand async

expand(scope: MemoryScope, seeds: Sequence[RetrievedMemory]) -> tuple[tuple[RetrievedMemory, ...], ActivationLog]

Walk out from seeds and return what lit up, plus the provenance.

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

async def expand(
    self, scope: MemoryScope, seeds: Sequence[RetrievedMemory]
) -> tuple[tuple[RetrievedMemory, ...], ActivationLog]:
    """Walk out from ``seeds`` and return what lit up, plus the provenance."""
    if not seeds or self.max_hops == 0:
        return (), ActivationLog()

    state = _Frontier(seen={m.record.record_id for m in seeds})
    for memory in seeds:
        state.nodes.append(ActivatedNode.of(memory.record, score=_unit(memory.score)))
        state.trail[memory.record.record_id] = (memory.record.record_id,)

    frontier = {m.record.record_id: _unit(m.score) for m in seeds}
    for hop in range(1, self.max_hops + 1):
        try:
            edges = await self.source.neighbours(scope, tuple(frontier))
        except MemoryUnavailable:
            return tuple(state.found), _log(state, degraded=True)
        frontier = self._admit(state, edges, frontier, hop, scope)
        if not frontier:
            break

    return tuple(state.found), _log(state)

WorkingContext dataclass

WorkingContext(turns: tuple[ConversationTurn, ...] = (), dropped: tuple[tuple[str, str], ...] = (), degraded: bool = False)

The conversation window as it will render, and what it left out.

render

render() -> str

One turn per line, oldest first.

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

def render(self) -> str:
    """One turn per line, oldest first."""
    return "\n".join(turn.line() for turn in self.turns)

WorkingWindow dataclass

WorkingWindow(source: ConversationSource, recent_turns: int = 0, exclude_speakers: frozenset[str] = frozenset())

Reads the last few turns of a conversation, ungated.

read async

read(scope: MemoryScope) -> WorkingContext

Read the window at scope, dropping only what would corrupt it.

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

async def read(self, scope: MemoryScope) -> WorkingContext:
    """Read the window at ``scope``, dropping only what would corrupt it."""
    if self.recent_turns == 0:
        return WorkingContext()
    try:
        recent = await self.source.recent(scope, self.recent_turns)
    except MemoryUnavailable:
        return WorkingContext(degraded=True)

    state = _Window()
    for turn in recent[-self.recent_turns :]:
        if not turn.text.strip():
            # An empty row renders as a bare "[working] " bullet, which reads
            # as a turn in which nobody said anything.
            state.dropped.append((turn.turn_id, "the turn carries no text"))
            continue
        if turn.speaker and turn.speaker in self.exclude_speakers:
            state.dropped.append(
                (turn.turn_id, f"speaker {turn.speaker!r} is excluded")
            )
            continue
        state.kept.append(turn)
    return WorkingContext(turns=tuple(state.kept), dropped=tuple(state.dropped))

WriteOutcome dataclass

WriteOutcome(request: WriteRequest, receipt: WriteReceipt | None = None, error: str = '')

What one background write did, once it finished.

WriteReceipt dataclass

WriteReceipt(accepted: tuple[str, ...] = (), rejected: tuple[tuple[str, str], ...] = (), degraded: bool = False)

What the write port did, per memory.

ok property

ok: bool

Whether every memory in the request was stored.

WriteRequest dataclass

WriteRequest(scope: MemoryScope, records: tuple[MemoryRecord, ...] = (), turn: int = 0)

One post-response write: a scope, the memories it produced, the turn.

validate

validate() -> None

Refuse a request no adapter should have to interpret.

Both checks are about identity, which is why they raise rather than landing in the receipt's rejected list: a record filed under another scope and two records sharing an id are ambiguities, and an adapter that resolved either one silently would resolve it differently from the next adapter.

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

def validate(self) -> None:
    """Refuse a request no adapter should have to interpret.

    Both checks are about *identity*, which is why they raise rather than
    landing in the receipt's ``rejected`` list: a record filed under another
    scope and two records sharing an id are ambiguities, and an adapter that
    resolved either one silently would resolve it differently from the next
    adapter.
    """
    seen: set[str] = set()
    for record in self.records:
        record.validate()
        if record.scope_path != self.scope.path:
            raise MemoryContractError(
                f"memory {record.record_id!r} declares scope {record.scope_path!r} in a "
                f"write to {self.scope.path!r}. A write states one scope; a record filed "
                "under another is a cross-scope write wearing a single-scope call."
            )
        if record.record_id in seen:
            raise MemoryContractError(
                f"memory {record.record_id!r} appears twice in one write. Ids are the "
                "upsert key, so the request does not say which of the two survives."
            )
        seen.add(record.record_id)

apply_erasure

apply_erasure(provenance: dict[str, Any] | None, erased_conversation_id: str, *, pii_policy: str = PII_POLICY_DELETE) -> tuple[dict[str, Any] | None, bool]

Remove one conversation from a provenance; report whether to delete.

Reference-counted rather than cascading: a fact corroborated by two conversations survives the erasure of one. When the last source goes, the default policy marks the memory for deletion — no orphaned personal data survives as an anonymous "fact".

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

def apply_erasure(
    provenance: dict[str, Any] | None,
    erased_conversation_id: str,
    *,
    pii_policy: str = PII_POLICY_DELETE,
) -> tuple[dict[str, Any] | None, bool]:
    """Remove one conversation from a provenance; report whether to delete.

    Reference-counted rather than cascading: a fact corroborated by two
    conversations survives the erasure of one. When the last source goes, the
    default policy marks the memory for deletion — no orphaned personal data
    survives as an anonymous "fact".
    """
    if not provenance:
        return provenance, False
    conversations = [
        conversation
        for conversation in provenance.get("source_conversation_ids", [])
        if conversation != erased_conversation_id
    ]
    scope_paths = [
        path
        for path in provenance.get("source_scope_paths", [])
        if erased_conversation_id not in path
    ]
    remaining = {
        **provenance,
        "source_conversation_ids": conversations,
        "source_scope_paths": scope_paths,
    }
    if conversations:
        return remaining, False
    return remaining, pii_policy != PII_POLICY_RETAIN

as_memory_scope

as_memory_scope(scope: Any) -> MemoryScope

Accept either scope type, so a host is not forced to pick one.

A platform scope and a memory scope are two spellings of one identity, and making a caller convert would put the translation in every composition root instead of here.

to_memory_scope() is tried first and its result is checked, because on FrameworkTenantScope that method returns another framework scope rather than a memory scope. Trusting it returned an object with no segments, and the mismatch surfaced far away -- inside a retrieval, as a missing attribute on a type nobody in that traceback had named. So the fallback below reads the scope's own path, which is the one identity both spellings agree on.

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

def as_memory_scope(scope: Any) -> MemoryScope:
    """Accept either scope type, so a host is not forced to pick one.

    A platform scope and a memory scope are two spellings of one identity, and
    making a caller convert would put the translation in every composition root
    instead of here.

    ``to_memory_scope()`` is tried first and its **result is checked**, because
    on ``FrameworkTenantScope`` that method returns another framework scope
    rather than a memory scope. Trusting it returned an object with no
    ``segments``, and the mismatch surfaced far away -- inside a retrieval,
    as a missing attribute on a type nobody in that traceback had named. So
    the fallback below reads the scope's own path, which is the one identity
    both spellings agree on.
    """
    if isinstance(scope, MemoryScope):
        return scope
    converted = getattr(scope, "to_memory_scope", None)
    if callable(converted):
        result = converted()
        if isinstance(result, MemoryScope):
            return result
    # A platform ``SubjectScope`` spells the same identity as an ordered tuple
    # of segments. Read before ``path`` because it is the narrower signal: a
    # scope carrying both means the same thing either way, and a scope carrying
    # only this one is what an authenticated request produces.
    segments = getattr(scope, "segments", None)
    if segments:
        ids = [str(segment) for segment in segments if segment]
        if ids:
            return MemoryScope(*ids[:3])

    levels = getattr(scope, "path", None)
    if levels:
        ids = [str(level.id) for level in levels if getattr(level, "id", None)]
        if ids:
            return MemoryScope(*ids[:3])
    raise TypeError(
        f"{type(scope).__name__} is neither a MemoryScope nor convertible to "
        "one, so the capability cannot be bound to a scope"
    )

as_tenant_scope

as_tenant_scope(scope: MemoryScope) -> TenantScope

The legacy scope value for a capability scope. Kinds from compat.

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

def tenant_scope(scope: MemoryScope) -> TenantScope:
    """The legacy scope value for a capability scope. Kinds from ``compat``."""
    levels = [
        ScopeLevel(kind=kind, id=segment)
        for kind, segment in zip(LEGACY_KINDS, scope.segments, strict=False)
    ]
    return TenantScope.from_path(levels)

build_provenance

build_provenance(*, source_conversation_id: str, source_scope_path: str, promoted_by: str, extraction_confidence: float, promoted_at: str | None = None) -> dict[str, Any]

The provenance a newly promoted memory carries.

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

def build_provenance(
    *,
    source_conversation_id: str,
    source_scope_path: str,
    promoted_by: str,
    extraction_confidence: float,
    promoted_at: str | None = None,
) -> dict[str, Any]:
    """The provenance a newly promoted memory carries."""
    return {
        "source_conversation_ids": [source_conversation_id],
        "source_scope_paths": [source_scope_path],
        "promoted_at": promoted_at or datetime.now(UTC).isoformat(),
        "promoted_by": promoted_by,
        "extraction_confidence": extraction_confidence,
    }

classification_identity

classification_identity(properties: Mapping[str, Any], *, layer: Any = None) -> tuple[tuple[str, str] | None, str]

Read the storage-owned atomic classification contract, lazily.

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

def classification_identity(
    properties: Mapping[str, Any], *, layer: Any = None,
) -> tuple[tuple[str, str] | None, str]:
    """Read the storage-owned atomic classification contract, lazily."""
    from symfonic.memory.classification import classification_identity as read
    return read(properties, layer=layer)

contribution_spec

contribution_spec(contribution: MemoryContribution) -> Mapping[str, object]

Project a validated declaration into compiler-ready keyword values.

A mapping rather than a compiler object: the composition root — which is allowed to see both capabilities — turns this into a PromptContribution, so neither capability can quietly start depending on the other's internals.

Enum members are emitted as their string values; the prompt contract's layers, tiers, and scopes are StrEnums over the same strings, so the root's conversion is total by construction.

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

def contribution_spec(contribution: MemoryContribution) -> Mapping[str, object]:
    """Project a validated declaration into compiler-ready keyword values.

    A mapping rather than a compiler object: the composition root — which is
    allowed to see both capabilities — turns this into a ``PromptContribution``,
    so neither capability can quietly start depending on the other's internals.

    Enum members are emitted as their string values; the prompt contract's
    layers, tiers, and scopes are ``StrEnum``s over the same strings, so the
    root's conversion is total by construction.
    """
    contribution.validate()
    return MappingProxyType(
        {
            "contribution_id": contribution.contribution_id,
            "source": contribution.source,
            "capability": contribution.capability,
            "layer": contribution.layer.value,
            "tier": contribution.tier.value,
            "scope": contribution.scope.value,
            "order": contribution.order,
            "inherit": contribution.inherit,
            "pinned": contribution.pinned,
            "requires_hydration": contribution.requires_hydration,
        }
    )

deep_phases

deep_phases(*, graph: Any, store: Any = None, policy: Any = None, entity_extractor: Any = None, scope_promoter: Any = None, entity_min_mention_count: int | None = None, entity_max_episodics_per_run: int | None = None, entity_confidence_threshold: float | None = None, **nightly: Any) -> tuple[ConsolidationPhase, ...]

Build the complete DEEP roster, in roster order.

Parameters:

Name	Type	Description	Default
graph	Any	the GraphBackend the store reads through, or a GraphMemoryStore.	required
store	Any	the memory store, for the phases that learn from a scope's turns. Without it, three of the seventeen decline.	None
entity_extractor	Any	what turns an episode into candidate entities. Absent, phase 12.5 declines -- which is the shipped default, not a misconfiguration.	None
scope_promoter	Any	async (context) -> int, promoting corroborated memories toward the root. Absent, the phase declines.	None
**nightly	Any	forwarded to :func:~.nightly.nightly_phases.	{}

Source code in src/symfonic/capabilities/memory/phases/deep.py

def deep_phases(
    *,
    graph: Any,
    store: Any = None,
    policy: Any = None,
    entity_extractor: Any = None,
    scope_promoter: Any = None,
    entity_min_mention_count: int | None = None,
    entity_max_episodics_per_run: int | None = None,
    entity_confidence_threshold: float | None = None,
    **nightly: Any,
) -> tuple[ConsolidationPhase, ...]:
    """Build the complete DEEP roster, in roster order.

    Args:
        graph: the ``GraphBackend`` the store reads through, or a
            ``GraphMemoryStore``.
        store: the memory store, for the phases that learn from a scope's
            turns. Without it, three of the seventeen decline.
        entity_extractor: what turns an episode into candidate entities.
            Absent, phase 12.5 declines -- which is the shipped default, not a
            misconfiguration.
        scope_promoter: ``async (context) -> int``, promoting corroborated
            memories toward the root. Absent, the phase declines.
        **nightly: forwarded to :func:`~.nightly.nightly_phases`.
    """
    from symfonic.capabilities.memory.rosters import PHASE_ROSTER
    from symfonic.capabilities.memory.schedule import ConsolidationCycle

    if policy is not None:
        # The deployment's own numbers, under the names the factories use.
        # Explicit arguments still win: a caller that passed both meant the one
        # it wrote at the call site, not the one its settings file carries.
        tuned = policy.as_phase_kwargs()
        if entity_min_mention_count is None:
            entity_min_mention_count = tuned.pop("entity_min_mention_count", None)
        if entity_max_episodics_per_run is None:
            entity_max_episodics_per_run = tuned.pop(
                "entity_max_episodics_per_run", None
            )
        if entity_confidence_threshold is None:
            entity_confidence_threshold = tuned.pop(
                "entity_confidence_threshold", None
            )
        for name in (
            "entity_min_mention_count",
            "entity_max_episodics_per_run",
            "entity_confidence_threshold",
        ):
            tuned.pop(name, None)
        nightly = {**tuned, **nightly}

    resolved = phase_graph(graph)
    settings = {
        key: value
        for key, value in (
            ("min_mention_count", entity_min_mention_count),
            ("max_episodics_per_run", entity_max_episodics_per_run),
            ("confidence_threshold", entity_confidence_threshold),
        )
        if value is not None
    }
    evidence = EpisodicEvidence(store) if store is not None else None
    if entity_extractor is not None and store is None:
        raise MemoryContractError(
            "deep_phases was given an entity extractor and no store, so the "
            "phase that mints entities has no turns to read them from. Pass "
            "the memory store, or neither."
        )

    by_name: dict[str, Any] = {
        "entity_links": EntityLinksPhase(
            resolved, evidence, extractor=entity_extractor, **settings
        ),
        "scope_promotion": ScopePromotionPhase(scope_promoter),
    }
    # Nightly's fifteen come from the factory that owns them -- one definition
    # of "what is a strengthen phase", not two.
    for phase in nightly_phases(graph=resolved, store=store, **nightly):
        by_name[phase.name] = phase

    order = PHASE_ROSTER[ConsolidationCycle.DEEP]
    return PhaseRoster((by_name[name] for name in order), resolved)

flatten

flatten(text: str) -> str

Collapse every line-break form in text to a single space.

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

def flatten(text: str) -> str:
    """Collapse every line-break form in ``text`` to a single space."""
    return _LINE_BREAK.sub(" ", text)

importance_to_salience

importance_to_salience(importance: float) -> float

Map the legacy 1–10 importance grid onto [0, 1] salience.

Clamped rather than refused: an out-of-range importance is a row legacy already stored (its own validator only fires on construction, not on read), and refusing to read it would make one bad row poison a whole retrieval.

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

def importance_to_salience(importance: float) -> float:
    """Map the legacy 1–10 importance grid onto ``[0, 1]`` salience.

    Clamped rather than refused: an out-of-range importance is a row legacy
    already stored (its own validator only fires on construction, not on read),
    and refusing to read it would make one bad row poison a whole retrieval.
    """
    clamped = min(max(float(importance), IMPORTANCE_FLOOR), IMPORTANCE_CEILING)
    # Twelve places, not six: the grid step is 1/9, and rounding a repeating
    # fraction at six places loses the round trip (``2 -> 0.111111 -> 1.999999``),
    # which would drift a memory's importance a little on every migration hop.
    return round((clamped - IMPORTANCE_FLOOR) / (IMPORTANCE_CEILING - IMPORTANCE_FLOOR), 12)

is_promotable

is_promotable(candidate: PromotionCandidate, *, confidence_floor: float = DEFAULT_PROMOTION_CONFIDENCE_FLOOR) -> bool

Whether candidate may be published to a broader scope.

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

def is_promotable(
    candidate: PromotionCandidate,
    *,
    confidence_floor: float = DEFAULT_PROMOTION_CONFIDENCE_FLOOR,
) -> bool:
    """Whether ``candidate`` may be published to a broader scope."""
    if candidate.durability != _PROMOTABLE_DURABILITY:
        return False
    return candidate.confidence >= confidence_floor

json_payload

json_payload(text: str) -> dict[str, Any] | None

The first JSON object in text, or None.

Same brace-matching heuristic the legacy parser uses, with one addition: a payload that parses to something other than an object is refused rather than returned. json.loads on [1, 2] succeeds, and a caller that then asks it for ops gets an AttributeError from inside a post-response stage instead of "the model did not answer in our format".

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

def json_payload(text: str) -> dict[str, Any] | None:
    """The first JSON *object* in ``text``, or ``None``.

    Same brace-matching heuristic the legacy parser uses, with one addition:
    a payload that parses to something other than an object is refused rather
    than returned. ``json.loads`` on ``[1, 2]`` succeeds, and a caller that
    then asks it for ``ops`` gets an ``AttributeError`` from inside a
    post-response stage instead of "the model did not answer in our format".
    """
    start = text.find("{")
    end = text.rfind("}") + 1
    if start < 0 or end <= start:
        return None
    try:
        payload = json.loads(text[start:end])
    except (json.JSONDecodeError, ValueError):
        return None
    return payload if isinstance(payload, dict) else None

layer_index

layer_index(layer: MemoryLayer) -> int

Position of layer on the ladder; lower renders earlier.

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

def layer_index(layer: MemoryLayer) -> int:
    """Position of ``layer`` on the ladder; lower renders earlier."""
    return LAYER_LADDER.index(layer)

legacy_entry_payload

legacy_entry_payload(record: MemoryRecord, *, metadata: Mapping[str, Any] | None = None) -> dict[str, Any]

Build the kwargs a legacy MemoryEntry is constructed from.

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

def legacy_entry_payload(
    record: MemoryRecord,
    *,
    metadata: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """Build the kwargs a legacy ``MemoryEntry`` is constructed from."""
    scope = record.scope
    # The record's own metadata first, so a caller's bag can still override it
    # -- the pending markers are written that way and must win.
    # The producer's own keys first and stripped of the capability's, so a
    # record cannot name its own scope, identity or durability; the caller's
    # bag then wins, because the pending markers are written that way.
    bag: dict[str, Any] = {
        **producer_metadata(record.metadata),
        **dict(metadata or {}),
    }
    bag[SCOPE_PATH_KEY] = legacy_scope_path(scope)
    bag[ORIGIN_KEY] = record.origin
    # TA8.74. Stamped in the properties as well as in ``id``, which is what
    # ``legacy_node_payload`` already did and this builder did not.
    # ``record_from_legacy_node`` reads the bag first and falls back to the
    # node's own id, so a store that mints its own identity used to hand back a
    # record nobody could match to what was written -- the write said ``m1``
    # and the read said a fresh uuid. The layers honour the id now; this keeps
    # the round trip true even where one does not.
    bag[RECORD_ID_KEY] = record.record_id
    return {
        "id": record.record_id,
        "layer": record.layer.value,
        "tenant_id": scope.tenant,
        "content": record.text,
        "metadata": bag,
        "importance": salience_to_importance(record.salience),
    }

legacy_node_payload

legacy_node_payload(record: MemoryRecord, *, durability: str = 'durable', provenance: Mapping[str, Any] | None = None, properties: Mapping[str, Any] | None = None) -> dict[str, Any]

Build the kwargs a legacy MemoryNode is constructed from.

label carries the text because that is where legacy keeps it — its commit_pending writes content=op.node.label. Everything this capability knows and legacy does not rides in properties, which every graph backend persists as an opaque bag.

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

def legacy_node_payload(
    record: MemoryRecord,
    *,
    durability: str = "durable",
    provenance: Mapping[str, Any] | None = None,
    properties: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """Build the kwargs a legacy ``MemoryNode`` is constructed from.

    ``label`` carries the text because that is where legacy keeps it — its
    ``commit_pending`` writes ``content=op.node.label``. Everything this
    capability knows and legacy does not rides in ``properties``, which every
    graph backend persists as an opaque bag.
    """
    scope = record.scope
    bag: dict[str, Any] = dict(properties or {})
    bag[SCOPE_PATH_KEY] = legacy_scope_path(scope)
    bag[DURABILITY_KEY] = durability
    bag[RECORD_ID_KEY] = record.record_id
    bag[ORIGIN_KEY] = record.origin
    # Written only when the record claims one, so a row that makes no claim
    # carries no key rather than an empty string a reader has to interpret.
    # Top level and nowhere else: ``producer_metadata`` strips the same name
    # from the nested bag, so there is one home and never two copies to
    # disagree.
    if record.edited_by:
        bag[EDITED_BY_KEY] = record.edited_by
    # TA-3-1-9. The field was public, accepted and discarded: this builder read
    # its ``properties`` keyword and never ``record.metadata``, while
    # ``legacy_entry_payload`` next door merged it -- so one direction honoured
    # the field and the other dropped it, and the store used the one that
    # dropped it. Written only when there is something to write, so a record
    # with no metadata does not grow an empty key in every stored row.
    if (own := producer_metadata(record.metadata)):
        bag[METADATA_KEY] = own
    if provenance is not None:
        bag[PROVENANCE_KEY] = dict(provenance)
    return {
        "id": record.record_id,
        "layer": record.layer.value,
        "tenant_id": scope.tenant,
        "label": record.text,
        "properties": bag,
        "importance": salience_to_importance(record.salience),
    }

legacy_scope_path

legacy_scope_path(scope: MemoryScope) -> str

Materialise scope into the string legacy stores and filters on.

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

def legacy_scope_path(scope: MemoryScope) -> str:
    """Materialise ``scope`` into the string legacy stores and filters on."""
    parts: list[str] = []
    for kind, segment in zip(LEGACY_SCOPE_KINDS, scope.segments, strict=False):
        parts.append(kind)
        parts.append(segment)
    return LEGACY_SCOPE_DELIMITER.join(parts)

link_entities

link_entities(records: Sequence[MemoryRecord], *, scope: MemoryScope, extractor: HeuristicEntityExtractor | None = None, min_mention_count: int = 2, confidence_threshold: float = 0.5, max_records: int = 200) -> LinkingResult

Mint entities for surfaces mentioned often enough, and link co-mentions.

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

def link_entities(
    records: Sequence[MemoryRecord],
    *,
    scope: MemoryScope,
    extractor: HeuristicEntityExtractor | None = None,
    min_mention_count: int = 2,
    confidence_threshold: float = 0.5,
    max_records: int = 200,
) -> LinkingResult:
    """Mint entities for surfaces mentioned often enough, and link co-mentions."""
    if len(records) > max_records:
        return LinkingResult(
            dropped=(
                (
                    "",
                    f"{len(records)} memories exceed the {max_records}-memory read "
                    "budget for one linking pass",
                ),
            )
        )
    reader = extractor if extractor is not None else HeuristicEntityExtractor()
    counts: dict[str, int] = {}
    per_record: list[tuple[str, ...]] = []
    dropped: list[tuple[str, str]] = []

    for record in records:
        kept: list[str] = []
        for mention in reader.extract(record.text):
            if mention.confidence < confidence_threshold:
                dropped.append(
                    (
                        mention.surface,
                        f"confidence {mention.confidence} is below the "
                        f"{confidence_threshold} threshold",
                    )
                )
                continue
            counts[mention.surface] = counts.get(mention.surface, 0) + 1
            kept.append(mention.surface)
        per_record.append(tuple(kept))

    minted = {
        surface for surface, count in counts.items() if count >= min_mention_count
    }
    for surface, count in sorted(counts.items()):
        if surface not in minted:
            dropped.append(
                (surface, f"mentioned {count} time(s), below the {min_mention_count} floor")
            )

    entities = tuple(
        _entity_record(surface, scope, counts[surface]) for surface in sorted(minted)
    )
    return LinkingResult(
        entities=entities,
        links=_links(per_record, minted),
        dropped=tuple(dropped),
    )

memory_capabilities

memory_capabilities(store: Any, scope: MemoryScope | Any, *, limit: int = 5, recall_budget: RecallBudget | None = None, extractor: MemoryExtractorPort | None = None, consolidation: Any | None = None, conversation: Any | None = None, recent_turns: int = 0, activation: Any | None = None) -> list[Any]

The capability and the grants it needs, as one list to compose.

Parameters:

Name	Type	Description	Default
store	Any	an HMS satisfying the retrieval, write and lifecycle ports.	required
scope	MemoryScope | Any	the scope this agent serves. Closed over by the capability, so one agent is one tenant.	required
limit	int	how many memories a turn recalls.	5
recall_budget	RecallBudget | None	explicit UTF-8 rendered recall ceilings. None preserves legacy character caps; unrelated to the working conversation window.	None
extractor	MemoryExtractorPort | None	a :class:~.ports.MemoryExtractorPort -- normally :class:~.extraction.MemoryExtractionService -- run after the final model round so a turn's durable facts are written without the deployment invoking anything. None composes memory that recalls and records the exchange and extracts nothing.	None
consolidation	Any | None	a :class:~.napping.ConsolidationCoordinator -- built over a :class:~.consolidation.ConsolidationRuntime whose roster came from :func:~.phases.quick.quick_phases. Given one, the turn's last act is to advance this scope's cadence and run the cycle it makes due, in the background. None composes memory that never consolidates on its own.	None
activation	Any | None	a :class:~.recall.SpreadingActivation over an :class:~.recall.AssociationSource. It adds bounded graph neighbours to direct recall. None preserves direct recall.	None

Raises:

Type	Description
ConfigurationError	if extractor cannot serve the port.

Returned together because a capability may not grant itself an effect and a caller that forgot one would get a fold refusal naming a grant rather than a missing feature. The three are what memory is.

Prompting is NOT included, and composing memory alone does not put recall in front of the model. Memory is a resolution stage: it reaches the store and leaves an entry in the turn's snapshot. Prompting is the compilation stage that reads that snapshot and renders it. Fold memory by itself and both ports are called, the block is composed, the snapshot is populated -- and the model receives the bare instructions, because nothing consumed the entry. That exact defect has been found in this codebase twice.

It is not added here because a factory named for memory that quietly composed prompting would decide a deployment's prompt on its behalf. A composition root wanting recall in the prompt adds a PromptingCapability beside this, and :mod:tests.platform.test_vertical_slice shows the pair.

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

def memory_capabilities(
    store: Any,
    scope: MemoryScope | Any,
    *,
    limit: int = 5,
    recall_budget: RecallBudget | None = None,
    extractor: MemoryExtractorPort | None = None,
    consolidation: Any | None = None,
    conversation: Any | None = None,
    recent_turns: int = 0,
    activation: Any | None = None,
) -> list[Any]:
    """The capability *and* the grants it needs, as one list to compose.

    Args:
        store: an HMS satisfying the retrieval, write and lifecycle ports.
        scope: the scope this agent serves. Closed over by the capability, so
            one agent is one tenant.
        limit: how many memories a turn recalls.
        recall_budget: explicit UTF-8 rendered recall ceilings. None preserves
            legacy character caps; unrelated to the working conversation window.
        extractor: a :class:`~.ports.MemoryExtractorPort` -- normally
            :class:`~.extraction.MemoryExtractionService` -- run after the
            final model round so a turn's durable facts are written without
            the deployment invoking anything. ``None`` composes memory that
            recalls and records the exchange and extracts nothing.
        consolidation: a :class:`~.napping.ConsolidationCoordinator` -- built
            over a :class:`~.consolidation.ConsolidationRuntime` whose roster
            came from :func:`~.phases.quick.quick_phases`. Given one, the
            turn's last act is to advance this scope's cadence and run the
            cycle it makes due, in the background. ``None`` composes memory
            that never consolidates on its own.
        activation: a :class:`~.recall.SpreadingActivation` over an
            :class:`~.recall.AssociationSource`. It adds bounded graph
            neighbours to direct recall. ``None`` preserves direct recall.

    Raises:
        ConfigurationError: if ``extractor`` cannot serve the port.

    Returned together because a capability may not grant itself an effect and
    a caller that forgot one would get a fold refusal naming a grant rather
    than a missing feature. The three are what memory is.

    **Prompting is NOT included, and composing memory alone does not put recall
    in front of the model.** Memory is a *resolution* stage: it reaches the
    store and leaves an entry in the turn's snapshot. Prompting is the
    *compilation* stage that reads that snapshot and renders it. Fold memory by
    itself and both ports are called, the block is composed, the snapshot is
    populated -- and the model receives the bare instructions, because nothing
    consumed the entry. That exact defect has been found in this codebase twice.

    It is not added here because a factory named for memory that quietly
    composed prompting would decide a deployment's prompt on its behalf. A
    composition root wanting recall in the prompt adds a ``PromptingCapability``
    beside this, and :mod:`tests.platform.test_vertical_slice` shows the pair.
    """
    from symfonic.kernel.contracts.effects import GrantEffects

    # ``memory-consolidate`` travels with the rest for the same reason the
    # other three do: it is granted only when a nap is composed, and a caller
    # who passed a coordinator and no grant would read a fold refusal naming an
    # effect rather than the feature they asked for.
    effects = ["memory-read", "memory-write", "memory-flush"]
    if consolidation is not None:
        effects.append("memory-consolidate")
    return [
        GrantEffects(*effects),
        MemoryBundleFactory(
            store, limit=limit, recall_budget=recall_budget,
            extractor=extractor, consolidation=consolidation,
            conversation=conversation, recent_turns=recent_turns,
            activation=activation,
        ).for_scope(scope),
    ]

merge_provenance

merge_provenance(existing: dict[str, Any] | None, *, source_conversation_id: str, source_scope_path: str, promoted_by: str, extraction_confidence: float) -> dict[str, Any]

Append a corroborating conversation to an existing provenance.

The highest confidence observed wins and the latest promotion time stands: a second conversation confirming a fact makes it more trustworthy, and taking the newer (possibly lower) confidence would let one weak restatement demote a well-established memory.

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

def merge_provenance(
    existing: dict[str, Any] | None,
    *,
    source_conversation_id: str,
    source_scope_path: str,
    promoted_by: str,
    extraction_confidence: float,
) -> dict[str, Any]:
    """Append a corroborating conversation to an existing provenance.

    The highest confidence observed wins and the latest promotion time stands:
    a second conversation confirming a fact makes it *more* trustworthy, and
    taking the newer (possibly lower) confidence would let one weak restatement
    demote a well-established memory.
    """
    if not existing:
        return build_provenance(
            source_conversation_id=source_conversation_id,
            source_scope_path=source_scope_path,
            promoted_by=promoted_by,
            extraction_confidence=extraction_confidence,
        )
    conversations = list(existing.get("source_conversation_ids", []))
    scope_paths = list(existing.get("source_scope_paths", []))
    if source_conversation_id not in conversations:
        conversations.append(source_conversation_id)
    if source_scope_path not in scope_paths:
        scope_paths.append(source_scope_path)
    return {
        "source_conversation_ids": conversations,
        "source_scope_paths": scope_paths,
        "promoted_at": datetime.now(UTC).isoformat(),
        "promoted_by": promoted_by,
        "extraction_confidence": max(
            _as_float(existing.get("extraction_confidence")), extraction_confidence
        ),
    }

mint_operation

mint_operation(op: Mapping[str, Any], index: int, request: ExtractionRequest, family: ProviderFamily, scrubber: CredentialScrubber) -> ExtractedMemory | tuple[str, str]

Mint one memory from one operation, or report why it produced none.

Returns an :class:ExtractedMemory, or (identifier, reason) where an empty reason means "this operation was a legitimate no-op" — a noop action is the model saying there is nothing to remember, which is an answer rather than a refusal.

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

def mint_operation(
    op: Mapping[str, Any],
    index: int,
    request: ExtractionRequest,
    family: ProviderFamily,
    scrubber: CredentialScrubber,
) -> ExtractedMemory | tuple[str, str]:
    """Mint one memory from one operation, or report why it produced none.

    Returns an :class:`ExtractedMemory`, or ``(identifier, reason)`` where an
    empty reason means "this operation was a legitimate no-op" — a ``noop``
    action is the model saying there is nothing to remember, which is an answer
    rather than a refusal.
    """
    identifier = f"op[{index}]"
    action = str(op.get("action", "noop"))
    if action == "noop":
        return identifier, ""
    if action not in NODE_ACTIONS:
        return identifier, (
            f"action {action!r} carries no memory text; only "
            f"{sorted(NODE_ACTIONS)} produce a record"
        )
    try:
        layer = resolve_layer(str(op.get("layer", MemoryLayer.SEMANTIC.value)))
    except MemoryCapabilityError as exc:
        return identifier, str(exc)

    raw_text = str(op.get("label", op.get("id", ""))).strip()
    if not raw_text:
        return identifier, "the operation carries no memory text"
    importance = _as_float(op.get("importance"), default=5.0)
    if importance < request.min_importance:
        return identifier, (
            f"importance {importance} is below the {request.min_importance} "
            "write threshold"
        )

    raw_properties = op.get("properties")
    properties, dropped_keys = scrubber.scrub_properties(
        raw_properties if isinstance(raw_properties, Mapping) else {}
    )
    identity, classification_error = classification_identity(properties, layer=layer)
    if classification_error:
        return identifier, classification_error
    if identity is None:
        return identifier, "classification_missing"
    properties["memory_category"], properties["subject"] = identity
    # A label such as ``preferred_name`` identifies a field but does not carry
    # the fact. Persist the scrubbed values in the retrievable text as well as
    # metadata, otherwise recall tells the model which fact existed and drops
    # its value (the live scaffold stored ``deployment_codename`` but not
    # ``ATLAS-8642``).
    rendered = raw_text
    # Include every property value the label does not already carry.  Testing
    # ``any`` value made ``person:Amiel`` suppress *all* properties because the
    # name was present, silently dropping ``occupation: software engineer``
    # from the only text retrieval renders.  Shared with the compatibility
    # reader so already-persisted rows gain the same complete representation.
    rendered = retrievable_text(
        raw_text,
        {key: value for key, value in properties.items() if key not in _CLASSIFICATION_KEYS},
    )
    scrubbed = scrubber.scrub_text(rendered)
    return ExtractedMemory(
        record=MemoryRecord(
            record_id=record_id(request, index, scrubbed.text),
            layer=layer,
            text=scrubbed.text,
            scope_path=request.scope.path,
            salience=importance_to_salience(importance),
            origin=f"extraction:{family.value}",
        ),
        importance=importance,
        action=action,
        properties=properties,
        redactions=scrubbed.redactions + tuple(f"KEY:{key}" for key in dropped_keys),
    )

new_owner_token

new_owner_token() -> str

A token this holder can prove and another cannot guess.

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

def new_owner_token() -> str:
    """A token this holder can prove and another cannot guess."""
    return uuid.uuid4().hex

nightly_phases

nightly_phases(*, graph: Any, store: Any = None, procedural: Any = None, pending_connections: Any = (), llm_summarise: Any = None, procedural_extractor: Any = None, procedural_model_name: str = '', stale_days: int | None = None, **quick: Any) -> tuple[ConsolidationPhase, ...]

Build the complete NIGHTLY roster, in roster order.

Parameters:

Name	Type	Description	Default
graph	Any	the GraphBackend the store reads through, or a GraphMemoryStore if the deployment has one.	required
store	Any	the memory store, for the two phases that learn from a scope's turns. Without it they decline: reading episodic evidence through anything but the retrieval port would mean a second answer to "what has this scope done", and the wrong one.	None
procedural	Any	where a draft skill is written. Without it phase 12 declines rather than extracting patterns it cannot store.	None
pending_connections	Any	inferred edges to materialise this cycle.	()
llm_summarise	Any	what names a cluster, for phase 4.	None
procedural_extractor	Any	a model-backed :class:~.drafts.ProceduralExtractor. When given it runs instead of the regex extractor -- the eval that added it rejected merged streams.	None
stale_days	int | None	the decay horizon, when a deployment tunes it.	None
**quick	Any	forwarded to :func:~.quick.quick_phases.	{}

Source code in src/symfonic/capabilities/memory/phases/nightly.py

def nightly_phases(
    *,
    graph: Any,
    store: Any = None,
    procedural: Any = None,
    pending_connections: Any = (),
    llm_summarise: Any = None,
    procedural_extractor: Any = None,
    procedural_model_name: str = "",
    stale_days: int | None = None,
    **quick: Any,
) -> tuple[ConsolidationPhase, ...]:
    """Build the complete NIGHTLY roster, in roster order.

    Args:
        graph: the ``GraphBackend`` the store reads through, or a
            ``GraphMemoryStore`` if the deployment has one.
        store: the memory store, for the two phases that learn from a scope's
            turns. Without it they decline: reading episodic evidence through
            anything but the retrieval port would mean a second answer to
            "what has this scope done", and the wrong one.
        procedural: where a draft skill is written. Without it phase 12
            declines rather than extracting patterns it cannot store.
        pending_connections: inferred edges to materialise this cycle.
        llm_summarise: what names a cluster, for phase 4.
        procedural_extractor: a model-backed
            :class:`~.drafts.ProceduralExtractor`. When given it runs
            *instead of* the regex extractor -- the eval that added it
            rejected merged streams.
        stale_days: the decay horizon, when a deployment tunes it.
        **quick: forwarded to :func:`~.quick.quick_phases`.
    """
    resolved = phase_graph(graph)
    everything = AllNodes(resolved)
    evidence = EpisodicEvidence(store) if store is not None else None
    if procedural is not None and store is None:
        raise MemoryContractError(
            "nightly_phases was given a procedural layer and no store, so the "
            "phase that writes drafts has nowhere to read the turns it learns "
            "from. Pass the memory store, or neither."
        )

    decay = {"stale_days": stale_days} if stale_days is not None else {}
    by_name: dict[str, Any] = {
        "cooccurrence": _OverAllNodes(
            "cooccurrence", create_cooccurrence_edges, resolved, everything
        ),
        "tag_risk": _OverAllNodes("tag_risk", tag_risk_nodes, resolved, everything),
        "prune_orphans": _OverAllNodes(
            "prune_orphans", prune_orphans, resolved, everything
        ),
        "meta_nodes": MetaNodesPhase(resolved, everything, llm_summarise=llm_summarise),
        "working_ttl": _OverAllNodes(
            "working_ttl", cleanup_working_ttl, resolved, everything
        ),
        "decay_importance": _OverAllNodes(
            "decay_importance", decay_importance, resolved, everything, **decay
        ),
        "expire_semantic_durability": _OverAllNodes(
            "expire_semantic_durability",
            expire_semantic_durability,
            resolved,
            everything,
        ),
        "prune_retracted": _OverScope("prune_retracted", prune_retracted, resolved),
        "pending_edges": PendingEdgesPhase(
            resolved, everything, connections=pending_connections
        ),
        "synthetic_links": SyntheticLinksPhase(resolved, evidence),
        "procedural_promotion": ProceduralPromotionPhase(
            procedural,
            evidence,
            llm_extractor=procedural_extractor,
            llm_model_name=procedural_model_name,
        ),
    }
    # Quick's four are nightly's too, built by the factory that owns them --
    # one implementation of "what is a strengthen phase", not two.
    for phase in quick_phases(graph=resolved, **quick):
        by_name[phase.name] = phase

    from symfonic.capabilities.memory.rosters import PHASE_ROSTER
    from symfonic.capabilities.memory.schedule import ConsolidationCycle

    order = PHASE_ROSTER[ConsolidationCycle.NIGHTLY]
    return PhaseRoster((by_name[name] for name in order), resolved)

phase_graph

phase_graph(graph: Any) -> Any

The store the phases write through, from a store or a bare backend.

isinstance rather than duck-typing: both objects answer to query_nodes and their signatures differ (the store takes layer=, the backend takes a filter mapping), so a check that guessed from the shape would guess wrong exactly where it mattered.

The backend is wrapped in :class:~symfonic.capabilities.memory.journal. JournalledGraph and the store in :class:~symfonic.capabilities.memory. fencing.FencedGraph, which is why every factory routes through this one function. Ten of the roster's phase modules write to the graph directly rather than through the write coordinator, so anything applied phase by phase would be a rule each of them -- and each one written later -- has to remember. Applied here it is structural: a phase gets both by being handed its graph.

Both are inert outside a cycle. The journal defers mutations only while one is running on this task and the fence checks only while a lease is held, so single-process use, an ordinary turn and every unit test behave exactly as before.

A deployment that shares one backend between the phases and its memory layers should wrap it once at the composition root instead -- see :class:~symfonic.capabilities.memory.journal.JournalledGraph. Wrapping here covers what the phases reach; it cannot cover what ProceduralLayer writes through a store this function never sees.

Source code in src/symfonic/capabilities/memory/phases/phase_graph.py

def phase_graph(graph: Any) -> Any:
    """The store the phases write through, from a store or a bare backend.

    ``isinstance`` rather than duck-typing: both objects answer to
    ``query_nodes`` and their signatures differ (the store takes ``layer=``,
    the backend takes a filter mapping), so a check that guessed from the shape
    would guess wrong exactly where it mattered.

    The backend is wrapped in :class:`~symfonic.capabilities.memory.journal.\
    JournalledGraph` and the store in :class:`~symfonic.capabilities.memory.\
    fencing.FencedGraph`, which is why every factory routes through this one
    function. Ten of the roster's phase modules write to the graph directly
    rather than through the write coordinator, so anything applied phase by
    phase would be a rule each of them -- and each one written later -- has to
    remember. Applied here it is structural: a phase gets both by being handed
    its graph.

    Both are inert outside a cycle. The journal defers mutations only while one
    is running on this task and the fence checks only while a lease is held, so
    single-process use, an ordinary turn and every unit test behave exactly as
    before.

    A deployment that shares one backend between the phases and its memory
    layers should wrap it once at the composition root instead -- see
    :class:`~symfonic.capabilities.memory.journal.JournalledGraph`. Wrapping
    here covers what the phases reach; it cannot cover what
    ``ProceduralLayer`` writes through a store this function never sees.
    """
    if isinstance(graph, FencedGraph):
        # Idempotent, because the deep and nightly factories resolve the graph
        # once and hand the result to the factories they compose.
        return graph
    if isinstance(graph, GraphMemoryStore):
        return FencedGraph(graph)
    if isinstance(graph, JournalledGraph):
        # Already journalled at the composition root, where it also covers
        # the memory layers. Wrapping again would nest one journal inside
        # another.
        return FencedGraph(GraphMemoryStore(graph))
    return FencedGraph(GraphMemoryStore(JournalledGraph(graph)))

procedural_layer

procedural_layer(graph: Any, **options: Any) -> Any

The procedural store over graph.

graph is the GraphBackend a composition root already has, or the GraphMemoryStore over it -- the same pair :func:~.phases.quick. phase_graph accepts, and for the same reason.

Imported inside the call: the layer pulls in the router, the predicates and the skill renderer, and a deployment that never learns a procedure should not pay for them in every import symfonic.agent.

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

def procedural_layer(graph: Any, **options: Any) -> Any:
    """The procedural store over ``graph``.

    ``graph`` is the ``GraphBackend`` a composition root already has, or the
    ``GraphMemoryStore`` over it -- the same pair :func:`~.phases.quick.\
    phase_graph` accepts, and for the same reason.

    Imported inside the call: the layer pulls in the router, the predicates and
    the skill renderer, and a deployment that never learns a procedure should
    not pay for them in every ``import symfonic.agent``.
    """
    from symfonic.capabilities.memory.phases.quick import phase_graph
    from symfonic.memory.layers.procedural import ProceduralLayer

    return ProceduralLayer(phase_graph(graph), **options)

promote

promote(candidate: PromotionCandidate, target: MemoryScope, *, promoted_by: str, promoted_at: str | None = None) -> Promotion

Restate candidate's memory at target, with its provenance.

The record id is kept. Ids are the upsert key within a scope, so re-promoting the same fact overwrites its own earlier promotion instead of accumulating near-duplicates of it, which is what makes running consolidation twice harmless.

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

def promote(
    candidate: PromotionCandidate,
    target: MemoryScope,
    *,
    promoted_by: str,
    promoted_at: str | None = None,
) -> Promotion:
    """Restate ``candidate``'s memory at ``target``, with its provenance.

    The record *id is kept*. Ids are the upsert key within a scope, so
    re-promoting the same fact overwrites its own earlier promotion instead of
    accumulating near-duplicates of it, which is what makes running
    consolidation twice harmless.
    """
    source = candidate.record.scope
    if not target.covers(source):
        raise ScopeViolation(
            f"promotion from {source.path!r} to {target.path!r} moves a memory *away* "
            "from the root. Visibility runs one way: a memory at a broader scope is "
            "visible to every scope beneath it, so a downward promotion republishes "
            "one scope's content as another's."
        )
    record = MemoryRecord(
        record_id=candidate.record.record_id,
        layer=candidate.record.layer,
        text=candidate.record.text,
        scope_path=target.path,
        salience=candidate.record.salience,
        origin=f"promotion:{promoted_by}",
        revision=candidate.record.revision,
    )
    return Promotion(
        record=record,
        provenance=build_provenance(
            source_conversation_id=candidate.conversation_id,
            source_scope_path=source.path,
            promoted_by=promoted_by,
            extraction_confidence=candidate.confidence,
            promoted_at=promoted_at,
        ),
        source_scope_path=source.path,
    )

quick_phases

quick_phases(*, graph: Any, episodic: Any | None = None, profile_fields: frozenset[str] | None = None, chat_model: Any | None = None, embedding_provider: Any | None = None, lookback_hours: float = 24.0, spreading_weight: float = 0.5, llm_summarise: Any | None = None, episodic_max_entries: int = 100, episodic_summarize_batch: int = 50, embedding_threshold: float = DEFAULT_EMBEDDING_THRESHOLD, lexical_threshold: float = DEFAULT_LEXICAL_THRESHOLD, max_pairs_per_run: int = 10) -> tuple[ConsolidationPhase, ...]

Build the complete QUICK roster, in roster order.

graph is the one hard requirement: three of the four phases read and write the semantic graph, and a roster built without one would be three phases that fail on their first call rather than a roster that was never composed. Everything else is optional, and its absence makes exactly one phase decline.

Source code in src/symfonic/capabilities/memory/phases/quick.py

def quick_phases(
    *,
    graph: Any,
    episodic: Any | None = None,
    profile_fields: frozenset[str] | None = None,
    chat_model: Any | None = None,
    embedding_provider: Any | None = None,
    lookback_hours: float = 24.0,
    spreading_weight: float = 0.5,
    llm_summarise: Any | None = None,
    episodic_max_entries: int = 100,
    episodic_summarize_batch: int = 50,
    embedding_threshold: float = DEFAULT_EMBEDDING_THRESHOLD,
    lexical_threshold: float = DEFAULT_LEXICAL_THRESHOLD,
    max_pairs_per_run: int = 10,
) -> tuple[ConsolidationPhase, ...]:
    """Build the complete QUICK roster, in roster order.

    ``graph`` is the one hard requirement: three of the four phases read and
    write the semantic graph, and a roster built without one would be three
    phases that fail on their first call rather than a roster that was never
    composed. Everything else is optional, and its absence makes exactly one
    phase decline.
    """
    if graph is None:
        raise MemoryContractError(
            "quick_phases needs a graph: strengthen, soul_corrections and "
            "semantic_merge all read and write the semantic graph, so a roster "
            "without one is three phases that would fail on their first call. "
            "Pass the ``GraphBackend`` your store is built over, or a "
            "``GraphMemoryStore`` if you have one."
        )
    graph = phase_graph(graph)
    window = RecentSemanticNodes(graph, lookback_hours=lookback_hours)
    return PhaseRoster((
        StrengthenPhase(graph, window, spreading_weight=spreading_weight),
        SoulCorrectionsPhase(graph, window, profile_fields=profile_fields),
        EpisodicSummaryPhase(
            episodic,
            llm_summarise=llm_summarise,
            max_entries=episodic_max_entries,
            summarize_batch=episodic_summarize_batch,
        ),
        SemanticMergePhase(
            graph,
            chat_model=chat_model,
            embedding_provider=embedding_provider,
            embedding_threshold=embedding_threshold,
            lexical_threshold=lexical_threshold,
            max_pairs_per_run=max_pairs_per_run,
        ),
    ), graph)

rank_key

rank_key(memory: RetrievedMemory) -> tuple[int, int, float, int, int, str]

Total order over retrieved memories.

A memory carrying a source_ordinal keeps the position its store gave it, and sorts ahead of everything that does not. That is not a preference for pre-ranked stores; it is the only way their order survives at all. The local key below is deterministic, which is what makes it dangerous: on equal scores, or on the all-None scores a keyword layer produces, it replaces an external ranking with a plausible-looking one and nothing looks wrong.

Everything else ranks by score, then nearer scope. An unscored memory uses 0.0 for ordering only; admission still preserves the absent signal.

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

def rank_key(memory: RetrievedMemory) -> tuple[int, int, float, int, int, str]:
    """Total order over retrieved memories.

    A memory carrying a ``source_ordinal`` keeps the position its store gave it,
    and sorts ahead of everything that does not. That is not a preference for
    pre-ranked stores; it is the only way their order survives at all. The local
    key below is *deterministic*, which is what makes it dangerous: on equal
    scores, or on the all-``None`` scores a keyword layer produces, it replaces
    an external ranking with a plausible-looking one and nothing looks wrong.

    Everything else ranks by score, then nearer scope. An unscored memory uses
    ``0.0`` for ordering only; admission still preserves the absent signal.
    """
    if memory.source_ordinal is not None:
        return (
            0 if memory.reserved else 1,
            0, float(memory.source_ordinal), 0, 0,
            memory.record.record_id,
        )
    return (
        0 if memory.reserved else 1,
        1,
        -(memory.score or 0.0),
        memory.scope_distance,
        layer_index(memory.record.layer),
        memory.record.record_id,
    )

read_reply

read_reply(response: Any) -> ProviderReply

Read response into text, or report that its shape is unknown.

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

def read_reply(response: Any) -> ProviderReply:
    """Read ``response`` into text, or report that its shape is unknown."""
    if isinstance(response, str):
        return ProviderReply(family=ProviderFamily.TEXT, text=response)
    if isinstance(response, Mapping):
        return _read_mapping(response)
    content = getattr(response, "content", None)
    if content is None:
        return ProviderReply(family=ProviderFamily.UNKNOWN)
    if isinstance(content, str):
        return ProviderReply(family=ProviderFamily.LANGCHAIN, text=content)
    if _is_block_list(content):
        text, ignored = _read_blocks(content)
        return ProviderReply(family=ProviderFamily.LANGCHAIN, text=text, ignored=ignored)
    return ProviderReply(family=ProviderFamily.UNKNOWN)

record_from_legacy_node

record_from_legacy_node(payload: Mapping[str, Any]) -> MemoryRecord

Read a stored legacy node back into a record.

A pre-v8 row with no scope_path reads as tenant-global — the same conservative backfill legacy's own dual-read applies, because those rows were tenant-global before the key existed.

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

def record_from_legacy_node(payload: Mapping[str, Any]) -> MemoryRecord:
    """Read a stored legacy node back into a record.

    A pre-v8 row with no ``scope_path`` reads as tenant-global — the same
    conservative backfill legacy's own dual-read applies, because those rows
    *were* tenant-global before the key existed.
    """
    bag = dict(payload.get("properties") or {})
    stored_path = bag.get(SCOPE_PATH_KEY)
    scope = (
        scope_from_legacy_path(str(stored_path))
        if isinstance(stored_path, str) and stored_path
        else MemoryScope(str(payload["tenant_id"]))
    )
    layer = payload.get("layer", MemoryLayer.SEMANTIC)
    stored_metadata = bag.get(METADATA_KEY)
    origin = str(bag.get(ORIGIN_KEY, ""))
    metadata = dict(stored_metadata) if isinstance(stored_metadata, Mapping) else {}
    label = str(payload.get("label", ""))
    # Expand a truncated display prefix, but preserve independent identifiers
    # such as Entity:person:mara_venn (whose content is a readable name).
    content = bag.get("content")
    text = (
        content if isinstance(content, str) and content.strip() and content.startswith(label)
        else label
    )
    return MemoryRecord(
        record_id=str(bag.get(RECORD_ID_KEY) or payload.get("id") or ""),
        layer=resolve_layer(str(getattr(layer, "value", layer))),
        text=retrievable_text(text, metadata) if origin.startswith("extraction:") else text,
        scope_path=scope.path,
        salience=importance_to_salience(payload.get("importance", 5.0)),
        origin=origin,
        # The other half of TA-3-1-9: this constructor took no ``metadata``
        # argument at all, so even a bag that carried it read back bare.
        metadata=metadata,
        # A legacy row may carry an authority this vocabulary does not know --
        # it was a free-form string there. Dropped rather than raised on the
        # read: refusing would make a scope with one such row unreadable, and
        # this is a reader, not the door that admits the claim.
        edited_by=_known_authority(bag.get(EDITED_BY_KEY)),
    )

resolve_contribution_scope

resolve_contribution_scope(scope: str | None) -> ContributionScope

Turn a caller's scope string into a member, or refuse it in-hierarchy.

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

def resolve_contribution_scope(scope: str | None) -> ContributionScope:
    """Turn a caller's scope string into a member, or refuse it in-hierarchy."""
    if scope is None:
        return ContributionScope.DEPLOYMENT
    try:
        return ContributionScope(scope)
    except ValueError as exc:
        raise MemoryContractError(
            f"scope {scope!r} is not a contribution scope; permitted values are "
            f"{[member.value for member in ContributionScope]}."
        ) from exc

resolve_layer

resolve_layer(value: str | MemoryLayer) -> MemoryLayer

Turn a caller's layer string into a member, or refuse it in-hierarchy.

MemoryLayer('reflective') raises a bare :class:ValueError, which a caller guarding on :class:~.errors.MemoryCapabilityError would miss. Every way of naming a layer wrong reports the same way.

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

def resolve_layer(value: str | MemoryLayer) -> MemoryLayer:
    """Turn a caller's layer string into a member, or refuse it in-hierarchy.

    ``MemoryLayer('reflective')`` raises a bare :class:`ValueError`, which a
    caller guarding on :class:`~.errors.MemoryCapabilityError` would miss. Every
    way of naming a layer wrong reports the same way.
    """
    try:
        return MemoryLayer(value)
    except ValueError as exc:
        raise MemoryContractError(
            f"{value!r} is not a memory layer; the Pentad is "
            f"{[member.value for member in LAYER_LADDER]}. An unrecognised layer is refused "
            "rather than guessed at — a misfiled memory is retrieved by the wrong turn."
        ) from exc

salience_to_importance

salience_to_importance(salience: float) -> float

Map [0, 1] salience back onto the legacy 1–10 grid.

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

def salience_to_importance(salience: float) -> float:
    """Map ``[0, 1]`` salience back onto the legacy 1–10 grid."""
    clamped = min(max(float(salience), 0.0), 1.0)
    return round(
        IMPORTANCE_FLOOR + clamped * (IMPORTANCE_CEILING - IMPORTANCE_FLOOR), 6
    )

scope_from_legacy_path

scope_from_legacy_path(path: str) -> MemoryScope

Read a stored scope_path back into a scope.

The kinds are discarded rather than validated: an adopter who named their levels org/brand/conversation has the same three-level hierarchy under different labels, and refusing their rows would make the migration a data conversion instead of a re-read.

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

def scope_from_legacy_path(path: str) -> MemoryScope:
    """Read a stored ``scope_path`` back into a scope.

    The kinds are discarded rather than validated: an adopter who named their
    levels ``org``/``brand``/``conversation`` has the same three-level
    hierarchy under different labels, and refusing their rows would make the
    migration a data conversion instead of a re-read.
    """
    parts = path.split(LEGACY_SCOPE_DELIMITER) if path else []
    if not parts or len(parts) % 2:
        raise MemoryContractError(
            f"legacy scope path {path!r} has {len(parts)} segments; the stored form is "
            "alternating kind/id pairs, so an odd count is a truncated write."
        )
    return MemoryScope(*parts[1::2])

scope_from_path

scope_from_path(path: str) -> MemoryScope

Parse the canonical path form back into a scope.

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

def scope_from_path(path: str) -> MemoryScope:
    """Parse the canonical path form back into a scope."""
    segments = path.split(SCOPE_SEPARATOR) if path else []
    if not 1 <= len(segments) <= 3:
        raise MemoryContractError(
            f"scope path {path!r} has {len(segments)} levels; a memory scope is "
            "tenant[/principal[/session]] — between one and three."
        )
    return MemoryScope(*segments)

select

select(memories: Iterable[RetrievedMemory], query: MemoryQuery, *, sources: Mapping[str, int] | None = None, unavailable: tuple[str, ...] = ()) -> RetrievalResult

Rank, filter, and cap what a store returned. Never mutates the input.

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

def select(
    memories: Iterable[RetrievedMemory],
    query: MemoryQuery,
    *,
    sources: Mapping[str, int] | None = None,
    unavailable: tuple[str, ...] = (),
) -> RetrievalResult:
    """Rank, filter, and cap what a store returned. Never mutates the input."""
    from symfonic.capabilities.memory.selection import select as admit
    return admit(memories, query, sources=sources, unavailable=unavailable)