Skip to content

symfonic.memory.backends.postgres_rendered_ack

postgres_rendered_ack

Access acknowledgement is part of the exact accepted journal transaction.

acknowledge_on async 

acknowledge_on(conn, scope, run_id, manifest, *, linked_edges=None)

Exact row/generation CAS and ACCEPTED insert share one rollback boundary.

The completion PK fences invocation replay. A durable completion read never calls this function. A refusal rolls back all access updates and acceptance. Replacement after handoff refuses the entire completion; no stale row is silently skipped, and the caller gains no authority to replay model/tools.

Source code in src/symfonic/memory/backends/postgres_rendered_ack.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
async def acknowledge_on(conn, scope, run_id, manifest, *, linked_edges=None):
    """Exact row/generation CAS and ACCEPTED insert share one rollback boundary.

    The completion PK fences invocation replay. A durable completion read never
    calls this function. A refusal rolls back all access updates and acceptance.
    Replacement after handoff refuses the entire completion; no stale row is
    silently skipped, and the caller gains no authority to replay model/tools.
    """
    from symfonic.memory.backends.postgres_graph import row_to_node

    records = decode(manifest, scope, run_id)
    # Match the table-before-row order used by native node mutation/forget.
    await conn.execute("LOCK TABLE memory_nodes IN ROW EXCLUSIVE MODE")
    await _refuse_if_erasing(conn, scope, records)
    if any("edge" in item for _, item in records):
        from symfonic.memory.backends.postgres_linked_ack import prepare_linked_on

        await prepare_linked_on(conn, scope, records, linked_edges)
    for publication, item in sorted(records, key=lambda record: record[0].record_id):
        row = await conn.fetchrow(
            'SELECT * FROM memory_nodes WHERE tenant_id COLLATE "C"=$1 '
            'AND id COLLATE "C"=$2 FOR UPDATE',
            scope.tenant, publication.record_id,
        )
        if row is None:
            _refuse()
        node = row_to_node(row)
        actual_owner = node_scope(node)
        actual = published_recall_ref(publication.record_id, actual_owner,
                                      node.properties, tenant_id=scope.tenant)
        if "edge" in item:
            from symfonic.memory.backends.postgres_linked_ack import linked_line_on

            if linked_edges is None:
                _refuse()
            rendered = await linked_line_on(conn, scope, publication, item, linked_edges)
        else:
            rendered = line(record_from_legacy_node(node_payload(node))).encode("utf-8")
        if (actual != publication or actual_owner != publication.owner
                or row["scope_path"] != storage_scope_path(publication.owner)
                or is_pending(node.properties) or is_retracted(node.properties)
                or hashlib.sha256(rendered).hexdigest() != item["line_digest"]
                or len(rendered) != item["end"] - item["start"]):
            _refuse()
        changed = await conn.fetchval(
            'UPDATE memory_nodes SET access_count=access_count+1 '
            'WHERE tenant_id COLLATE "C"=$1 AND id COLLATE "C"=$2 '
            'AND scope_path COLLATE "C"=$3 '
            'AND properties->>$4=$5 AND properties->$6=$7::jsonb RETURNING id',
            scope.tenant, publication.record_id, storage_scope_path(publication.owner),
            GENERATION_KEY, publication.generation, OWNER_KEY,
            json.dumps(owner_document(publication.owner)),
        )
        if changed != publication.record_id:
            _refuse()