Skip to content

symfonic.memory.backends.postgres_conversation_generation

postgres_conversation_generation

Durable erase generations: what a turn read, proved still current at write time.

An erase intent fences reads while an admin forget runs. It cannot fence a turn that read before the intent committed: that turn holds erased transcript or memory in its context and may stage, publish or append it after the forget returned a completed receipt (CORE-199 P3-4).

Every admin forget advances a generation row for its scope in its transcript erase, the transaction that also retires the intent; until then the standing intent refuses every check (shared pools have no intent and erase in one transaction). A turn captures the generations covering its conversation before it reads anything, and every memory mutation and the final transcript append re-read them under the cursor-table lock the advance conflicts with. A changed or new covering row means the turn's reads predate an erase, so the write refuses. Rows are never deleted, so a stale turn stays stale after the intent clears or the process restarts.

Rows hold public scope paths compared by complete level, as intents do. A read names the path's own complete-level ancestors -- a finite list derived from the path, at most :data:MAX_DEPTH long -- and looks each up by primary key, so it does not grow with the table, which only ever grows.

advance async 

advance(conn: Any, scope: MemoryScope) -> None

Invalidate every turn under scope; the caller holds the cursor lock.

A shared-pool forget takes the tenant lock first, as a mutation does, so the two orders agree and cannot deadlock.

Source code in src/symfonic/memory/backends/postgres_conversation_generation.py 
106
107
108
109
110
111
112
113
114
115
116
async def advance(conn: Any, scope: MemoryScope) -> None:
    """Invalidate every turn under ``scope``; the caller holds the cursor lock.

    A shared-pool forget takes the tenant lock first, as a mutation does, so
    the two orders agree and cannot deadlock.
    """
    from symfonic.memory.backends.postgres_conversation_erase_log import log_advance

    await require_installed(conn)
    await conn.execute(_ADVANCE, scope.path)
    await log_advance(conn, scope.path)  # CORE-171: a standalone log's anchored lineage

capture async 

capture(pool: Any, path: str) -> tuple[tuple[str, int], ...]

The generations covering path now; take before the turn reads.

A stale snapshot can only return an older token, which refuses later.

Source code in src/symfonic/memory/backends/postgres_conversation_generation.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
async def capture(pool: Any, path: str) -> tuple[tuple[str, int], ...]:
    """The generations covering ``path`` now; take before the turn reads.

    A stale snapshot can only return an older token, which refuses later.
    """
    conn = await pool.acquire()
    try:
        return await _covering(conn, path)
    except MemoryContractError:
        raise
    except Exception as exc:  # noqa: BLE001 - one unavailable-store contract
        raise MemoryUnavailable("conversation erase generation could not be read") from exc
    finally:
        await pool.release(conn)

covering_paths 

covering_paths(path: str) -> tuple[str, ...]

path and each complete-level ancestor, root first, all canonical.

Exactly the stored paths whose scope covers path: an advance stores scope.path, and a scope covers path only as one of its ancestors of the same identity version. A noncanonical or overdeep path is refused.

Source code in src/symfonic/memory/backends/postgres_conversation_generation.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
def covering_paths(path: str) -> tuple[str, ...]:
    """``path`` and each complete-level ancestor, root first, all canonical.

    Exactly the stored paths whose scope covers ``path``: an advance stores
    ``scope.path``, and a scope covers ``path`` only as one of its ancestors
    of the same identity version. A noncanonical or overdeep path is refused.
    """
    scope = scope_from_path(path)
    depth = len(scope.segments)
    if depth > MAX_DEPTH:
        raise MemoryContractError(
            f"a fenced conversation scope has at most {MAX_DEPTH} levels (CORE-199)")
    return tuple(scope.ancestor(level).path for level in range(1, depth + 1))

observe async 

observe(pool: Any, path: str, *, tolerant: bool) -> tuple[tuple[str, int], ...]

:func:capture, except a tolerant authority without the table reads empty.

Memory's own pool is tolerant: a database no conversation log was installed in has had no joint forget, and a table created later starts empty, so an empty token stays exact.

Source code in src/symfonic/memory/backends/postgres_conversation_generation.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
async def observe(pool: Any, path: str, *, tolerant: bool) -> tuple[tuple[str, int], ...]:
    """:func:`capture`, except a ``tolerant`` authority without the table reads empty.

    Memory's own pool is tolerant: a database no conversation log was
    installed in has had no joint forget, and a table created later starts
    empty, so an empty token stays exact.
    """
    conn = await pool.acquire()
    try:
        if not await _tolerated(conn, tolerant):
            return ()
        return await _covering(conn, path)
    except MemoryContractError:
        raise
    except Exception as exc:  # noqa: BLE001 - one unavailable-store contract
        raise MemoryUnavailable("conversation erase generation could not be read") from exc
    finally:
        await pool.release(conn)

require async 

require(pool: Any, path: str, token: tuple[tuple[str, int], ...], *, tolerant: bool = False) -> None

Prove token current in a transaction that conflicts with an advance.

A memory mutation calls this while holding the tenant mutation lock, which the memory erase also takes. Either this check precedes the advance, and the erase then waits for the mutation and removes it, or it sees the advance and refuses.

Source code in src/symfonic/memory/backends/postgres_conversation_generation.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
async def require(
    pool: Any, path: str, token: tuple[tuple[str, int], ...], *, tolerant: bool = False,
) -> None:
    """Prove ``token`` current in a transaction that conflicts with an advance.

    A memory mutation calls this while holding the tenant mutation lock, which
    the memory erase also takes. Either this check precedes the advance, and
    the erase then waits for the mutation and removes it, or it sees the
    advance and refuses.
    """
    joined = bound_connection(pool) is not None
    try:
        async with pool.transaction() as locked:
            if not tolerant:
                await _fresh(locked, joined)
            conn = await pool.acquire()
            try:
                if not await _tolerated(conn, tolerant):
                    return
                if tolerant:
                    await _fresh(conn, joined)
                await require_on(conn, path, token)
            finally:
                await pool.release(conn)
    except MemoryContractError:
        raise
    except Exception as exc:  # noqa: BLE001 - fail closed, hide backend details
        raise MemoryUnavailable("conversation erase generation could not be checked") from exc

require_accounted async 

require_accounted(pool: Any, path: str, accounted: frozenset[tuple[str, int]], *, tolerant: bool, admitted: bool) -> None

Refuse a mutation of path unless accounted names every covering row.

The mutation's own scope, not the turn's: a mutation that presents no admission accounts for nothing, so it may not write where any completed joint forget reached. Ordered against an advance exactly as :func:require.

Source code in src/symfonic/memory/backends/postgres_conversation_generation.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
async def require_accounted(
    pool: Any, path: str, accounted: frozenset[tuple[str, int]], *,
    tolerant: bool, admitted: bool,
) -> None:
    """Refuse a mutation of ``path`` unless ``accounted`` names every covering row.

    The mutation's own scope, not the turn's: a mutation that presents no
    admission accounts for nothing, so it may not write where any completed
    joint forget reached. Ordered against an advance exactly as :func:`require`.
    """
    joined = bound_connection(pool) is not None
    try:
        async with pool.transaction() as locked:
            if not tolerant:
                await _fresh(locked, joined)
            conn = await pool.acquire()
            try:
                if not await _tolerated(conn, tolerant):
                    return
                if tolerant:
                    await _fresh(conn, joined)
                await refuse_if_pending(conn, path)
                rows = await _covering(conn, path)
            finally:
                await pool.release(conn)
    except MemoryContractError:
        raise
    except Exception as exc:  # noqa: BLE001 - fail closed, hide backend details
        raise MemoryUnavailable("conversation erase generation could not be checked") from exc
    if not set(rows) <= accounted:
        raise MemoryContractError(STALE_TURN if admitted else ERASED_SCOPE)

require_installed async 

require_installed(conn: Any) -> None

Refuse before any erase step when the generation table is absent or shadowed.

Source code in src/symfonic/memory/backends/postgres_conversation_generation.py 
75
76
77
78
79
80
81
async def require_installed(conn: Any) -> None:
    """Refuse before any erase step when the generation table is absent or shadowed."""
    await refuse_if_shadowed(conn)
    if not await conn.fetchval(
        "SELECT to_regclass('native_conversation_generations') IS NOT NULL"
    ):
        raise MemoryContractError(UNMIGRATED)

require_on async 

require_on(conn: Any, path: str, token: tuple[tuple[str, int], ...]) -> None

Refuse unless token is current; the caller holds the cursor lock.

Source code in src/symfonic/memory/backends/postgres_conversation_generation.py 
135
136
137
138
139
async def require_on(conn: Any, path: str, token: tuple[tuple[str, int], ...]) -> None:
    """Refuse unless ``token`` is current; the caller holds the cursor lock."""
    await refuse_if_pending(conn, path)
    if await _covering(conn, path) != token:
        raise MemoryContractError(STALE_TURN)