async def add_node(pool: Any, scope: Any, node: Any) -> Any:
"""The conflict predicate is evaluated after PostgreSQL locks the winner."""
from symfonic.memory.backends.postgres_erase_jobs import refuse_if_erasing
from symfonic.memory.backends.postgres_graph import (
_encode_embedding,
_node_properties_jsonb,
)
from symfonic.memory.backends.postgres_pending_claims import record, token_of
conn = await pool.acquire()
try:
async with conn.transaction():
row = await conn.fetchrow(
f"""
INSERT INTO memory_nodes
(id, tenant_id, layer, label, properties, embedding,
created_at, updated_at, access_count, importance, scope_path)
VALUES ($1, $2, $3, $4, $5::jsonb, $6::vector, $7, $8, $9, $10, $11)
ON CONFLICT (id, tenant_id) DO UPDATE
SET label=EXCLUDED.label,
properties=EXCLUDED.properties,
embedding=EXCLUDED.embedding,
updated_at=EXCLUDED.updated_at,
scope_path=EXCLUDED.scope_path,
graph_incarnation=pg_catalog.gen_random_uuid()
WHERE memory_nodes.id COLLATE "C" = EXCLUDED.id COLLATE "C"
AND memory_nodes.tenant_id COLLATE "C" =
EXCLUDED.tenant_id COLLATE "C"
AND memory_nodes.scope_path COLLATE "C" =
EXCLUDED.scope_path COLLATE "C"
AND memory_nodes.layer = EXCLUDED.layer
AND (NOT {pending_jsonb_condition('memory_nodes.properties', 12)}
OR (NOT starts_with(COALESCE(
memory_nodes.properties->>'__symfonic_pending_batch__', ''),
'v11:') AND NOT starts_with(COALESCE(
EXCLUDED.properties->>'__symfonic_pending_batch__', ''),
'v11:'))
OR (memory_nodes.properties->>'__symfonic_pending_batch__'
= EXCLUDED.properties->>'__symfonic_pending_batch__'
AND EXCLUDED.properties ? '__symfonic_pending_batch__'))
AND NOT (
memory_nodes.layer = 'procedural'
AND COALESCE(
CASE WHEN memory_nodes.properties ? 'status'
THEN memory_nodes.properties->>'status'
ELSE memory_nodes.properties->'record_metadata'->>'status'
END, 'draft'
) IN ('approved', 'rejected')
AND COALESCE(
memory_nodes.properties->>'procedure_review_authority', ''
) IN ('human', 'policy')
)
RETURNING id
""",
str(node.id), scope.tenant_id, node.layer.value, node.label,
_node_properties_jsonb(node), _encode_embedding(node.embedding, conn),
node.created_at, node.updated_at, node.access_count,
node.importance, scope.scope_path, PENDING_KEY,
)
# The upsert holds ROW EXCLUSIVE and, on conflict, the row lock.
await refuse_if_erasing(conn, scope.tenant_id, (scope,))
if row is None:
await _raise_conflict(conn, scope, node)
token = node.properties.get(PENDING_BATCH_KEY)
if is_pending(node.properties) and isinstance(token, str) and token.startswith("v11:"):
await record(conn, scope, str(node.id), token_of(node.properties))
finally:
await pool.release(conn)
return node