symfonic.capabilities.memory.journal

journal

Defer a consolidation cycle's graph writes until its atomic commit.

The context-bound journal reads from an overlay, then replays mutations under commit authority. Unbound calls pass through to the durable backend.

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._node_versions: dict[tuple[int, str, str, str], Any] = {}
    self._overlays: dict[int, tuple[Any, ScopeOverlay]] = {}
    self._cycle = cycle
    self._edge_limit = max_new_edges_per_sweep
    self._sealed = None
    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.
    """
    from symfonic.capabilities.memory.journal_seal import replay
    return await replay(self)

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._node_versions.clear()
    self._overlays.clear()
    if self._sealed is not None:
        self._sealed = ()
    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 kwargs.get("scope"))
    return await getattr(overlay, name)(*args, **kwargs)

seal

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

Freeze queued writes before receipt validation or any await.

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

def seal(self) -> tuple[Any, ...]:
    """Freeze queued writes before receipt validation or any await."""
    from symfonic.capabilities.memory.journal_seal import seal
    return seal(self)

stage_guarded

stage_guarded(durable: Any, adapter: Any, name: str, args: tuple[Any, ...]) -> None

Queue a guarded native operation against the durable graph domain.

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

def stage_guarded(self, durable: Any, adapter: Any, name: str,
                  args: tuple[Any, ...]) -> None:
    """Queue a guarded native operation against the durable graph domain."""
    self._require_open()
    key = id(durable)
    if key not in self._overlays:
        self._overlays[key] = (durable, ScopeOverlay())
    self._phase.remember(key, durable, self._overlays[key][1])
    self._ops.setdefault(key, []).append((adapter, name, args, {}))

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.
    """
    self._require_open()
    from symfonic.capabilities.memory.journal_node_patch import refuse_after_patch
    refuse_after_patch(self, durable, name, args, kwargs)
    overlay = await self._overlay(durable, args[0] if args else kwargs.get("scope"))
    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
    if name == "update_owned_edge" and applied is None:
        return None
    self._require_open()
    self._ops.setdefault(id(durable), []).append((durable, name, args, kwargs))
    return applied

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

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.

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)