Skip to content

symfonic.capabilities.memory

memory

Public facade for hierarchical memory capabilities.

Association dataclass 

Association(source_id: str, target: MemoryRecord, relationship: str = 'associated', weight: float = 1.0, source_label: str = '', selected_publication: PublishedRecallRef | None = None, linked_edge: LinkedEdgeWitness | None = None)

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 
71
72
73
74
75
76
77
78
79
80
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 
50
51
52
53
54
55
56
57
58
59
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 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
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)
    if (getattr(self._domain, "supports_bounded_workset", True) is False
            and getattr(runtime, "_edge_limit", None) is not None):
        raise MemoryContractError(
            "receipt domain lacks a canonical bounded workset for edge-limited cycles"
        )
    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.

requires_operation property 

requires_operation: bool

Whether every direct cycle needs a caller-persisted receipt identity.

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 
193
194
195
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 
187
188
189
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 = '', pending_connections: Sequence[Mapping[str, Any]] | None = None, operation: MemoryOperation | None = None) -> ConsolidationState

Run one cycle over one scope, alone.

pending_connections is a per-call override of the composed inferred edges: None keeps the default, an empty sequence means none.

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 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
async def run(
    self,
    scope: MemoryScope,
    cycle: ConsolidationCycle,
    *,
    run_id: str = "",
    root_run_id: str = "",
    pending_connections: Sequence[Mapping[str, Any]] | None = None,
    operation: receipt_contract.MemoryOperation | None = None,
) -> ConsolidationState:
    """Run one cycle over one scope, alone.

    ``pending_connections`` is a per-call override of the composed inferred
    edges: ``None`` keeps the default, an empty sequence means none.

    ``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()
    receipt_contract.validate_transaction_operation(
        self._domain, operation, receipt_contract.MemoryOperationVerb.COMMIT_CYCLE, scope,
    )
    if (scope.identity_version and getattr(self._domain, "requires_native_exact_quick", False)
            and (cycle is not ConsolidationCycle.QUICK
                 or not getattr(self._runtime, "native_exact_owner", False))):
        raise MemoryContractError("receipt-backed v1 requires an exact owner QUICK roster")
    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:
        reserve = getattr(self._domain, "reserve", None)
        if callable(reserve) and not await reserve(operation, owner=lease.owner):
            state = ConsolidationState(
                scope_path=scope.path, cycle=cycle, registered=self.registered,
                run_id=run_id, root_run_id=root_run_id,
            )
            state.replayed = True
            state.finished_at = datetime.now(UTC)
            return state
        # 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,
                    pending_connections=pending_connections,
                    # 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,
                    operation=operation,
                )
    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.
        try:
            abandon = getattr(self._domain, "abandon_reservation", None)
            if callable(abandon):
                abandon(operation, owner=lease.owner)
        finally:
            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 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
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._native_exact_owner = bool(getattr(phases, "native_exact_owner", False))
    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 phase graph, declared by a shipped factory or the caller.

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, pending_connections: Sequence[Mapping[str, Any]] | None = None, operation: MemoryOperation | None = 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 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
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,
    pending_connections: Sequence[Mapping[str, Any]] | None = None,
    operation: receipt_contract.MemoryOperation | None = 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")
    receipt_contract.validate_transaction_operation(
        domain, operation, receipt_contract.MemoryOperationVerb.COMMIT_CYCLE, scope,
    )

    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)
    overrides = None if pending_connections is None else tuple(map(dict, pending_connections))
    context = PhaseContext(
        scope=scope,
        cycle=cycle,
        started_at=state.started_at,
        writes=buffered,
        run_id=run_id,
        root_run_id=root_run_id,
        pending_connections=overrides,
    )
    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, operation=operation
        )
    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 
142
143
144
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 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
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 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
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, replayed: 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 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
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 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
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 
66
67
68
69
70
71
72
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 
55
56
57
58
59
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 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
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 = compile_credential_pattern(parts)

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 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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 
108
109
110
111
112
113
114
115
116
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))

EdgeDeleteService 

EdgeDeleteService(backend: GraphBackend | None)

Delete an edge visible to a caller, preserving backend no-op semantics.

Source code in src/symfonic/capabilities/memory/edge_delete.py 
16
17
def __init__(self, backend: GraphBackend | None) -> None:
    self._backend = backend

EpisodicCreateService 

EpisodicCreateService(graph: Any, vectors: Any, embedder: Any)

Write an event through the vector-backed episodic layer at a resolved scope.

Source code in src/symfonic/capabilities/memory/episodic_create.py 
28
29
30
31
32
33
34
35
36
37
def __init__(self, graph: Any, vectors: Any, embedder: Any) -> None:
    if graph is None:
        raise ValueError("episodic create needs a graph authority backend")
    if vectors is None or embedder is None:
        raise ValueError("episodic create needs a vector backend and embedder")
    self._graph = GraphMemoryStore(graph)
    self._layer = EpisodicLayer(vectors, embedder)
    # The live HTTP path removes credential-shaped metadata keys. It does
    # not scan content or metadata values, so preserve that wire behavior.
    self._scrubber = CredentialScrubber(value_patterns=())

EpisodicLifecycleEvent dataclass 

EpisodicLifecycleEvent(occurred_at: str, scope_path: str, committed_record_ids: tuple[str, ...], run_id: str, root_run_id: str, parent_run_id: str | None = None)

A committed episodic lifecycle receipt safe to send to a sink.

run_id, root_run_id and parent_run_id identify the invocation which ran the lifecycle stage. They intentionally describe the lifecycle invocation, not authorship of every committed record: a flush may publish pending work from an earlier interrupted invocation. scope_path is the exact scope supplied to flush; callers must scope any downstream access independently.

EpisodicTelemetrySink

Bases: Protocol

Receives successful native episodic lifecycle publications.

emit async 

emit(event: EpisodicLifecycleEvent) -> None

Observe one successful lifecycle publication.

Source code in src/symfonic/capabilities/memory/telemetry.py 
57
58
59
async def emit(self, event: EpisodicLifecycleEvent) -> None:
    """Observe one successful lifecycle publication."""
    ...

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 
183
184
185
186
187
188
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.

ForgetInProgress dataclass 

ForgetInProgress(scope_path: str, job_id: str, discarded: tuple[str, ...] = ())

A forget whose erasure has not finished.

Not a receipt and not a refusal: a durable job committed, and from that commit the subtree reads as erased to every reader and every write under it refuses. What remains is erased in later transactions: claim identities, and, when the content exceeded one transaction, the content itself, which no read returns meanwhile. Forget the same scope again to advance it; only a :class:LifecycleReceipt reports the forget complete.

GraphAdminService 

GraphAdminService(graph: Any, records: Any = None, *, conversation: 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 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def __init__(
    self, graph: Any, records: Any = None, *, conversation: 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
    self._conversation = conversation if conversation is not None else getattr(
        records, "_conversation", None
    )

clusters async 

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

Published semantic label groups, including visible ancestors.

Unlike legacy clusters, this hides staged and quarantined writes. It is not the descendant inventory served by :meth:nodes. Reads never increment counters.

Source code in src/symfonic/capabilities/memory/graph_admin.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
async def clusters(self, scope: MemoryScope) -> list[dict[str, Any]]:
    """Published semantic label groups, including visible ancestors.

    Unlike legacy clusters, this hides staged and quarantined writes. It
    is not the descendant inventory served by :meth:`nodes`. Reads never
    increment counters.
    """
    from symfonic.memory.pending import is_pending
    from symfonic.memory.retraction import is_retracted
    from symfonic.memory.types import MemoryLayer

    nodes = await self._graph.query_nodes(
        tenant_scope(scope), {"layer": MemoryLayer.SEMANTIC}, limit=None,
    )
    groups: dict[str, list[Any]] = {}
    for node in nodes:
        if is_pending(node.properties) or is_retracted(node.properties):
            continue
        label = str(node.label)
        prefix = label.split(":", 1)[0].strip() if ":" in label else (
            label.split()[0] if label else "unknown"
        )
        groups.setdefault(prefix, []).append(node)
    return [
        {
            "prefix": prefix,
            "count": len(members),
            "members": [{"id": str(node.id), "label": str(node.label)}
                        for node in members],
            "representative": {
                "id": str(best.id), "label": str(best.label),
            },
        }
        for prefix, members in sorted(
            groups.items(), key=lambda item: len(item[1]), reverse=True,
        )
        for best in [max(members, key=lambda node: node.importance)]
    ]

edges async 

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

One bounded page of relationships, in newest backend order.

Source code in src/symfonic/capabilities/memory/graph_admin.py 
74
75
76
77
78
79
80
81
82
async def edges(
    self, scope: MemoryScope, *, limit: int = 500, offset: int = 0, admit: Any = None,
) -> 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
    )
    from symfonic.capabilities.memory.graph_owner_reads import admitted_edges
    return [edge_body(edge) for edge in await admitted_edges(self._graph, scope, found, admit)]

export async 

export(scope: MemoryScope, *, admit: Any = None) -> 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 
231
232
233
234
235
236
237
238
239
240
241
242
async def export(self, scope: MemoryScope, *, admit: Any = None) -> 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.
    """
    from symfonic.capabilities.memory.graph_owner_reads import export
    return await export(
        self, scope, admit, body=_record_body, schema_version=EXPORT_SCHEMA_VERSION,
    )

neighborhood async 

neighborhood(scope: MemoryScope, node_id: str, *, depth: int = 2, min_weight: float = 0, admit: Any = None) -> dict[str, Any]

A visible center, filtered BFS neighbors, and induced edges.

Neighbor reachability is independent of min_weight: that parameter filters only returned edges, as on the established graph route.

Source code in src/symfonic/capabilities/memory/graph_admin.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
async def neighborhood(
    self, scope: MemoryScope, node_id: str, *, depth: int = 2,
    min_weight: float = 0, admit: Any = None,
) -> dict[str, Any]:
    """A visible center, filtered BFS neighbors, and induced edges.

    Neighbor reachability is independent of ``min_weight``: that parameter
    filters only returned edges, as on the established graph route.
    """
    from symfonic.memory.retraction import is_retracted
    from symfonic.memory.types import NodeId

    # An overriding subclass must declare support for its own query_edges;
    # inheriting the parent's marker does not certify a replacement.
    if type(self._graph).__dict__.get("supports_visible_edge_owners") is not True:
        raise RuntimeError("graph backend lacks visible edge owner reads")
    trusted = tenant_scope(scope)
    center = await self._graph.get_node(trusted, NodeId(node_id))
    if center is None or is_retracted(center.properties) or (
        admit is not None and not await admit(center)
    ):
        raise GraphNodeNotFound(f"Node {node_id} not found")

    visited = {node_id}
    neighbors: list[dict[str, Any]] = []
    queue = deque([(node_id, 0)])
    while queue:
        current, distance = queue.popleft()
        if distance >= depth:
            continue
        incident = await self._graph.query_edges(
            trusted, {"incident_node_id": current, "visible_edge_owners": True},
            limit=None,
        )
        for edge in incident:
            source, target = str(edge.source), str(edge.target)
            if current not in (source, target):
                continue
            neighbor_id = target if source == current else source
            if neighbor_id in visited:
                continue
            node = await self._graph.get_node(trusted, NodeId(neighbor_id))
            if node is None or is_retracted(node.properties) or (
                admit is not None and not await admit(node)
            ):
                continue
            visited.add(neighbor_id)
            neighbors.append({**_node_body(node), "depth": distance + 1})
            queue.append((neighbor_id, distance + 1))

    # The backend filters both endpoints before sorting or limiting. A
    # single call avoids repeatedly scanning unrelated tenant edges and
    # keeps the read proportional to the output's induced edge set.
    induced = await self._graph.query_edges(
        trusted, {"induced_node_ids": tuple(sorted(visited)),
                  "visible_edge_owners": True},
        limit=None,
    )
    edges = []
    for edge in induced:
        if (str(edge.source) not in visited or str(edge.target) not in visited
            or edge.weight < min_weight):
            continue
        serialized = edge_body(edge)
        edges.append({key: serialized[key] for key in (
            "id", "source", "target", "relationship", "weight",
        )})
    return {"center": _node_body(center), "neighbors": neighbors, "edges": edges}

nodes async 

nodes(scope: MemoryScope, *, limit: int = 500, offset: int = 0, admit: Any = None) -> 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 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
async def nodes(
    self, scope: MemoryScope, *, limit: int = 500, offset: int = 0, admit: Any = None,
) -> 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)
    from symfonic.capabilities.memory.graph_owner_reads import admitted_nodes
    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 await admitted_nodes(found, admit)]
    # 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 await admitted_nodes(found[offset:through], admit)]

shortest_path async 

shortest_path(scope: MemoryScope, start: str, end: str, *, admit: Any = None) -> dict[str, Any]

Fewest-hop published path through visible nodes, without counters.

Source code in src/symfonic/capabilities/memory/graph_admin.py 
84
85
86
87
88
89
90
async def shortest_path(
    self, scope: MemoryScope, start: str, end: str, *, admit: Any = None,
) -> dict[str, Any]:
    """Fewest-hop published path through visible nodes, without counters."""
    from symfonic.capabilities.memory.graph_paths import shortest_path

    return await shortest_path(self._graph, scope, start, end, admit=admit)

GraphAuditService 

GraphAuditService(backend: GraphBackend)

Project the legacy audit response without composing an agent.

Source code in src/symfonic/capabilities/memory/graph_audit.py 
21
22
def __init__(self, backend: GraphBackend) -> None:
    self._graph = GraphMemoryStore(backend)

audit async 

audit(scope: MemoryScope, *, min_edge_weight: float = 1.5, admit: Any = None) -> dict[str, Any]

Count all visible semantic nodes and at most 10,000 visible edges.

Source code in src/symfonic/capabilities/memory/graph_audit.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
async def audit(
    self, scope: MemoryScope, *, min_edge_weight: float = 1.5, admit: Any = None,
) -> dict[str, Any]:
    """Count all visible semantic nodes and at most 10,000 visible edges."""
    scoped = tenant_scope(scope)
    nodes = await self._graph.query_nodes(scoped, layer=MemoryLayer.SEMANTIC)
    edges = await self._graph.list_edges(scoped, limit=10_000)
    from symfonic.capabilities.memory.graph_owner_reads import admitted_edges
    edges = await admitted_edges(self._graph.backend, scope, edges, admit)
    total_nodes, total_edges = len(nodes), len(edges)
    density = (
        round(2 * total_edges / (total_nodes * (total_nodes - 1)), 4)
        if total_nodes > 1 else 0.0
    )

    stale_before = datetime.now(UTC) - timedelta(days=7)
    orphans: list[dict[str, Any]] = []
    for node in nodes:
        neighbors = await self._graph.get_neighbors(scoped, node.id)
        edge_count = len(neighbors)
        if edge_count <= 1 and node.importance < 4.0 and node.updated_at < stale_before:
            reason = ", ".join((
                "no edges" if edge_count == 0 else "only 1 edge",
                f"importance {node.importance} < 4.0",
                "stale (>7 days)",
            ))
            orphans.append({
                "id": str(node.id), "label": str(node.label),
                "importance": node.importance, "edge_count": edge_count,
                "reason": reason,
            })

    weak = [
        {
            "id": str(edge.id), "source": str(edge.source),
            "target": str(edge.target), "weight": edge.weight,
            "reason": f"weight {edge.weight} below {min_edge_weight} threshold",
        }
        for edge in edges if edge.weight < min_edge_weight
    ]
    return {
        "total_nodes": total_nodes, "total_edges": total_edges,
        "density": density, "orphan_nodes": orphans, "weak_edges": weak,
        "recommendations": {
            "prune_orphans": len(orphans), "prune_weak_edges": len(weak),
        },
    }

GraphBackedHms 

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

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

Graph and vector candidates merge by record ID before final ranking.

Source code in src/symfonic/capabilities/memory/graph_store.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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._vectors = vectors
    self._direct_claims: dict[tuple[str, str], str] = {}
    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 graph store used by consolidation's domain check.

transaction_participants property 

transaction_participants: tuple[Any, ...]

Backends participating in mutation and vector publication.

discard async 

discard(scope: MemoryScope, *, operation: Any = None) -> 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 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
async def discard(self, scope: MemoryScope, *, operation: Any = None) -> 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.
    """
    from symfonic.capabilities.memory.graph_claimed_lifecycle import discard_pending
    from symfonic.capabilities.memory.graph_direct_claims import external_transaction, retire
    from symfonic.capabilities.memory.mutations import lifecycle_fence
    graph_receipt.validate(
        self, operation, graph_receipt.MemoryOperationVerb.DISCARD, scope)
    self._check_mutation(scope, graph_receipt.MemoryOperationVerb.DISCARD)
    outer = external_transaction(self, scope)
    async with lifecycle_fence(
        self._graph, self._vectors, scope, self._mutations,
        operation=operation, verb=graph_receipt.MemoryOperationVerb.DISCARD,
    ):
        claims = self._direct_claims.copy()
        # Sealed terminal procedures survive; the rest go in one transaction.
        discarded = await discard_pending(self, await self._pending_under(scope))
    retire(self, scope, discarded, outer, claims)
    return LifecycleReceipt(scope_path=scope.path, discarded=tuple(sorted(discarded)))

forget async 

forget(scope: MemoryScope, *, operation: Any = None) -> LifecycleReceipt | ForgetInProgress

Erase the subtree; native PG reports deleted IDs atomically. Other backends still inventory before deletion and can omit late rows. A composed CORE-199 eraser answers ForgetInProgress until its job ends.

Source code in src/symfonic/capabilities/memory/graph_store.py 
257
258
259
260
261
262
263
264
265
266
267
268
async def forget(
    self, scope: MemoryScope, *, operation: Any = None,
) -> LifecycleReceipt | ForgetInProgress:
    """Erase the subtree; native PG reports deleted IDs atomically.
    Other backends still inventory before deletion and can omit late rows.
    A composed CORE-199 eraser answers ForgetInProgress until its job ends.
    """
    from symfonic.capabilities.memory.graph_forget import forget_subtree
    graph_receipt.validate(
        self, operation, graph_receipt.MemoryOperationVerb.FORGET, scope)
    self._check_mutation(scope, graph_receipt.MemoryOperationVerb.FORGET)
    return await forget_subtree(self, scope, operation=operation)

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 
135
136
137
138
async def retrieve(self, query: MemoryQuery) -> RetrievalResult:
    """Both routes, merged by ``record_id`` and ranked once."""
    candidates, sources, unavailable = await graph_receipt.gather(self, 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).

Retrieval's select would apply limits this scan leaves to callers.

Vector candidates retain their own similarity-search top-k budget.

Source code in src/symfonic/capabilities/memory/graph_store.py 
183
184
185
186
187
188
189
190
191
192
193
async def scan_candidates(self, query: MemoryQuery) -> RetrievalResult:
    """Every visible candidate, uncapped by ``query.limit`` (CandidateScan).

    Retrieval's ``select`` would apply limits this scan leaves to callers.

    Vector candidates retain their own similarity-search top-k budget.
    """
    candidates, sources, unavailable = await graph_receipt.gather(self, 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 
114
115
116
117
118
119
120
121
122
123
124
125
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 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
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 
202
203
204
205
206
207
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 | ForgetInProgress

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

Passes a CORE-199 ForgetInProgress through: not a receipt, not done.

Source code in src/symfonic/capabilities/memory/bridge.py 
209
210
211
212
213
214
215
216
217
async def forget(self, scope: MemoryScope) -> LifecycleReceipt | ForgetInProgress:
    """Erase a subtree. The privacy seam (SEC-PRIV), not an invocation stage.

    Passes a CORE-199 ``ForgetInProgress`` through: not a receipt, not done.
    """
    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 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
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_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 
188
189
190
191
192
193
194
195
196
197
198
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 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
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 
71
72
73
74
75
76
77
78
79
80
81
82
83
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] = _stamped(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 
201
202
203
204
205
206
207
208
209
210
211
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 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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 (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 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
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]] = {}
    _TABLES.add(self)

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 
218
219
220
221
222
223
224
225
226
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)

Bases: JournalCreateOnly

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 
255
256
257
258
259
260
261
262
263
264
265
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
    )
    if all(callable(getattr(self._durable, name, None))
           for name in ("read_owned_node", "patch_owned_node")):
        from symfonic.capabilities.memory.journal_node_patch import bind_graph
        bind_graph(self)

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 
150
151
152
153
154
155
156
157
158
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 remains collaborator-selected: no entity extractor is a normal composition. Phase 12's extractor selector is different: the legacy setting chose between two extractors that can both run when a procedural destination is composed. It therefore reaches the roster as an explicit selector; asking for the LLM branch without composing its extractor is refused at composition time.

Source code in src/symfonic/capabilities/memory/policy.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
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`` remains collaborator-selected: no entity
    extractor is a normal composition.  Phase 12's extractor selector is
    different: the legacy setting chose between two extractors that can
    both run when a procedural destination is composed.  It therefore
    reaches the roster as an explicit selector; asking for the LLM branch
    without composing its extractor is refused at composition time.
    """
    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,
        "synthetic_min_co_count": self.synthetic_link_min_co_count,
        "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,
        "min_pattern_count": self.promotion_min_pattern_count,
        "recency_days": self.promotion_recency_days,
        "max_drafts_per_run": self.promotion_max_drafts_per_run,
        "use_tool_calls_fallback": self.promotion_use_tool_calls_fallback,
        "promote_assistant_content": self.promotion_promote_assistant_content,
        "use_llm_extractor": self.phase_12_use_llm_extractor,
        "llm_max_episodes_per_run": self.phase_12_llm_max_episodes_per_run,
        "llm_max_drafts_per_run": self.phase_12_llm_max_drafts_per_run,
    }

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 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@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, generation: str | None = None)

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 a scope and its ancestors and descendants.

acquire async 

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

Take scope, or None if an overlapping scope is held.

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 
104
105
106
107
108
109
110
111
112
async def acquire(self, scope: MemoryScope, *, owner: str, ttl_seconds: float) -> Lease | None:
    """Take ``scope``, or ``None`` if an overlapping scope is held.

    ``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 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
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 
114
115
116
117
118
119
120
121
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 
151
152
153
154
155
156
157
158
159
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 
141
142
143
144
145
146
147
148
149
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.
    """
    ...

LegacyGraphOperationsAdapter 

LegacyGraphOperationsAdapter(native: MemoryExtractionService)

Make a native extractor accept the bundled legacy directive explicitly.

Compose this around :class:MemoryExtractionService only when rendering a bundled legacy extraction template. Native action/properties operations retain their normal service path, which lets a template migrate independently without a behavior switch in the post-response stage.

Source code in src/symfonic/capabilities/memory/legacy_extraction.py 
54
55
def __init__(self, native: MemoryExtractionService) -> None:
    self._native = native

extract async 

extract(request: ExtractionRequest) -> ExtractionResult

Delegate fallback extraction without changing its provider contract.

Source code in src/symfonic/capabilities/memory/legacy_extraction.py 
57
58
59
async def extract(self, request: ExtractionRequest) -> ExtractionResult:
    """Delegate fallback extraction without changing its provider contract."""
    return await self._native.extract(request)

parse_inline 

parse_inline(operations: tuple[Mapping[str, Any], ...], request: ExtractionRequest) -> ExtractionResult

Translate an all-legacy envelope or retain the native inline path.

Source code in src/symfonic/capabilities/memory/legacy_extraction.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def parse_inline(
    self, operations: tuple[Mapping[str, Any], ...], request: ExtractionRequest
) -> ExtractionResult:
    """Translate an all-legacy envelope or retain the native inline path."""
    if not operations:
        return self._native.parse_inline(operations, request)

    legacy = [
        isinstance(operation, Mapping)
        and "type" in operation
        and "action" not in operation
        for operation in operations
    ]
    native = [
        isinstance(operation, Mapping)
        and "action" in operation
        and "type" not in operation
        for operation in operations
    ]
    if all(native):
        return self._native.parse_inline(operations, request)
    if not all(legacy):
        return self._rejected(
            request,
            "the tagged operations mix native, legacy, or malformed schemas",
        )

    translated: list[Mapping[str, Any]] = []
    dropped: list[tuple[str, str]] = []
    for index, operation in enumerate(operations):
        converted, reason = _translate(operation)
        if reason:
            dropped.append((f"op[{index}]", reason))
        elif converted is not None:
            translated.append(converted)
    if dropped:
        return ExtractionResult(
            scope=request.scope,
            turn=request.turn,
            dropped=tuple(dropped),
            family=ProviderFamily.TEXT,
            reason="the inline legacy extraction schema was rejected",
        )
    return self._native.parse_inline(tuple(translated), request)

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, conversation: Any = None, operation_journal: 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.

 required
operation_journal Any

only a native Mongo v1 store's forget records in it.

 None
Source code in src/symfonic/capabilities/memory/admin.py 
52
53
54
55
56
57
58
59
60
61
def __init__(self, store: Any, *, procedural: Any = None,
             conversation: Any = None, operation_journal: Any = None) -> None:
    """
    Args:
        store: an HMS satisfying the retrieval, write and lifecycle ports.
            Held privately and never returned.
        operation_journal: only a native Mongo v1 store's forget records in it.
    """
    self._store, self._procedural = store, procedural
    self._conversation, self._journal = conversation, operation_journal

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 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
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")
    try:
        return await layer.approve_skill(
            _legacy(scope), _node_id(procedure),
            action_tool=action_tool, precondition=precondition,
        )
    except KeyError as exc:
        if action_tool is not None or precondition is not None:
            raise MemoryContractError(
                "cannot bind a procedure outside the supplied scope"
            ) from exc
        raise

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 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
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

decide_draft_procedure async 

decide_draft_procedure(scope: MemoryScope, procedure: Any, *, status: str, action_tool: str | None = None, precondition: Any = None, expected_source: object | None = None, expected_source_present: bool = True, expected_action_tool: str | None = None, expected_action_tool_present: bool = False, actor: str | None = None) -> Any | None

Apply an automated decision only while the scoped row is a draft.

The layer re-reads and serialises the native in-process transition.

Source code in src/symfonic/capabilities/memory/admin.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
async def decide_draft_procedure(
    self,
    scope: MemoryScope,
    procedure: Any,
    *,
    status: str,
    action_tool: str | None = None,
    precondition: Any = None,
    expected_source: object | None = None,
    expected_source_present: bool = True,
    expected_action_tool: str | None = None,
    expected_action_tool_present: bool = False,
    actor: str | None = None,
) -> Any | None:
    """Apply an automated decision only while the scoped row is a draft.

    The layer re-reads and serialises the native in-process transition.
    """
    _validate_draft_decision(status, action_tool, precondition)
    layer = self._require_procedural("apply an automated procedure decision")
    return await layer.decide_draft_skill(
        _legacy(scope),
        _node_id(procedure),
        status=status,
        action_tool=action_tool,
        precondition=precondition,
        expected_source=expected_source,
        expected_source_present=expected_source_present,
        expected_action_tool=expected_action_tool,
        expected_action_tool_present=expected_action_tool_present,
        actor=actor,
    )

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 
73
74
75
76
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 
268
269
270
async def discard(self, scope: MemoryScope) -> Any:
    """Drop staging without committing any of it."""
    return await operator_call(self._store, scope, "discard", scope)

forget async 

forget(scope: MemoryScope) -> Any

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

Idempotent: an empty scope answers with an empty receipt. When composed with a conversation log, erase that scope's transcript as well. A CORE-199 ForgetInProgress means an erase job remains; the transcript waits for the forget that completes it.

Source code in src/symfonic/capabilities/memory/admin.py 
272
273
274
275
276
277
278
279
280
281
async def forget(self, scope: MemoryScope) -> Any:
    """Erase ``scope`` and everything below it, staged and committed alike.

    Idempotent: an empty scope answers with an empty receipt. When composed
    with a conversation log, erase that scope's transcript as well.
    A CORE-199 ``ForgetInProgress`` means an erase job remains; the
    transcript waits for the forget that completes it.
    """
    from symfonic.capabilities.memory.admin_erase import forget
    return await forget(self._store, self._conversation, scope, self._journal)

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 
68
69
70
71
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 
78
79
80
81
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 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
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 
264
265
266
async def publish(self, scope: MemoryScope) -> Any:
    """Commit what ``scope`` staged, and what anything below it staged."""
    return await operator_call(self._store, scope, "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 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
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 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
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 
175
176
177
178
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.

Over raw PostgreSQL v1 memory it presents the running turn's erase admission and, outside a run or in one that captured none, refuses (CORE-199, operator_admission): it never observes a fresh one.

Source code in src/symfonic/capabilities/memory/admin.py 
115
116
117
118
119
120
121
122
123
124
async def stage(
    self, scope: MemoryScope, records: tuple[MemoryRecord, ...]
) -> Any:
    """Prepare ``records`` under ``scope``. Nothing is durable until publish.

    Over raw PostgreSQL v1 memory it presents the running turn's erase
    admission and, outside a run or in one that captured none, refuses
    (CORE-199, ``operator_admission``): it never observes a fresh one.
    """
    return await operator_call(self._store, scope, "write", WriteRequest(scope, records))

MemoryBundleFactory 

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

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

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

Parameters:

Name Type Description Default
store Any

an HMS satisfying the retrieval, write and lifecycle ports. One object for all three because staging, publishing and erasing are operations on one place -- and because a deployment that split them would have to answer what happens when only two are present, which the ports already refuse.

 required
limit int

how many memories a turn recalls.

 5
recall_budget RecallBudget | None

explicit UTF-8 block/item ceilings; independent of working-turn retention. None preserves legacy character caps.

 None
extractor MemoryExtractorPort | None

a :class:~.ports.MemoryExtractorPort, normally :class:~.extraction.MemoryExtractionService built over an extraction model. Given one, the capability runs it after the final model round and writes what it returns, so a turn that said something worth keeping is remembered without the deployment calling anything. None -- the default -- means this deployment does not extract: it still recalls, records the exchange and erases. Checked here rather than mid-turn; see :func:~.ports.validate_extractor.

 None
summary STMSummaryPort | None

an explicit evicted-turn summary collaborator. It is materialized only into JIT prompts; omitting it disables the contribution and makes no source call.

 None
reflection ReflectionPort | None

a completed-exchange insight collaborator. It runs after publication and persists accepted scoped insights.

 None
consolidation Any | None

a :class:~.napping.ConsolidationCoordinator. Given one, the capability naps on the cadence it carries -- after the turn's memories are published, on the run's background registry, once per scope. None -- the default -- means this deployment consolidates from its own scheduler or not at all, and one shared coordinator across scopes is correct: it counts turns per scope and locks per scope.

 None

Raises:

Type Description
ConfigurationError

if extractor, summary, or reflection cannot serve its port.
Source code in src/symfonic/capabilities/memory/factory.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def __init__(
    self,
    store: Any,
    *,
    limit: int = 5,
    recall_budget: RecallBudget | None = None,
    extractor: MemoryExtractorPort | None = None,
    summary: STMSummaryPort | None = None,
    reflection: ReflectionPort | None = None,
    consolidation: Any | None = None,
    conversation: Any | None = None,
    recent_turns: int = 0,
    activation: Any | None = None,
    retrieval_policy: RetrievalPolicy | None = None,
    retrieval: RetrievalCoordinator | None = None,
    exclude_speakers: frozenset[str] = frozenset(),
    episodic_telemetry_sink: EpisodicTelemetrySink | None = None,
    operation_journal: MemoryOperationJournalPort | None = None,
    generations: Any = None,  # minted by memory_capabilities only (CORE-199)
    triage: IntentTriageCapability | None = None,
) -> None:
    """
    Args:
        store: an HMS satisfying the retrieval, write and lifecycle ports.
            One object for all three because staging, publishing and
            erasing are operations on one place -- and because a
            deployment that split them would have to answer what happens
            when only two are present, which the ports already refuse.
        limit: how many memories a turn recalls.
        recall_budget: explicit UTF-8 block/item ceilings; independent of
            working-turn retention. None preserves legacy character caps.
        extractor: a :class:`~.ports.MemoryExtractorPort`, normally
            :class:`~.extraction.MemoryExtractionService` built over an
            extraction model. Given one, the capability runs it after the
            final model round and writes what it returns, so a turn that
            said something worth keeping is remembered without the
            deployment calling anything. ``None`` -- the default -- means
            this deployment does not extract: it still recalls, records
            the exchange and erases. Checked here rather than mid-turn;
            see :func:`~.ports.validate_extractor`.
        summary: an explicit evicted-turn summary collaborator. It is
            materialized only into JIT prompts; omitting it disables the
            contribution and makes no source call.
        reflection: a completed-exchange insight collaborator. It runs
            after publication and persists accepted scoped insights.
        consolidation: a
            :class:`~.napping.ConsolidationCoordinator`. Given one, the
            capability naps on the cadence it carries -- after the turn's
            memories are published, on the run's background registry, once
            per scope. ``None`` -- the default -- means this deployment
            consolidates from its own scheduler or not at all, and one
            shared coordinator across scopes is correct: it counts turns
            per scope and locks per scope.

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

for_scope 

for_scope(scope: MemoryScope | Any) -> Any

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

Source code in src/symfonic/capabilities/memory/factory.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def for_scope(self, scope: MemoryScope | Any) -> Any:
    """Compose kind-aware paths, or an explicitly positional MemoryScope."""
    from symfonic.capabilities.memory.operation_journal import receipt_backed
    memory_scope = composition_scope(scope)
    if receipt_backed(self._store):
        from symfonic.capabilities.memory.receipt_composition import receipt_for_scope
        return receipt_for_scope(self, memory_scope)
    if memory_scope.identity_version:
        from symfonic.capabilities.memory.graph_native_admission import require_composition
        require_composition(self, memory_scope)
    from symfonic.capabilities.memory.turn_generation import guarded_writer
    writer = guarded_writer(self._store, self._conversation, memory_scope, self._generations)
    hydrator = hydrator_for(
        self._store, conversation=self._conversation, recent_turns=self._recent_turns,
        activation=self._activation,
        retrieval_policy=self._retrieval_policy,
        retrieval=self._retrieval,
        exclude_speakers=self._exclude_speakers,
    )
    from symfonic.capabilities.memory.port_inspection import pair_guard
    return MemoryCapability(
        hydrator,
        scope=memory_scope,
        limit=self._limit,
        recall_budget=self._recall_budget,
        writer=pair_guard(writer, hydrator, memory_scope),
        # Closed over the scope, not read from the turn. ``Agent.run``
        # takes a prompt and nothing about tenancy -- that is the published
        # facade decision -- so the turn request carries no scope and a
        # producer that looked for one filed nothing at all. The agent IS
        # the scope: it was composed for exactly this one, which is the
        # same fact that makes a host keep one agent per scope.
        records_from=_producer_for(memory_scope),
        lifecycle=writer,
        extractor=self._extractor,
        summary=self._summary,
        reflection=self._reflection,
        consolidation=self._consolidation,
        episodic_telemetry_sink=self._episodic_telemetry_sink,
        operation_journal=self._operation_journal,
        triage=self._triage,
    )

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, jit_only: bool = False)

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 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
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 
112
113
114
115
116
117
118
119
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 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
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 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
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)

parse_inline 

parse_inline(operations: tuple[Mapping[str, Any], ...], request: ExtractionRequest) -> ExtractionResult

Mint operations emitted by the response model's directive.

The response adapter has already recognised and stripped the envelope, so this method receives only JSON objects. It deliberately shares the same minting path as a dedicated extraction model: inline operations get the same validation, credential scrubbing, write receipt and lifecycle publication rather than becoming an unreviewed side path.

Source code in src/symfonic/capabilities/memory/extraction.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def parse_inline(
    self, operations: tuple[Mapping[str, Any], ...], request: ExtractionRequest
) -> ExtractionResult:
    """Mint operations emitted by the response model's directive.

    The response adapter has already recognised and stripped the envelope,
    so this method receives only JSON objects.  It deliberately shares the
    same minting path as a dedicated extraction model: inline operations
    get the same validation, credential scrubbing, write receipt and
    lifecycle publication rather than becoming an unreviewed side path.
    """
    legacy = tuple(
        (f"op[{index}]", "the tagged operation uses the legacy type/payload schema")
        for index, operation in enumerate(operations)
        if "type" in operation and "action" not in operation
    )
    if legacy:
        return ExtractionResult(
            scope=request.scope,
            turn=request.turn,
            dropped=legacy,
            family=ProviderFamily.TEXT,
            reason="the inline extraction schema is not supported by this service",
        )
    return self._mint({"ops": list(operations)}, request, ProviderFamily.TEXT, ())

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 
121
122
123
124
125
126
127
128
129
130
131
132
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 
190
191
192
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 
154
155
156
async def flush(self, scope: MemoryScope) -> LifecycleReceipt:
    """Commit every pending memory in ``scope`` and its descendants."""
    ...

forget async 

forget(scope: MemoryScope) -> LifecycleReceipt | ForgetInProgress

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.

Only a :class:LifecycleReceipt reports the erasure complete. A store that erases a large subtree as a durable job (CORE-199) answers :class:ForgetInProgress until it finishes; forget the same scope again to advance it.

Source code in src/symfonic/capabilities/memory/ports.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
async def forget(self, scope: MemoryScope) -> LifecycleReceipt | ForgetInProgress:
    """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.

    Only a :class:`LifecycleReceipt` reports the erasure complete. A store
    that erases a large subtree as a durable job (CORE-199) answers
    :class:`ForgetInProgress` until it finishes; forget the same scope
    again to advance it.
    """
    ...

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, read_admission: ReadAdmission | 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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def validate(self) -> None:
    """Refuse a query whose ceilings cannot admit anything."""
    if self.read_admission is not None:
        if not isinstance(self.read_admission, ReadAdmission):
            raise MemoryContractError("read_admission must be a ReadAdmission")
        self.read_admission.validate_for(self.scope)
    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, _selected_lines: tuple[object, ...] = ())

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 = '', updated_at: datetime | None = None)

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 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
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)

MemoryReflectionService 

MemoryReflectionService(model: ReflectionModelPort, *, prompt_template: str | None = None, scrubber: CredentialScrubber | None = None)

Render a reflection prompt and turn its JSON array into semantic records.

The service contains no store and never publishes by itself. The memory capability owns those effects at the FINALIZE rung, after ordinary extraction and its flush have completed.

Source code in src/symfonic/capabilities/memory/reflection.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def __init__(
    self,
    model: ReflectionModelPort,
    *,
    prompt_template: str | None = None,
    scrubber: CredentialScrubber | None = None,
) -> None:
    complete = getattr(model, "complete", None)
    if not callable(complete) or not inspect.iscoroutinefunction(complete):
        raise ConfigurationError(
            "MemoryReflectionService needs an async complete(prompt) model collaborator."
        )
    template = prompt_template or _DEFAULT_PROMPT
    if "{conversation}" not in template:
        raise ConfigurationError(
            "reflection prompt_template must contain {conversation}; otherwise the "
            "configured collaborator would receive no completed exchange."
        )
    self._model = model
    self._prompt_template = template
    self._scrubber = scrubber if scrubber is not None else CredentialScrubber()

prompt 

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

Return prompt bytes for the collaborator, with credentials redacted.

Source code in src/symfonic/capabilities/memory/reflection.py 
118
119
120
121
122
123
124
125
def prompt(self, request: ReflectionRequest) -> tuple[str, tuple[str, ...]]:
    """Return prompt bytes for the collaborator, with credentials redacted."""
    user = self._scrubber.scrub_text(request.user_message)
    assistant = self._scrubber.scrub_text(request.assistant_message)
    conversation = f"User: {user.text}\nAssistant: {assistant.text}"
    return self._prompt_template.format(
        conversation=conversation
    ), user.redactions + assistant.redactions

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 
87
88
89
90
91
92
93
94
95
96
97
98
99
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 = '', *, _levels: tuple[tuple[str, str], ...] = ())

Where a memory lives, preserving positional or native identity.

Positional levels are 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. Native levels retain explicit kinds and can have arbitrary depth.

identity_version property 

identity_version: int

Zero is positional; one retains every native kind.

levels property 

levels: tuple[tuple[str, str], ...]

Complete root-first identity including kinds.

path property 

path: str

Canonical capability path, preserving native kinds on roundtrip.

segments property 

segments: tuple[str, ...]

The populated levels, outermost first.

ancestor 

ancestor(depth: int) -> MemoryScope

Select a populated ancestor without losing kinds or version.

Source code in src/symfonic/capabilities/memory/scope.py 
136
137
138
139
140
141
142
def ancestor(self, depth: int) -> MemoryScope:
    """Select a populated ancestor without losing kinds or version."""
    if type(depth) is not int or not 1 <= depth <= len(self.segments):
        raise MemoryContractError("scope ancestor depth must select a populated level")
    if self._levels:
        return MemoryScope.from_levels(self._levels[:depth])
    return MemoryScope(*self.segments[:depth])

covers 

covers(other: MemoryScope) -> bool

Whether a memory written at self is visible at other.

Complete-level, 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 
144
145
146
147
148
149
150
151
152
153
154
155
156
def covers(self, other: MemoryScope) -> bool:
    """Whether a memory written at ``self`` is visible at ``other``.

    Complete-level, never string-prefix: ``acme`` does not cover ``acmecorp``,
    and a rule written with :meth:`str.startswith` would say it does.
    """
    mine = self.levels
    theirs = other.levels
    return (
        self.identity_version == other.identity_version
        and 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 
158
159
160
161
162
163
164
165
166
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)

from_levels classmethod 

from_levels(levels: Iterable[tuple[str, str]]) -> MemoryScope

Snapshot explicit kind/ID pairs without inferring missing kinds.

Source code in src/symfonic/capabilities/memory/scope.py 
62
63
64
65
66
67
68
69
70
71
72
@classmethod
def from_levels(cls, levels: Iterable[tuple[str, str]]) -> MemoryScope:
    """Snapshot explicit kind/ID pairs without inferring missing kinds."""
    try:
        values = tuple(levels)
    except TypeError as error:
        raise MemoryContractError("native scope levels must be iterable") from error
    if not values:
        raise MemoryContractError("a native memory scope must name a level")
    _validate_levels(values)
    return cls(values[0][1], _levels=values)

validate 

validate() -> None

Refuse every scope this capability cannot compare.

Source code in src/symfonic/capabilities/memory/scope.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def validate(self) -> None:
    """Refuse every scope this capability cannot compare."""
    if type(self._levels) is not tuple:
        raise MemoryContractError("scope levels must be an immutable tuple")
    if self._levels:
        _validate_levels(self._levels)
        if self.tenant != self._levels[0][1] or self.principal or self.session:
            raise MemoryContractError("native scope flat fields disagree with its levels")
        return
    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."
            )

MemoryStatusService 

MemoryStatusService(graph: Any = None, *, required_labels: Iterable[str] = (), enabled_layers: Iterable[str] = ('semantic',), records: Any = None)

Report discovered labels and required pending blocks for one scope.

The graph backend is the same scoped backend used by native memory writes. enabled_layers is deployment policy, not inferred from stored rows.

Source code in src/symfonic/capabilities/memory/status.py 
29
30
31
32
33
34
35
36
37
38
39
40
def __init__(
    self,
    graph: Any = None,
    *,
    required_labels: Iterable[str] = (),
    enabled_layers: Iterable[str] = ("semantic",),
    records: Any = None,
) -> None:
    self._graph = graph
    self._records = records
    self._required_labels = tuple(required_labels)
    self._enabled_layers = frozenset(enabled_layers)

blocks async 

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

Return the legacy status fields without reaching an orchestrator.

Source code in src/symfonic/capabilities/memory/status.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
async def blocks(self, scope: MemoryScope) -> list[dict[str, Any]]:
    """Return the legacy status fields without reaching an orchestrator."""
    # A kind-aware scope has no positional graph address; a deployment wired
    # with records alone has no graph to query. Both read the inventory.
    if scope.identity_version or (self._graph is None and self._records is not None):
        return self._results(await self._native_groups(scope))
    nodes = (
        await self._graph.query_nodes(
            tenant_scope(scope),
            {"layer": MemoryLayer.SEMANTIC},
            limit=None,
        )
        if self._graph is not None
        else ()
    )
    groups: dict[str, dict[str, Any]] = {}
    for node in nodes:
        # Legacy distinct_labels excludes retracted rows only. Native status
        # also hides unpublished pending writes: a draft cannot satisfy a
        # required block before publish.
        if is_pending(node.properties) or is_retracted(node.properties):
            continue
        entry = groups.setdefault(node.label, {"count": 0, "last_updated": None})
        entry["count"] += 1
        stamp = store_time(node.updated_at)
        if stamp is not None and (
            entry["last_updated"] is None or stamp > entry["last_updated"]
        ):
            entry["last_updated"] = stamp
    return self._results(groups)

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 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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]] = []
    self._write_lock = asyncio.Lock()

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_claims property 

pending_claims: tuple[tuple[str, str, str], ...]

Accepted durable identities still awaiting confirmed publication.

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 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
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()
    for path in self._covered_paths(scope):
        self._pending[path].abandon_requested = True
    mine = tuple(
        sorted(
            record_id
            for path in self._covered_paths(scope)
            for record_id in self._pending[path].ids
        )
    )
    from symfonic.capabilities.memory.native_pending_claims import using

    claims = tuple(
        claim for path in self._covered_paths(scope)
        for claim in self._pending[path].claims
    )
    if not isinstance(self._lifecycle, MemoryDiscardPort):
        self._forget_covered(scope)
        return LifecycleReceipt(
            scope_path=scope.path, discarded=mine, degraded=True
        )
    try:
        with using(claims):
            receipt = await self._lifecycle.discard(scope)
        if not receipt.degraded or not claims or not scope.identity_version:
            self._forget_covered(scope)
        return receipt
    except MemoryUnavailable:
        if not claims or not scope.identity_version:
            self._forget_covered(scope)
        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 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
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()
    if any(self._pending[path].abandon_requested for path in self._covered_paths(scope)):
        raise MemoryContractError("abandon requested; retry discard, not publication")
    from symfonic.capabilities.memory.native_pending_claims import published, using

    checkpoint = snapshot_pending(self._pending, self._covered_paths(scope))
    claims = tuple(claim for snap in checkpoint.values() for claim in snap.claims)
    try:
        with using(claims):
            receipt = await self._lifecycle.flush(scope)
    except MemoryUnavailable:
        return LifecycleReceipt(scope_path=scope.path, degraded=True)
    if not receipt.degraded:
        confirmed = set(receipt.committed) | await published(self._lifecycle, claims)
        if not set(required_ids).issubset(confirmed):
            raise MemoryContractError("required buffered records were not published")
        retire_snapshot(self._pending, checkpoint, confirmed)
        receipt = LifecycleReceipt(
            scope_path=receipt.scope_path, committed=tuple(sorted(confirmed)),
        )
    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 
188
189
190
191
192
193
194
195
196
197
198
199
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 settle(request, task))
    return tuple(outcomes)

write async 

write(request: WriteRequest) -> WriteReceipt

Serialize claim receipt bookkeeping with the durable write.

Source code in src/symfonic/capabilities/memory/writes.py 
130
131
132
133
async def write(self, request: WriteRequest) -> WriteReceipt:
    """Serialize claim receipt bookkeeping with the durable write."""
    async with self._write_lock:
        return await self._write_locked(request)

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 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
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 
138
139
140
141
142
143
144
145
146
147
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 = '', pending_connections: tuple[dict[str, Any], ...] | None = None, 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 
68
69
70
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 
64
65
66
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 
72
73
74
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)

ProceduralPublishService 

ProceduralPublishService(graph: Any, scrubber: Any = None)

Review-gated publication of org skills to the platform corpus.

Parameters:

Name Type Description Default
graph Any

graph backend (reads/writes procedural nodes).

 required
scrubber Any

redaction engine with scrub_row(row, drop_on_doubt=...) returning {scrubbed, match_counts, unresolved, dropped_reason}. Injected (not imported) so deployments own their pattern set.

 None
Source code in src/symfonic/capabilities/memory/procedure_publish.py 
66
67
68
69
70
71
72
73
74
75
76
def __init__(self, graph: Any, scrubber: Any = None) -> None:
    from symfonic.memory.graph.store import GraphMemoryStore
    from symfonic.memory.privacy import CorpusScrubber

    # Raw backend in, store interface inside -- the same wrap
    # ProcedureAdminService applies, so callers pass backends[0].
    # Default scrubber: publication without redaction is not an
    # option the API offers. A deployment that tuned its own
    # patterns passes its instance instead.
    self._graph = GraphMemoryStore(graph)
    self._scrubber = scrubber if scrubber is not None else CorpusScrubber()

list_proposals async 

list_proposals(scope: Any) -> list[dict[str, Any]]

Pending proposals in one tenant (the review queue).

Source code in src/symfonic/capabilities/memory/procedure_publish.py 
227
228
229
230
231
232
233
234
235
236
237
async def list_proposals(self, scope: Any) -> list[dict[str, Any]]:
    """Pending proposals in one tenant (the review queue)."""
    scoped = tenant_scope(scope)
    nodes = await self._graph.query_nodes(
        scoped, layer=MemoryLayer.PROCEDURAL, limit=None
    )
    return [
        self._row(node)
        for node in nodes
        if (node.properties or {}).get("status") == PROPOSAL_STATUS
    ]

list_published async 

list_published() -> list[dict[str, Any]]

Active platform skills (what cross-org recall can see).

Source code in src/symfonic/capabilities/memory/procedure_publish.py 
239
240
241
242
243
244
245
246
247
248
249
250
async def list_published(self) -> list[dict[str, Any]]:
    """Active platform skills (what cross-org recall can see)."""
    nodes = await self._graph.query_nodes(
        platform_scope(), layer=MemoryLayer.PROCEDURAL, limit=None
    )
    out = []
    for node in nodes:
        props = node.properties or {}
        status, active = procedure_lifecycle(props)
        if status == "approved" and active and props.get("publication"):
            out.append(self._row(node))
    return out

propose_publication async 

propose_publication(scope: Any, node_id: str, *, proposer: str) -> dict[str, Any]

Scrub an approved skill and record a platform proposal for it.

Source code in src/symfonic/capabilities/memory/procedure_publish.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
async def propose_publication(
    self, scope: Any, node_id: str, *, proposer: str
) -> dict[str, Any]:
    """Scrub an approved skill and record a platform proposal for it."""

    scoped = tenant_scope(scope)
    node = await self._graph.get_node(scoped, NodeId(node_id))
    self._require_skill(node, node_id)
    status, active = procedure_lifecycle(node.properties or {})
    if status != "approved" or not active:
        raise PublicationRefused(
            f"only approved active skills can be proposed (node {node_id} "
            f"is {status})"
        )
    scrubbed = self._scrubber.scrub_row(
        {
            "label": node.label,
            "content": (node.properties or {}).get("content", ""),
            "steps": (node.properties or {}).get("steps", []),
            "context": (node.properties or {}).get("context", ""),
        },
        drop_on_doubt=True,
    )
    if scrubbed.scrubbed is None:
        raise PublicationRefused(
            f"scrubber refused the skill: {scrubbed.dropped_reason} "
            f"(matches={dict(scrubbed.match_counts)})"
        )
    from symfonic.memory.layers.procedural.sharing import proposal_node

    proposal = proposal_node(
        scoped, node, scrubbed.scrubbed,
        {
            "match_counts": dict(scrubbed.match_counts),
            "unresolved": scrubbed.unresolved,
        },
        proposer,
    )
    async with review_lock_for(self._graph):
        created = await self._graph.add_node(scoped, proposal)
    return self._row(created)

review_publication async 

review_publication(scope: Any, proposal_id: str, *, decision: str, reviewer: str, reason: str = '') -> dict[str, Any]

Approve (publish the scrubbed copy) or refuse a proposal.

Only the source tenant's owners may decide; the transport layer enforces that from the verified principal before calling here.

Source code in src/symfonic/capabilities/memory/procedure_publish.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
async def review_publication(
    self,
    scope: Any,
    proposal_id: str,
    *,
    decision: str,
    reviewer: str,
    reason: str = "",
) -> dict[str, Any]:
    """Approve (publish the scrubbed copy) or refuse a proposal.

    Only the source tenant's owners may decide; the transport layer
    enforces that from the verified principal before calling here.
    """

    if decision not in ("approve", "refuse"):
        raise PublicationRefused(f"unknown decision {decision!r}")
    scoped = tenant_scope(scope)
    proposal = await self._graph.get_node(scoped, NodeId(proposal_id))
    if proposal is None:
        raise PublicationNotFound(f"proposal {proposal_id} not found")
    current = (proposal.properties or {}).get("status")
    if current != PROPOSAL_STATUS:
        raise PublicationConflict(
            f"proposal {proposal_id} already decided ({current})"
        )
    publication = dict((proposal.properties or {}).get("publication", {}))
    if decision == "refuse":
        return await self._transition(
            scoped, proposal, REFUSED_STATUS,
            {"reviewed_by": reviewer, "reviewed_at": _now(), "reason": reason},
        )
    snapshot = publication.get("snapshot") or {}
    platform_scope_ = platform_scope()
    # No silent duplicates: an identical label already published resolves
    # to a refusal naming the existing copy (re-approving the same skill
    # -- retry, double review -- must not mint a second platform row).
    # Normalized the same way the read path dedupes tenant-vs-platform.
    wanted = str(snapshot.get("label", "") or "").strip().casefold()
    if wanted:
        existing = await self._graph.query_nodes(
            platform_scope_, layer=MemoryLayer.PROCEDURAL, limit=None
        )
        for node in existing:
            props = node.properties or {}
            status, active = procedure_lifecycle(props)
            if status != "approved" or not active:
                continue
            if str(node.label or "").strip().casefold() == wanted:
                raise PublicationConflict(
                    f"already published as platform node {node.id}"
                )
    from symfonic.memory.layers.procedural.sharing import platform_copy_node

    copy = platform_copy_node(
        snapshot,
        {
            **publication,
            "reviewed_by": reviewer,
            "reviewed_at": _now(),
            "reason": reason,
        },
        proposal.importance,
    )
    async with review_lock_for(self._graph):
        created = await self._graph.add_node(platform_scope_, copy)
        row = await self._transition(
            scoped, proposal, APPROVED_STATUS,
            {
                "reviewed_by": reviewer, "reviewed_at": _now(), "reason": reason,
                "platform_node_id": str(created.id),
            },
            lock_held=True,
        )
    row["platform_node_id"] = str(created.id)
    return row

revoke_publication async 

revoke_publication(caller_scope: Any, platform_node_id: str, *, actor: str) -> dict[str, Any]

Deactivate a platform copy. Only the publishing org (or a platform admin calling with its tenant header) may revoke, checked against the copy's own provenance -- never against a caller claim.

Source code in src/symfonic/capabilities/memory/procedure_publish.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
async def revoke_publication(
    self, caller_scope: Any, platform_node_id: str, *, actor: str
) -> dict[str, Any]:
    """Deactivate a platform copy. Only the publishing org (or a platform
    admin calling with its tenant header) may revoke, checked against
    the copy's own provenance -- never against a caller claim."""
    scoped = tenant_scope(caller_scope)
    node = await self._graph.get_node(platform_scope(), NodeId(platform_node_id))
    if node is None:
        raise PublicationNotFound(f"platform node {platform_node_id} not found")
    provenance = (node.properties or {}).get("publication") or {}
    if not provenance.get("source_tenant"):
        raise PublicationNotFound(
            f"node {platform_node_id} is not a platform publication"
        )
    if provenance["source_tenant"] != scoped.tenant_id:
        raise PublicationForbidden(
            "only the publishing org may revoke this publication"
        )
    async with review_lock_for(self._graph):
        updated = await self._graph.update_node(
            platform_scope(), node.id,
            {"active": False, "revoked": {"by": actor, "at": _now()}},
        )
    return self._row(updated)

ProcedureAdminService 

ProcedureAdminService(graph: Any)

List one scope's stored procedures without composing an agent.

Source code in src/symfonic/capabilities/memory/procedure_admin.py 
28
29
def __init__(self, graph: Any) -> None:
    self._graph = GraphMemoryStore(graph)

amend_procedure async 

amend_procedure(scope: MemoryScope, node_id: str, *, steps: Any = None, properties: dict[str, Any] | None = None, active: bool | None = None) -> dict[str, Any] | None

Edit a visible procedure through the locked lifecycle transition.

Source code in src/symfonic/capabilities/memory/procedure_admin.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
async def amend_procedure(
    self, scope: MemoryScope, node_id: str, *, steps: Any = None,
    properties: dict[str, Any] | None = None, active: bool | None = None,
) -> dict[str, Any] | None:
    """Edit a visible procedure through the locked lifecycle transition."""
    scoped = tenant_scope(scope)
    # The layer method guards review state; this port also rejects nodes
    # outside the procedure surface before invoking that transition.
    candidate = await self._graph.backend.get_node(scoped, NodeId(node_id))
    if (candidate is None or candidate.layer != MemoryLayer.PROCEDURAL
            or is_retracted(candidate.properties)):
        return None
    from symfonic.memory.layers.procedural.layer import ProceduralLayer

    try:
        node = await ProceduralLayer(self._graph).amend_skill(
            scoped, NodeId(node_id), steps=steps, properties=properties, active=active,
        )
    except KeyError:
        return None
    props = node.properties or {}
    status, node_active = procedure_lifecycle(props)
    return {
        "id": str(node.id), "label": node.label, "steps": props.get("steps", []),
        "properties": props, "status": status, "active": node_active,
    }

delete_procedure async 

delete_procedure(scope: MemoryScope, node_id: str) -> bool

Atomically delete one exact-owner procedural row, if supported.

Source code in src/symfonic/capabilities/memory/procedure_admin.py 
37
38
39
40
41
42
43
async def delete_procedure(self, scope: MemoryScope, node_id: str) -> bool:
    """Atomically delete one exact-owner procedural row, if supported."""
    backend = self._graph.backend
    delete = getattr(backend, "delete_procedure_atomic", None)
    if not getattr(backend, "supports_atomic_procedure_delete", False) or not callable(delete):
        raise NotImplementedError("atomic procedure deletion unavailable")
    return await delete(tenant_scope(scope), NodeId(node_id))

get_procedure async 

get_procedure(scope: MemoryScope, node_id: str) -> dict[str, Any] | None

Read one procedural node in the authenticated tenant's scope.

Source code in src/symfonic/capabilities/memory/procedure_admin.py 
132
133
134
135
136
137
138
139
async def get_procedure(
    self, scope: MemoryScope, node_id: str
) -> dict[str, Any] | None:
    """Read one procedural node in the authenticated tenant's scope."""
    node = await self._graph.get_node(tenant_scope(scope), NodeId(node_id))
    if node is None or node.layer != MemoryLayer.PROCEDURAL:
        return None
    return self._row(node)

list_draft_procedures async 

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

Project draft rows to the legacy review-queue wire shape.

Source code in src/symfonic/capabilities/memory/procedure_admin.py 
161
162
163
164
165
166
167
async def list_draft_procedures(self, scope: MemoryScope) -> list[dict[str, Any]]:
    """Project draft rows to the legacy review-queue wire shape."""
    rows = await self.list_procedures(scope, status="draft")
    fields = (
        "id", "label", "content", "steps", "status", "active", "importance", "created_at",
    )
    return [{field: row[field] for field in fields} for row in rows]

list_procedures async 

list_procedures(scope: MemoryScope, *, status: str | None = None) -> list[dict[str, Any]]

Preserve the legacy list row and status filter over all matching nodes.

Source code in src/symfonic/capabilities/memory/procedure_admin.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
async def list_procedures(
    self, scope: MemoryScope, *, status: str | None = None
) -> list[dict[str, Any]]:
    """Preserve the legacy list row and status filter over all matching nodes."""
    nodes = await self._graph.query_nodes(
        tenant_scope(scope),
        layer=MemoryLayer.PROCEDURAL,
        limit=None,
    )
    rows = []
    for node in nodes:
        props = node.properties or {}
        node_status, active = procedure_lifecycle(props)
        if (status is None and node_status != "approved") or (
            status not in (None, "all") and status != node_status
        ):
            continue
        rows.append(self._row(node))
    return rows

toggle_procedure async 

toggle_procedure(scope: MemoryScope, node_id: str) -> dict[str, Any] | None

Toggle a visible procedure under the same lock as review writes.

Source code in src/symfonic/capabilities/memory/procedure_admin.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
async def toggle_procedure(
    self, scope: MemoryScope, node_id: str,
) -> dict[str, Any] | None:
    """Toggle a visible procedure under the same lock as review writes."""
    scoped = tenant_scope(scope)
    async with review_lock_for(self._graph.backend):
        candidate = await self._graph.backend.get_node(scoped, NodeId(node_id))
        if (candidate is None or candidate.layer != MemoryLayer.PROCEDURAL
                or is_retracted(candidate.properties)):
            return None
        node = await self._graph.backend.get_node(scoped, NodeId(node_id))
        if (node is None or node.layer != MemoryLayer.PROCEDURAL
                or is_retracted(node.properties)):
            return None
        props = dict(node.properties or {})
        nested = props.get("record_metadata")
        if isinstance(nested, dict):
            reserved = {
                "status", "active", "procedure_review_authority",
                "procedure_review_actor", "procedure_lifecycle_origin",
            }
            props["record_metadata"] = {
                key: value for key, value in nested.items() if key not in reserved
            }
        status, active = procedure_lifecycle(props)
        props["active"] = not active if status == "approved" else False
        props["_last_edited_by"] = "user_manual_edit"
        try:
            node = await self._graph.update_node(
                scoped, NodeId(node_id), {
                    "properties": props,
                    "access_count": node.access_count + 1,
                },
            )
        except KeyError:
            return None
        props = dict(node.properties or {})
        status, active = procedure_lifecycle(props)
        return {
            "id": str(node.id), "label": node.label, "status": status,
            "active": active, "properties": props,
        }

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.

ProspectiveCreateService 

ProspectiveCreateService(graph: Any, *, embedder: Any = None)

Write a trigger through the existing prospective domain operation.

Source code in src/symfonic/capabilities/memory/prospective_create.py 
21
22
23
24
25
26
def __init__(self, graph: Any, *, embedder: Any = None) -> None:
    if graph is None:
        raise ValueError("prospective create needs a graph backend")
    self._layer = ProspectiveLayer(GraphMemoryStore(graph), embedder)
    # The live HTTP route scrubs credential-shaped keys, not values.
    self._scrubber = CredentialScrubber(value_patterns=())

ProviderFamily

Bases: StrEnum

The reply shapes this capability knows how to read.

ReadAdmission dataclass 

ReadAdmission(allowed_scope_paths: tuple[str, ...])

Canonical exact owner paths for an explicitly narrowed read.

distance 

distance(record: MemoryScope, query: MemoryScope) -> int

Return a score distance, or -1 for an unadmitted owner.

Source code in src/symfonic/capabilities/memory/read_admission.py 
49
50
51
52
53
54
55
56
57
def distance(self, record: MemoryScope, query: MemoryScope) -> int:
    """Return a score distance, or -1 for an unadmitted owner."""
    if (
        record.path not in self.allowed_scope_paths
        or not _within_query_authority(record, query)
    ):
        return -1
    inherited = record.distance(query)
    return inherited if inherited >= 0 else 1

validate_for 

validate_for(scope: MemoryScope) -> None

An explicit sibling may not widen the query to another principal.

Source code in src/symfonic/capabilities/memory/read_admission.py 
41
42
43
44
45
46
47
def validate_for(self, scope: MemoryScope) -> None:
    """An explicit sibling may not widen the query to another principal."""
    if any(not _within_query_authority(scope_from_path(path), scope)
           for path in self.allowed_scope_paths):
        raise MemoryContractError(
            "read admission must name a visible ancestor or same-principal session"
        )

RetrievalResult dataclass 

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

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 
205
206
207
def render(self) -> str:
    """One memory per line, in the order selection decided."""
    return "\n".join(memory.line() for memory in (*self.memories, *self.linked))

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 
209
210
211
212
213
214
215
216
217
218
219
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, *self.linked)
    )
    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, score_is_final: bool = False, selected_publication: PublishedRecallRef | None = None, linked_edge: LinkedEdgeWitness | None = None)

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 
171
172
173
174
175
176
177
178
179
180
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 
82
83
84
85
86
87
88
89
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 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@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 
91
92
93
94
95
96
97
98
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 
78
79
80
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.

SemanticCreateService 

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

Create a fact on the graph at the caller's resolved scope.

The graph backend stamps the scope path during insertion. The service never obtains an agent or derives a scope from request body data.

Source code in src/symfonic/capabilities/memory/semantic_create.py 
29
30
31
32
33
34
35
36
37
38
def __init__(
    self, graph: Any, *, vectors: Any = None, embedder: Any = None,
) -> None:
    if (vectors is None) != (embedder is None):
        raise ValueError("semantic create needs a vector backend and embedder together")
    self._graph = GraphMemoryStore(graph)
    self._vectors = VectorRecall(vectors, embedder) if vectors is not None else None
    # Legacy HTTP creation removes credential-shaped keys, including in
    # metadata. Keep that wire contract while using the capability scrubber.
    self._scrubber = CredentialScrubber(value_patterns=())

SpreadingActivation dataclass 

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

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

expand async 

expand(scope: MemoryScope, seeds: Sequence[RetrievedMemory], admission: ReadAdmission | None = None) -> 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 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
async def expand(
    self, scope: MemoryScope, seeds: Sequence[RetrievedMemory],
    admission: ReadAdmission | None = None,
) -> tuple[tuple[RetrievedMemory, ...], ActivationLog]:
    """Walk out from ``seeds`` and return what lit up, plus the provenance."""
    if self.linked_policy is not None:
        from symfonic.capabilities.memory.linked_spreading import expand_linked
        return await expand_linked(self, scope, seeds, admission)
    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, admission)
        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 
 99
100
101
def render(self) -> str:
    """One turn per line, oldest first."""
    return "\n".join(turn.line() for turn in self.turns)

WorkingMemorySummaryService 

WorkingMemorySummaryService(source: EvictedWorkingSource, *, mode: STMSummaryMode = 'extractive', max_chars: int = 1000, model_provider: Any | None = None, model_config: Any | None = None)

Summarize an explicit evicted-turn source for one requested scope.

Source code in src/symfonic/capabilities/memory/summary.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def __init__(
    self,
    source: EvictedWorkingSource,
    *,
    mode: STMSummaryMode = "extractive",
    max_chars: int = 1000,
    model_provider: Any | None = None,
    model_config: Any | None = None,
) -> None:
    evicted = getattr(source, "evicted", None)
    if not callable(evicted) or not inspect.iscoroutinefunction(evicted):
        raise ConfigurationError(
            "STM summary needs an async evicted(scope) source; a recent-turn source "
            "cannot identify what the working window already discarded."
        )
    if mode not in {"off", "extractive", "llm"}:
        raise ConfigurationError(f"unknown STM summary mode {mode!r}")
    if max_chars < 1:
        raise ConfigurationError("STM summary max_chars must be positive")
    if mode == "llm" and model_provider is None:
        raise ConfigurationError(
            "STM summary mode 'llm' needs an explicit model_provider; native composition "
            "does not borrow the response model."
        )
    self._source = source
    self._mode = mode
    self._max_chars = max_chars
    self._model_provider = model_provider
    self._model_config = model_config

WorkingWindow dataclass 

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

Reads the last few turns of a conversation, ungated.

read async 

read(scope: MemoryScope, *, session_id: str = '') -> WorkingContext

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

Source code in src/symfonic/capabilities/memory/working.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
async def read(self, scope: MemoryScope, *, session_id: str = "") -> WorkingContext:
    """Read the window at ``scope``, dropping only what would corrupt it."""
    if self.recent_turns == 0:
        return WorkingContext()
    try:
        if isinstance(self.source, SessionConversationSource):
            if not session_id:
                raise MemoryContractError(
                    "a session conversation window requires session_id on this request"
                )
            recent = await self.source.recent_for_session(
                scope, session_id, self.recent_turns
            )
        else:
            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, selected_manifest: SelectedRecordManifest | None = None)

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 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
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 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
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

Refuse paths deeper than this positional memory scope can preserve.

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

Source code in src/symfonic/capabilities/memory/binding.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def as_memory_scope(scope: Any) -> MemoryScope:
    """Refuse paths deeper than this positional memory scope can preserve.

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

as_native_memory_scope 

as_native_memory_scope(scope: Any) -> MemoryScope

Opt in to complete, kind-aware identity for memory composition.

Source code in src/symfonic/capabilities/memory/binding.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def as_native_memory_scope(scope: Any) -> MemoryScope:
    """Opt in to complete, kind-aware identity for memory composition."""
    if isinstance(scope, MemoryScope):
        if scope.identity_version != 1:
            raise MemoryContractError("native scope requires complete kind/id levels")
        return scope
    path = getattr(scope, "path", None)
    if type(path) is not tuple or not path:
        raise MemoryContractError("native scope requires a nonempty canonical path")
    pairs: list[tuple[str, str]] = []
    for level in path:
        kind = getattr(level, "kind", None)
        id_ = getattr(level, "id", None)
        if type(kind) is not str or type(id_) is not str:
            raise MemoryContractError("native scope path needs complete kind/id levels")
        pairs.append((kind, id_))
    result = MemoryScope.from_levels(tuple(pairs))
    segments = getattr(scope, "segments", None)
    if segments is not None:
        if type(segments) not in (tuple, list):
            raise MemoryContractError("native scope segments must be a complete sequence")
        if tuple(segments) != result.segments:
            raise MemoryContractError("native scope path and segments disagree")
    root_id = getattr(scope, "tenant_id", None)
    if root_id is not None and root_id != result.segments[0]:
        raise MemoryContractError("native scope root and tenant_id disagree")
    child_id = getattr(scope, "sub_tenant_id", None)
    if child_id is not None and (len(result.segments) < 2 or child_id != result.segments[1]):
        raise MemoryContractError("native scope child and sub_tenant_id disagree")
    converted = getattr(scope, "to_memory_scope", None)
    if callable(converted):
        custom = converted()
        if isinstance(custom, MemoryScope):
            if custom.identity_version != 1 or custom.levels != result.levels:
                raise MemoryContractError("to_memory_scope lost native scope kind/id levels")
        elif custom is not scope:
            raise MemoryContractError("to_memory_scope returned a malformed native scope")
    return result

as_tenant_scope 

as_tenant_scope(scope: MemoryScope) -> TenantScope

Shared direct services use positional storage until individually migrated.

Source code in src/symfonic/capabilities/memory/graph_rows.py 
112
113
114
115
116
117
118
119
120
def tenant_scope(scope: MemoryScope) -> TenantScope:
    """Shared direct services use positional storage until individually migrated."""
    if scope.identity_version:
        raise MemoryContractError("native scope requires an admitted graph adapter")
    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 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
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 
228
229
230
231
232
233
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 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
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,
            "jit_only": contribution.jit_only,
        }
    )

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 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
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:
        if policy.enable_entity_linker and entity_extractor is None:
            raise MemoryContractError(
                "LearningPolicy.enable_entity_linker=True needs an "
                "entity_extractor collaborator. Build one explicitly (for "
                "example with symfonic.core.learning.build_entity_extractor) "
                "and pass it to deep_phases."
            )
        if not policy.enable_entity_linker and entity_extractor is not None:
            raise MemoryContractError(
                "LearningPolicy.enable_entity_linker=False conflicts with an "
                "entity_extractor collaborator. Enable the policy before "
                "composing the entity-link phase."
            )
        # 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 
57
58
59
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 
118
119
120
121
122
123
124
125
126
127
128
129
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 
86
87
88
89
90
91
92
93
94
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 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
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 
71
72
73
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 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
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] = storage_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 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
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] = storage_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
    if (own := producer_metadata(record.metadata)):
        bag[METADATA_KEY] = own
    # Public procedural writes admit only unreviewed drafts.
    if getattr(record.layer, "value", record.layer) == MemoryLayer.PROCEDURAL.value:
        from symfonic.memory.layers.procedural.lifecycle import normalise_procedure_lifecycle
        bag = normalise_procedure_lifecycle(bag)
    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

Retain the shipped positional row format, for v0 only.

Source code in src/symfonic/capabilities/memory/scope_codec.py 
17
18
19
20
21
22
23
def legacy_scope_path(scope: MemoryScope) -> str:
    """Retain the shipped positional row format, for v0 only."""
    if scope.identity_version:
        raise MemoryContractError("native scope requires versioned storage identity")
    return LEGACY_SCOPE_DELIMITER.join(
        part for pair in zip(LEGACY_SCOPE_KINDS, scope.segments, strict=False) for part in pair
    )

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 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
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, summary: STMSummaryPort | None = None, reflection: ReflectionPort | None = None, consolidation: Any | None = None, conversation: Any | None = None, recent_turns: int = 0, activation: Any | None = None, retrieval_policy: RetrievalPolicy | None = None, retrieval: RetrievalCoordinator | None = None, exclude_speakers: frozenset[str] = frozenset(), episodic_telemetry_sink: EpisodicTelemetrySink | None = None, operation_journal: MemoryOperationJournalPort | None = None, triage: IntentTriageCapability | 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

typed paths retain every native kind/ID level. Explicit MemoryScope values retain their version, including positional v0.

 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
conversation Any | None

an optional conversation source. NativeConversationHistory also installs final-only publication when supplied.

 None
recent_turns int

number of recent utterances; zero skips the window read.

 0
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
summary STMSummaryPort | None

an explicit evicted-turn summary collaborator, normally :class:~.summary.WorkingMemorySummaryService. It supplies a JIT-only prompt contribution. None disables it without reading a working source.

 None
reflection ReflectionPort | None

a completed-exchange insight collaborator. None disables it without calling its collaborator.

 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
retrieval_policy RetrievalPolicy | None

the public relevance, salience, layer, and hierarchy policy for the default coordinator. A policy is applied before the block is compiled; changing a floor or blend changes the recalled records, not merely a diagnostic.

 None
retrieval RetrievalCoordinator | None

a fully composed :class:~.RetrievalCoordinator. Use this with an adopter CandidateSource when the store's pre-ranked retrieval is not the desired scorer. It owns policy and activation, so passing either alongside it is refused.

 None
exclude_speakers frozenset[str]

transcript and explicitly tagged recalled speakers omitted. Untagged recalled records remain eligible; speaker is never inferred.

 frozenset()
episodic_telemetry_sink EpisodicTelemetrySink | None

optional async observer for successful memory.lifecycle publications. It receives scope and run correlation identifiers plus committed record ids, never record text, prompts, responses, or metadata. Sink failures are logged and do not roll back a publication; None is disabled.

 None
triage IntentTriageCapability | None

the composed IntentTriageCapability (from symfonic.capabilities.triage) retrieval routes on. Pass the same instance composed into the agent: pure-action turns skip the store round trip. None -- the default -- hydrates every turn.

 None

Raises:

Type Description
ConfigurationError

if extractor, summary, or reflection cannot serve its 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. 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.

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.

Source code in src/symfonic/capabilities/memory/bundle.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def memory_capabilities(
    store: Any,
    scope: MemoryScope | Any,
    *,
    limit: int = 5,
    recall_budget: RecallBudget | None = None,
    extractor: MemoryExtractorPort | None = None,
    summary: STMSummaryPort | None = None,
    reflection: ReflectionPort | None = None,
    consolidation: Any | None = None,
    conversation: Any | None = None,
    recent_turns: int = 0,
    activation: Any | None = None,
    retrieval_policy: RetrievalPolicy | None = None,
    retrieval: RetrievalCoordinator | None = None,
    exclude_speakers: frozenset[str] = frozenset(),
    episodic_telemetry_sink: EpisodicTelemetrySink | None = None,
    operation_journal: MemoryOperationJournalPort | None = None,
    triage: IntentTriageCapability | 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: typed paths retain every native kind/ID level. Explicit
            MemoryScope values retain their version, including positional v0.
        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.
        conversation: an optional conversation source. NativeConversationHistory
            also installs final-only publication when supplied.
        recent_turns: number of recent utterances; zero skips the window read.
        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.
        summary: an explicit evicted-turn summary collaborator, normally
            :class:`~.summary.WorkingMemorySummaryService`. It supplies a
            JIT-only prompt contribution. ``None`` disables it without reading
            a working source.
        reflection: a completed-exchange insight collaborator. ``None``
            disables it without calling its collaborator.
        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.
        retrieval_policy: the public relevance, salience, layer, and hierarchy
            policy for the default coordinator. A policy is applied before the
            block is compiled; changing a floor or blend changes the recalled
            records, not merely a diagnostic.
        retrieval: a fully composed :class:`~.RetrievalCoordinator`. Use this
            with an adopter ``CandidateSource`` when the store's pre-ranked
            retrieval is not the desired scorer. It owns policy and activation,
            so passing either alongside it is refused.
        exclude_speakers: transcript and explicitly tagged recalled speakers omitted.
            Untagged recalled records remain eligible; speaker is never inferred.
        episodic_telemetry_sink: optional async observer for successful
            ``memory.lifecycle`` publications.  It receives scope and run
            correlation identifiers plus committed record ids, never record
            text, prompts, responses, or metadata.  Sink failures are logged
            and do not roll back a publication; ``None`` is disabled.
        triage: the composed ``IntentTriageCapability`` (from
            ``symfonic.capabilities.triage``) retrieval routes on. Pass the
            same instance composed into the agent: pure-action turns skip the
            store round trip. ``None`` -- the default -- hydrates every turn.

    Raises:
        ConfigurationError: if ``extractor``, ``summary``, or ``reflection``
            cannot serve its 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.** 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.

    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."""
    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"]
    from symfonic.capabilities.memory.turn_generation import bind_turn_generations, history_stage
    conversation, generations = bind_turn_generations(store, scope, conversation)
    if consolidation is not None:
        effects.append("memory-consolidate")
    capabilities = [
        GrantEffects(*effects),
        MemoryBundleFactory(
            store, limit=limit, recall_budget=recall_budget,
            extractor=extractor, summary=summary, reflection=reflection,
            consolidation=consolidation,
            conversation=conversation, recent_turns=recent_turns,
            activation=activation, retrieval_policy=retrieval_policy,
            retrieval=retrieval, exclude_speakers=exclude_speakers,
            episodic_telemetry_sink=episodic_telemetry_sink,
            operation_journal=operation_journal, generations=generations,
            triage=triage,
        ).for_scope(scope),
    ]
    return [*capabilities, *history_stage(conversation, generations, composition_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 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
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 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
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
    if layer is MemoryLayer.PROCEDURAL:
        # A turn can propose a procedure through any node-producing action,
        # not solely ``store_skill``.  Lifecycle is owned by the procedural
        # review domain, so producer supplied terminal values are replaced at
        # the capability boundary before they become nested record metadata.
        properties["status"] = "draft"
        properties["active"] = False
    # 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
         and (layer is not MemoryLayer.PROCEDURAL or key not in {"status", "active"})},
    )
    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 
64
65
66
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 = '', min_pattern_count: int | None = None, recency_days: int | None = None, max_drafts_per_run: int | None = None, use_tool_calls_fallback: bool | None = None, promote_assistant_content: bool | None = None, use_llm_extractor: bool | None = None, llm_max_episodes_per_run: int | None = None, llm_max_drafts_per_run: int | None = None, synthetic_min_co_count: int | None = None, 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
use_llm_extractor bool | None

explicit Phase-12 branch selection from :class:LearningPolicy. None preserves the direct-factory convention that a supplied extractor selects the LLM branch.

 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 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
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 = "",
    min_pattern_count: int | None = None,
    recency_days: int | None = None,
    max_drafts_per_run: int | None = None,
    use_tool_calls_fallback: bool | None = None,
    promote_assistant_content: bool | None = None,
    use_llm_extractor: bool | None = None,
    llm_max_episodes_per_run: int | None = None,
    llm_max_drafts_per_run: int | None = None,
    synthetic_min_co_count: int | None = None,
    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.
        use_llm_extractor: explicit Phase-12 branch selection from
            :class:`LearningPolicy`. ``None`` preserves the direct-factory
            convention that a supplied extractor selects the LLM branch.
        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."
        )
    if use_llm_extractor is True and procedural_extractor is None:
        raise MemoryContractError(
            "nightly_phases selected the LLM procedural extractor but no "
            "procedural_extractor was composed"
        )

    promotion = {
        key: value
        for key, value in (
            ("min_pattern_count", min_pattern_count),
            ("recency_days", recency_days),
            ("max_drafts_per_run", max_drafts_per_run),
            ("use_tool_calls_fallback", use_tool_calls_fallback),
            ("promote_assistant_content", promote_assistant_content),
            ("llm_max_episodes_per_run", llm_max_episodes_per_run),
            ("llm_max_drafts_per_run", llm_max_drafts_per_run),
        )
        if value is not None
    }
    selected_extractor = (
        procedural_extractor if use_llm_extractor is not False else None
    )

    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, min_co_count=synthetic_min_co_count
        ),
        "procedural_promotion": ProceduralPromotionPhase(
            procedural,
            evidence,
            llm_extractor=selected_extractor,
            llm_model_name=procedural_model_name,
            **promotion,
        ),
    }
    # 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 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
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.native_create import NativeProceduralLayer

    return NativeProceduralLayer(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 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
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, enable_semantic_merge: bool = False, chat_model: Any | None = None, embedding_provider: Any | None = None, lookback_hours: float = 24.0, exact_session_owner: bool = False, session_kind: str | None = None, spreading_weight: float | None = None, policy: Any | None = None, 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 required. Native exact owner mode requires the app's session_kind and an exact in-memory journal composition.

Source code in src/symfonic/capabilities/memory/phases/quick.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
def quick_phases(
    *,
    graph: Any,
    episodic: Any | None = None,
    profile_fields: frozenset[str] | None = None,
    enable_semantic_merge: bool = False,
    chat_model: Any | None = None,
    embedding_provider: Any | None = None,
    lookback_hours: float = 24.0,
    exact_session_owner: bool = False,
    session_kind: str | None = None,
    spreading_weight: float | None = None,
    policy: Any | None = None,
    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 required. Native exact owner mode requires the app's
    ``session_kind`` and an exact in-memory journal composition.
    """
    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."
        )
    if exact_session_owner and (profile_fields or episodic is not None or enable_semantic_merge):
        raise MemoryContractError("exact_session_owner forbids profile_fields, episodic or merge")
    if enable_semantic_merge and chat_model is None:
        raise MemoryContractError(
            "enable_semantic_merge=True needs a chat_model judge collaborator. "
            "Pass the consolidation-role model that should decide whether two "
            "semantic memories express the same fact."
        )
    if policy is not None:
        from symfonic.capabilities.memory.policy import LearningPolicy

        if not isinstance(policy, LearningPolicy):
            raise MemoryContractError("policy must be a LearningPolicy")
        if spreading_weight is None:
            spreading_weight = policy.phase1_spreading_weight
        if episodic_max_entries == 100:
            episodic_max_entries = policy.episodic_summarization_max_entries
        if episodic_summarize_batch == 50:
            episodic_summarize_batch = policy.episodic_summarization_batch_size
    if spreading_weight is None:
        spreading_weight = 0.5
    graph = phase_graph(graph)
    window = RecentSemanticNodes(graph, lookback_hours=lookback_hours,
                                 exact_session_owner=exact_session_owner,
                                 session_kind=session_kind)
    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,
            enabled=enable_semantic_merge,
            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, native_exact_owner=exact_session_owner)

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 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
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 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
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 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
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_storage_path(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 {}
    from symfonic.capabilities.memory.operations import classification_identity, session_stamp
    if "session_id" in bag or "session_id" in metadata:
        metadata["session_id"] = session_stamp(bag)

    identity, error = classification_identity(bag, layer=layer)
    if identity is not None and not error:
        metadata["memory_category"], metadata["subject"] = identity
    elif error:
        # Preserve invalid admission; dropping a conflicting flat spelling
        # would silently turn the nested spelling into a durable identity.
        metadata["memory_category"] = metadata["subject"] = None
    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
    )
    text = retrievable_text(text, metadata) if origin.startswith("extraction:") else text
    if isinstance(bag.get("speaker"), str) and "speaker" not in metadata:
        metadata["speaker"] = bag["speaker"]
    return MemoryRecord(
        record_id=str(bag.get(RECORD_ID_KEY) or payload.get("id") or ""),
        layer=resolve_layer(str(getattr(layer, "value", layer))),
        text=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``
        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)),
        updated_at=store_time(payload.get("updated_at")),
    )

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 
101
102
103
104
105
106
107
108
109
110
111
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 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
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 
132
133
134
135
136
137
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 positional rows without claiming their kinds are native.

Source code in src/symfonic/capabilities/memory/scope_codec.py 
26
27
28
29
30
31
32
33
34
def scope_from_legacy_path(path: str) -> MemoryScope:
    """Read positional rows without claiming their kinds are native."""
    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 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def scope_from_path(path: str) -> MemoryScope:
    """Parse the canonical path form back into a scope."""
    if type(path) is not str:
        raise MemoryContractError("scope path must be a string")
    if path.startswith("["):
        try:
            value = json.loads(path)
            if (
                type(value) is not list
                or len(value) != 3
                or value[0] != _NATIVE_FORMAT
                or type(value[1]) is not int
                or value[1] != 1
                or type(value[2]) is not list
                or any(type(level) is not list for level in value[2])
            ):
                raise ValueError("invalid native envelope")
            scope = MemoryScope.from_levels(tuple(tuple(level) for level in value[2]))
            if scope.path != path:
                raise ValueError("noncanonical native envelope")
            return scope
        except (ValueError, TypeError) as error:
            raise MemoryContractError("invalid native scope identity envelope") from error
    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 
259
260
261
262
263
264
265
266
267
268
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)