async def write_records(
store: Any, request: WriteRequest, *, operation: Any = None,
) -> WriteReceipt:
"""Stage records without replacing a human/policy sealed procedure ID."""
request.validate()
scope = tenant_scope(request.scope)
root = tenant_scope(request.scope.ancestor(1))
batch_id = f"{request.scope.path}:{id(request)}"
accepted: list[str] = []
selected_claims = []
rejected: list[tuple[str, str]] = []
claims: list[tuple[str, str, str]] = []
from symfonic.capabilities.memory.graph_direct_claims import external_transaction
from symfonic.capabilities.memory.mutations import _ReceiptMutationFence
from symfonic.capabilities.memory.operation_receipt import MemoryOperationVerb
outer = external_transaction(store, request.scope)
fence = store._mutations
receipt_claims = {}
if isinstance(fence, _ReceiptMutationFence):
from symfonic.capabilities.memory.selected_records import PendingRecordClaim
# Allocate before opening the one-shot receipt transaction. A failed
# receipt never authorizes replay or regeneration of these claims.
receipt_claims = {record.record_id: PendingRecordClaim(
record.record_id, request.scope, uuid4().hex + uuid4().hex,
uuid4().hex + uuid4().hex) for record in request.records}
hold = (fence.hold(request.scope, operation=operation, verb=MemoryOperationVerb.WRITE)
if isinstance(fence, _ReceiptMutationFence) else fence.hold(
scope, **({"path": request.scope.path} if request.scope.identity_version == 1
else {})))
# Take the review lock first: a Postgres fence may reserve a pool
# connection, which the review holder itself needs to finish.
async with (
review_lock_for(store._graph), # noqa: SLF001
hold,
_native_batch_savepoint(store, request),
):
for record in request.records:
# IDs are tenant-global in the graph backend; a scoped get_node
# can miss a sealed row in a sibling or descendant scope.
occupied = await store._graph.query_subtree( # noqa: SLF001
root, {"id": NodeId(record.record_id)}, limit=1,
)
existing = occupied[0] if occupied else None
if (
existing is not None
and existing.layer == MemoryLayer.PROCEDURAL
and procedure_is_sealed_terminal(existing.properties)
):
reason = "reviewed procedure id is immutable"
if (
request.scope.identity_version
and existing.properties.get("scope_path") != scope.scope_path
):
reason = "node id already exists in a different scope"
rejected.append((record.record_id, reason))
continue
payload = legacy_node_payload(record)
durable = getattr(store._graph, "durable", None) # noqa: SLF001
if (request.scope.identity_version and store._vectors is None # noqa: SLF001
and durable is not None
and type(durable).__module__ in PG_GRAPH_MODULES):
raise MemoryContractError("native pending writes require exact PostgreSQL graph")
native_pg = bool(request.scope.identity_version # noqa: SLF001
and store._vectors is None # noqa: SLF001
and exact_pg_graph(store._graph)) # noqa: SLF001
if native_pg:
from symfonic.capabilities.memory.native_pending_claims import active, bound
from symfonic.memory.backends.postgres_pending_claims import state_of
owned = {
record_id: token for path, record_id, token in active()
if path == request.scope.path
} if bound() else {
record_id: token for (path, record_id), token
in store._direct_claims.items() if path == request.scope.path # noqa: SLF001
}
remembered = owned.get(record.record_id)
if remembered is not None:
state = await state_of(store._graph.pool, scope, record.record_id, remembered)
if state != "pending" and (bound() or outer):
raise MemoryContractError(
"reconcile prior native claim before same-ID rewrite"
)
token = (remembered if remembered is not None and state == "pending"
else "v11:" + uuid4().hex)
else:
token = (receipt_claims[record.record_id].pending_claim
if receipt_claims else batch_id)
payload["properties"] = pending_properties(
payload["properties"], batch_id=token,
)
if receipt_claims:
from symfonic.capabilities.memory.selected_records import (
CLAIM_KEY,
GENERATION_KEY,
OWNER_KEY,
owner_document,
)
claim = receipt_claims[record.record_id]
payload["properties"].update({CLAIM_KEY: claim.pending_claim,
GENERATION_KEY: claim.publication_generation,
OWNER_KEY: owner_document(claim.owner)})
try:
await store._graph.add_node( # noqa: SLF001
scope, MemoryNode(**as_node_kwargs(payload)),
)
except NodeWriteConflict as conflict:
rejected.append((record.record_id, str(conflict)))
continue
accepted.append(record.record_id)
if receipt_claims:
selected_claims.append(receipt_claims[record.record_id])
if native_pg:
claims.append((request.scope.path, record.record_id, token))
if claims:
return _ClaimedWriteReceipt(
accepted=tuple(accepted), rejected=tuple(rejected), _claims=tuple(claims),
)
manifest = None
if isinstance(fence, _ReceiptMutationFence):
from symfonic.capabilities.memory.selected_records import SelectedRecordManifest
manifest = SelectedRecordManifest(tuple(selected_claims))
return WriteReceipt(accepted=tuple(accepted), rejected=tuple(rejected),
selected_manifest=manifest)