symfonic.capabilities.memory.journal

journal

A consolidation cycle mutates the graph all at once, or not at all.

The fence stopped a cycle that lost its lease from writing again. It could not undo what the cycle had already written, and on this roster that is most of the danger: ten phase modules call add_node, update_node, delete_node, add_edge and delete_edge on the graph directly across nineteen call sites, and each one is durable the moment it returns. A cycle abandoned halfway therefore left a visible half -- strengthened importances without the merge that justified them, pruned orphans without the meta nodes they were orphaned from -- and the runtime's own promise that a cycle which recorded an error publishes nothing was true only of the staged writes.

Nor was the lease the only way to get there. Any ordinary phase failure in the middle of Deep Sleep produced the same split: _commit withheld the staged half while the direct half was already in the store.

So the cycle gets a journal. Every graph mutation goes into :class:CycleJournal instead of the store; reads are answered from an overlay of the scope with those mutations applied, so a phase sees its own work and the work of the phases before it; and at the end the whole batch is applied at once under the lease, or thrown away. Failure, cancellation and a lost lease all take the same path -- the journal is discarded and the store never knew.

What it covers, exactly. Graph mutations -- which on this roster is every mutation a phase makes. No shipped phase writes through context.writes, and the vector index is written on the turn path (GraphBackedHms.write) rather than by any phase. A future phase that staged a record through the write coordinator would have its graph row in the batch and its vector row written immediately, so a discarded cycle would leave an orphan vector; that is worth knowing before writing such a phase rather than discovering afterwards.

Bound per cycle, through a ContextVar, like the fence and like run attribution. The wrapper is built once by the composition root, long before any cycle exists, and a journal threaded through every factory argument would be a thing each future phase author has to remember. Unbound -- the default -- means no cycle is running on this task and every call passes straight through, which is what an ordinary turn does while a nap runs beside it.

CycleJournal

CycleJournal(*, cycle: str | None = None, max_new_edges_per_sweep: int | None = None)

One cycle's deferred mutations, and the scope they are read against.

Keyed by durable backend, because a deployment may compose more than one and a cycle that mixed their operations into one ordered list would replay them against the wrong store.

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

def __init__(self, *, cycle: str | None = None,
             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._ops: dict[int, list[Any]] = {}
    self._overlays: dict[int, tuple[Any, ScopeOverlay]] = {}
    self._cycle = cycle
    self._edge_limit = max_new_edges_per_sweep
    self.begin_phase()

pending property

pending: int

How many mutations are waiting. 0 for a cycle that only read.

apply async

apply() -> int

Write every recorded mutation to its store, in order. Returns the count.

The caller is responsible for having checked authority, and for doing it close enough to this call that nothing can intervene -- see :func:symfonic.capabilities.memory.commit.commit_cycle, which does both inside one database transaction where the store offers one.

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

async def apply(self) -> int:
    """Write every recorded mutation to its store, in order. Returns the count.

    The caller is responsible for having checked authority, and for doing
    it close enough to this call that nothing can intervene -- see
    :func:`symfonic.capabilities.memory.commit.commit_cycle`, which does
    both inside one database transaction where the store offers one.
    """
    written = 0
    for ops in self._ops.values():
        for durable, name, args, kwargs in ops:
            await getattr(durable, name)(*args, **kwargs)
            written += 1
    for durable, overlay in self._overlays.values():
        for scope, expected, following in getattr(overlay, "progress", {}).values():
            from symfonic.capabilities.memory.growth import following_progress
            following = following_progress(overlay, scope, following, self._edge_limit)
            await durable.save_workset_progress(scope, self._cycle, expected, following)
        getattr(overlay, "progress", {}).clear()
    self._ops.clear()
    return written

discard

discard() -> int

Throw the batch away. Returns how many mutations never happened.

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

def discard(self) -> int:
    """Throw the batch away. Returns how many mutations never happened."""
    pending = self.pending
    self._ops.clear()
    self._overlays.clear()
    return pending

read async

read(durable: Any, name: str, args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any

Answer a read from the overlay, seeding it from durable first.

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

async def read(self, durable: Any, name: str, args: tuple[Any, ...],
               kwargs: dict[str, Any]) -> Any:
    """Answer a read from the overlay, seeding it from ``durable`` first."""
    overlay = await self._overlay(durable, args[0] if args else None)
    return await getattr(overlay, name)(*args, **kwargs)

stores

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

The durable backends this cycle touched, for the commit to lock.

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

def stores(self) -> tuple[Any, ...]:
    """The durable backends this cycle touched, for the commit to lock."""
    return tuple(store for store, _ in self._overlays.values())

write async

write(durable: Any, name: str, args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any

Apply a mutation to the overlay and record it for the commit.

Applied to the overlay first so its return value is the one the durable store would have produced -- the persisted node, the upserted edge with its incremented weight -- and so the next phase reads the result rather than the input.

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

async def write(self, durable: Any, name: str, args: tuple[Any, ...],
                kwargs: dict[str, Any]) -> Any:
    """Apply a mutation to the overlay and record it for the commit.

    Applied to the overlay *first* so its return value is the one the
    durable store would have produced -- the persisted node, the upserted
    edge with its incremented weight -- and so the next phase reads the
    result rather than the input.
    """
    overlay = await self._overlay(durable, args[0] if args else None)
    reserved = None
    if self._edge_limit is not None and name in {"add_edge", "upsert_edge"}:
        from symfonic.capabilities.memory.growth import reserve
        reserved = await reserve(overlay, args[0], args[1], name, self._edge_limit)
    try:
        applied = await getattr(overlay, name)(*args, **kwargs)
    except BaseException:
        if reserved is not None:
            projected: Any = overlay
            projected.growth_created[reserved] -= 1
        raise
    self._ops.setdefault(id(durable), []).append((durable, name, args, kwargs))
    return applied

JournalledGraph

JournalledGraph(durable: Any)

The graph backend a deployment composes once, for everything.

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

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

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

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

durable property

durable: Any

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

current_journal

current_journal() -> CycleJournal | None

The journal this cycle is running under, or None outside one.

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

def current_journal() -> CycleJournal | None:
    """The journal this cycle is running under, or ``None`` outside one."""
    return _JOURNAL.get()

journalled async

journalled(journal: CycleJournal | None) -> AsyncIterator[CycleJournal | None]

Defer this task's graph mutations into journal for the duration.

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

@asynccontextmanager
async def journalled(journal: CycleJournal | None) -> AsyncIterator[CycleJournal | None]:
    """Defer this task's graph mutations into ``journal`` for the duration."""
    token = _JOURNAL.set(journal)
    try:
        yield journal
    finally:
        _JOURNAL.reset(token)