Skip to content

symfonic.capabilities.memory.cycle_writes

cycle_writes

A phase produces requests; only the fenced cycle may reach a write port.

CycleWrites 

CycleWrites(target: MemoryWriteCoordinator)

Bases: MemoryWriteCoordinator

Run-local request buffer with phase offsets, not a scope-wide discard.

Accepted means accepted into this cycle's journal, not persisted. Store refusal at commit aborts the same transaction as graph mutations. No raw pending row (or overwritten record) exists before then, so rollback cannot delete a concurrent turn's or an earlier phase's data.

Source code in src/symfonic/capabilities/memory/cycle_writes.py 
52
53
54
55
56
57
58
def __init__(self, target: MemoryWriteCoordinator) -> None:
    super().__init__(writes=target.write_port, lifecycle=target.lifecycle)
    self._target = target
    self._requests: list[WriteRequest] = []
    self._offset = 0
    self._native_checkpoint: dict[str, PendingSnapshot] | None = None
    self._native_replay_claims: tuple[tuple[str, str, str], ...] = ()

confirm_native 

confirm_native(committed: tuple[str, ...]) -> None

Retire only unchanged, confirmed pending entries after outer COMMIT.

Source code in src/symfonic/capabilities/memory/cycle_writes.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def confirm_native(
    self, committed: tuple[str, ...],
) -> None:
    """Retire only unchanged, confirmed pending entries after outer COMMIT."""
    checkpoint = self._native_checkpoint
    if checkpoint is None:
        raise MemoryContractError("native cycle has no flush checkpoint to confirm")
    self._native_checkpoint = None
    confirmed = set(committed)
    retireable = list(self._native_replay_claims)
    self._native_replay_claims = ()
    retireable.extend(retire_snapshot(self._target._pending, checkpoint, confirmed))
    from symfonic.capabilities.memory.graph_direct_claims import retire_exact

    retire_exact(self._target.lifecycle, tuple(retireable), confirmed)

join async 

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

Settle the target's background writes before the commit lock.

Source code in src/symfonic/capabilities/memory/cycle_writes.py 
89
90
91
async def join(self) -> tuple[Any, ...]:
    """Settle the target's background writes before the commit lock."""
    return await self._target.join()

pending_checkpoint 

pending_checkpoint(scope: MemoryScope) -> dict[str, PendingSnapshot]

Capture pending entries at the native lifecycle boundary.

Source code in src/symfonic/capabilities/memory/cycle_writes.py 
93
94
95
def pending_checkpoint(self, scope: MemoryScope) -> dict[str, PendingSnapshot]:
    """Capture pending entries at the native lifecycle boundary."""
    return snapshot_pending(self._target._pending, self._target._covered_paths(scope))