symfonic.capabilities.memory.writes

writes

Deliverable 2: the coordinator between a produced memory and a stored one.

Three jobs, and each one exists because the shipped path does it wrong in a way that only shows up under load or under failure:

• Background writes are owned. core/runtime.py and agent/engine.py each hold a module-level _background_tasks set. A write detached there outlives every run that added to it — nothing waits for it, nothing reports it, and engine.py needs an atexit hook to warn that work nobody owned may never have run. T2.3.4 replaced that with a registry that refuses unattributed work; this coordinator spawns through it, or refuses to spawn.
• Flush waits for what it commits. A flush that races an in-flight write publishes half a turn: the memories that happened to land are retrievable and the rest silently are not. :meth:MemoryWriteCoordinator.flush joins its own background writes first.
• A lost flush does not lose the write. When the store is unreachable the receipt is degraded and the pending bookkeeping stays, so the next flush still has something to commit. Clearing it would turn one transport blip into permanent memory loss with no error anywhere.

Rollback is the pending buffer. Per the port contract, a written memory is not retrievable until it is flushed, so a turn that ends badly abandons its scope and nothing it produced was ever visible — really so where the adapter implements :class:~.work.MemoryDiscardPort, and reported as unenforced where it does not.

The two borrowed surfaces (owned background work, optional pending-discard) and the per-write outcome live in :mod:.work.

MemoryWriteCoordinator

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

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

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

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

in_flight property

in_flight: int

Background writes spawned and not yet joined.

lifecycle property

lifecycle: Any

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

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

pending_ids property

pending_ids: tuple[str, ...]

Every uncommitted record id this coordinator wrote, sorted.

pending_scopes property

pending_scopes: tuple[str, ...]

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

write_port property

write_port: MemoryWritePort

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

abandon async

abandon(scope: MemoryScope) -> LifecycleReceipt

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

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

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

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

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

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

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

flush async

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

Commit scope and everything below it.

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

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

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

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

join async

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

Await every background write spawned since the last join.

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

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

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

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

write async

write(request: WriteRequest) -> WriteReceipt

Write request now, degrading rather than failing the turn.

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

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

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

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

write_in_background

write_in_background(request: WriteRequest) -> Any

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

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

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

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

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