Writing to the legacy layer stores through this capability's write port.
LegacyRetrievalPort's counterpart, and it exists for the same reason: an
engine whose memories live in the legacy stores could not otherwise hand the
kernel a bundle that serves consolidation, so every deployment that records
memories stayed on the legacy body.
Staged, not published. The write/flush split has no legacy equivalent —
legacy writes and the row is live. symfonic.memory.pending already settled
how a persistent adapter represents an unpublished row: durability
="transient" plus the namespaced pending keys, chosen because the legacy
reader already hides those rows. A marker legacy did not honour would make the
rollback story fiction, so this port uses that contract rather than inventing a
second one. Publishing is :meth:flush; a turn that never flushes leaves rows
the legacy reader will not surface.
Refusals are named. A record whose layer has no store is rejected with its
reason, never dropped. A memory that vanishes with nothing said is the failure
this adapter family keeps being written to avoid.
Unreachable degrades. The store being down ends the write, not the turn:
falling back is what the switch is for, and raising here would turn a
capability gap into an outage.
LegacyWritePort
LegacyWritePort(layers: Mapping[MemoryLayer, Any], *, scope_of: Any, graph: Any = None)
Stages records into the legacy layer stores, and publishes on flush.
Parameters:
| Name |
Type |
Description |
Default |
layers
|
Mapping[MemoryLayer, Any]
|
the legacy store for each layer this port may write to. A
layer absent here is a layer this deployment cannot record to,
and a record aimed at one is refused by name rather than
written somewhere else.
|
required
|
scope_of
|
Any
|
turns a capability :class:MemoryScope into whatever the
legacy stores take. Injected for the reason the read port
injects it: this module never depends on the transport that
happens to own the translation today.
|
required
|
Source code in src/symfonic/capabilities/memory/legacy_write.py
| def __init__(
self,
layers: Mapping[MemoryLayer, Any],
*,
scope_of: Any,
graph: Any = None,
) -> None:
"""
Args:
layers: the legacy store for each layer this port may write to. A
layer absent here is a layer this deployment cannot record to,
and a record aimed at one is refused by name rather than
written somewhere else.
scope_of: turns a capability :class:`MemoryScope` into whatever the
legacy stores take. Injected for the reason the read port
injects it: this module never depends on the transport that
happens to own the translation today.
"""
self._layers = dict(layers)
self._scope_of = scope_of
#: ``scope path -> (batch id, [entry payloads])``. Keyed by scope so a
#: flush commits one tenant's staging and never another's.
self._staged: dict[str, tuple[str, list[dict[str, Any]]]] = {}
# TA8.77. The subtree reader/eraser ``forget`` needs. Optional, so a
# deployment that wires no graph keeps the port it had -- and its
# ``forget`` says it erased nothing rather than pretending.
self._graph = graph
|
discard
async
discard(scope: MemoryScope) -> LifecycleReceipt
Drop this scope's staging without publishing any of it.
The operation a failed or cancelled turn needs, and the reason staging
had to stop being durable first: discarding a row the store already
holds is not a discard, it is a delete that has to find every copy.
Here there is nothing in the store to find.
Reports what it dropped rather than returning silently -- a turn that
threw away four records and said nothing is indistinguishable from one
that had nothing to throw away.
Source code in src/symfonic/capabilities/memory/legacy_write.py
| async def discard(self, scope: MemoryScope) -> LifecycleReceipt:
"""Drop this scope's staging without publishing any of it.
The operation a failed or cancelled turn needs, and the reason staging
had to stop being durable first: discarding a row the store already
holds is not a discard, it is a delete that has to find every copy.
Here there is nothing in the store to find.
Reports what it dropped rather than returning silently -- a turn that
threw away four records and said nothing is indistinguishable from one
that had nothing to throw away.
"""
entry = self._staged.pop(scope.path, None)
dropped = tuple(payload["id"] for payload in (entry[1] if entry else ()))
return LifecycleReceipt(
scope_path=scope.path, committed=(), discarded=dropped
)
|
flush
async
flush(scope: MemoryScope) -> LifecycleReceipt
Publish what scope staged, and what anything below it staged.
The only operation that reaches the store. Idempotent by construction:
each batch is popped, so a second flush finds nothing left and
publishes nothing rather than writing the batch again.
TA8.77 made it walk the subtree. It popped exactly one key, so a turn
that staged under acme/alice/s1 and a caller that flushed
acme/alice left the rows pending forever -- invisible by contract
and never published by anything. The scope a caller names is the scope
it owns, and it owns what is under it: the same anchoring forget
erases with and covers reads with.
Source code in src/symfonic/capabilities/memory/legacy_write.py
| async def flush(self, scope: MemoryScope) -> LifecycleReceipt:
"""Publish what ``scope`` staged, and what anything below it staged.
The only operation that reaches the store. Idempotent by construction:
each batch is popped, so a second flush finds nothing left and
publishes nothing rather than writing the batch again.
TA8.77 made it walk the subtree. It popped exactly one key, so a turn
that staged under ``acme/alice/s1`` and a caller that flushed
``acme/alice`` left the rows pending forever -- invisible by contract
and never published by anything. The scope a caller names is the scope
it owns, and it owns what is under it: the same anchoring ``forget``
erases with and ``covers`` reads with.
"""
paths = sorted(path for path in self._staged if _covered_by(scope, path))
if not paths:
return LifecycleReceipt(scope_path=scope.path)
committed: list[str] = []
for path in paths:
_batch, staged = self._staged.pop(path)
# Published under the scope it was STAGED at, not the one being
# flushed: a descendant's rows belong to the descendant, and
# restamping them would move a session's memory up to its tenant.
legacy_scope = self._scope_of(scope_from_path(path))
for payload in staged:
store = self._layers.get(MemoryLayer(payload["layer"]))
if store is None: # pragma: no cover - staged implies a store
continue
published = dict(payload)
published["metadata"] = published_properties(payload["metadata"])
try:
await store.write(legacy_scope, _entry(published))
except Exception: # noqa: BLE001
return LifecycleReceipt(
scope_path=scope.path,
committed=tuple(committed),
degraded=True,
)
committed.append(payload["id"])
return LifecycleReceipt(scope_path=scope.path, committed=tuple(committed))
|
forget
async
forget(scope: MemoryScope) -> LifecycleReceipt
Erase scope and everything below it, staged and published alike.
TA8.77, and a different operation from :meth:discard. discard
drops what this turn staged and has not published; forget erases
what is already durable. Conflating them would make a privacy request
depend on whether a turn happened to have finished.
The subtree is the unit, because a scope that erased only its own level
would leave a session's rows behind when its principal was forgotten --
and the caller has no way to enumerate what it did not erase.
Siblings and ancestors are untouched: delete_subtree is anchored at
the named scope, which is the same anchoring covers uses to decide
what a query may read.
Idempotent, and explicit about it: forgetting a scope with nothing in
it reports an empty receipt rather than raising, because "there was
nothing to erase" and "the erasure failed" must not look the same to a
caller acting on a deletion request.
Source code in src/symfonic/capabilities/memory/legacy_write.py
| async def forget(self, scope: MemoryScope) -> LifecycleReceipt:
"""Erase ``scope`` and everything below it, staged and published alike.
TA8.77, and a different operation from :meth:`discard`. ``discard``
drops what this turn staged and has not published; ``forget`` erases
what is already durable. Conflating them would make a privacy request
depend on whether a turn happened to have finished.
The subtree is the unit, because a scope that erased only its own level
would leave a session's rows behind when its principal was forgotten --
and the caller has no way to enumerate what it did not erase.
Siblings and ancestors are untouched: ``delete_subtree`` is anchored at
the named scope, which is the same anchoring ``covers`` uses to decide
what a query may read.
Idempotent, and explicit about it: forgetting a scope with nothing in
it reports an empty receipt rather than raising, because "there was
nothing to erase" and "the erasure failed" must not look the same to a
caller acting on a deletion request.
"""
erased: list[str] = []
# Staged first. A pending row the caller can no longer see is still a
# row this port would publish on the next flush.
for path in [p for p in self._staged if _covered_by(scope, p)]:
_batch, staged = self._staged.pop(path)
erased.extend(payload["id"] for payload in staged)
if self._graph is not None:
legacy_scope = self._scope_of(scope)
try:
nodes = await self._graph.query_subtree(legacy_scope, {})
erased.extend(str(node.id) for node in nodes)
await self._graph.delete_subtree(legacy_scope)
except Exception: # noqa: BLE001 - degrade, never raise
return LifecycleReceipt(
scope_path=scope.path,
discarded=tuple(sorted(set(erased))),
degraded=True,
)
return LifecycleReceipt(
scope_path=scope.path, discarded=tuple(sorted(set(erased)))
)
|
write
async
write(request: WriteRequest) -> WriteReceipt
Stage request's records. Nothing reaches the store here.
TA8.74. This used to write a pending row into the layer immediately,
and flush wrote a second, published one. Two consequences, both
measured on a generated project: one turn left four durable rows for
two records, and -- worse -- staging was durable, so the transactional
semantics TA8.71 established were never real. A turn that failed after
staging had already put its rows in the store, and no discard
implemented later could have made that atomic.
So staging is in memory and flush is the only operation that
persists. Records are validated here, and a layer that does not exist
is reported here too: finding that out at flush would report it after
the turn believed it had recorded.
Source code in src/symfonic/capabilities/memory/legacy_write.py
| async def write(self, request: WriteRequest) -> WriteReceipt:
"""Stage ``request``'s records. Nothing reaches the store here.
TA8.74. This used to write a pending row into the layer immediately,
and ``flush`` wrote a second, published one. Two consequences, both
measured on a generated project: one turn left four durable rows for
two records, and -- worse -- staging was durable, so the transactional
semantics TA8.71 established were never real. A turn that failed after
staging had already put its rows in the store, and no ``discard``
implemented later could have made that atomic.
So staging is in memory and ``flush`` is the only operation that
persists. Records are validated here, and a layer that does not exist
is reported here too: finding that out at flush would report it after
the turn believed it had recorded.
"""
request.validate()
scope_path = request.scope.path
batch, staged = self._staged.setdefault(scope_path, (uuid.uuid4().hex, []))
accepted: list[str] = []
rejected: list[tuple[str, str]] = []
for record in request.records:
if record.layer not in self._layers:
rejected.append((
record.record_id,
f"this deployment has no {record.layer.value} store, so the "
"record was not written anywhere",
))
continue
payload = legacy_entry_payload(record)
payload["metadata"] = pending_properties(
payload["metadata"], batch_id=batch
)
# Staged by id, so a record written twice in one turn replaces
# itself rather than queueing twice for publication. The port's
# contract promises an upsert; queueing two payloads with one id
# would break it before the layer ever saw them.
for index, existing in enumerate(staged):
if existing["id"] == payload["id"]:
staged[index] = payload
break
else:
staged.append(payload)
accepted.append(record.record_id)
return WriteReceipt(accepted=tuple(accepted), rejected=tuple(rejected))
|