symfonic.memory.backends.postgres_conversation_erase

postgres_conversation_erase

Durable transcript-erase intents: the saga record of a cross-pool admin forget.

A memory store and a conversation log on two pools cannot commit together. The admin forget therefore runs memory first and erases the transcript only after memory completes (CORE-199 P3-4). Between those two commits -- across every ForgetInProgress step, and across a crash after memory finished -- the transcript still exists while its memory is gone. An agent turn reading it then would carry erased content back into new memory.

An intent closes that window. It commits before memory is touched, and while it stands no turn reads, appends to, or forks a covered conversation. The transcript erase deletes the intent in the same transaction as the transcript, so a standing intent is durable evidence that an erase is unfinished, and the only way to retire it is the forget that finishes.

Intent rows hold public scope paths, compared by complete level. Lexical SQL narrows candidates; :meth:MemoryScope.covers decides.

Every statement here names its tables unqualified, so it means whichever schema the connection resolved them in. Two schemas on the active search_path holding a same-named native_conversation_* relation make that identity ambiguous, and a statement prepared before the earlier one existed keeps reading the later one. :func:refuse_if_shadowed refuses then, before any fence read, rather than choosing either copy.

clear async

clear(conn: Any, scope: MemoryScope, child_prefix: str) -> None

Delete intents scope covers, inside the transcript-erase transaction.

An ancestor's intent is not this forget's to retire: its erase is wider.

Source code in src/symfonic/memory/backends/postgres_conversation_erase.py

async def clear(conn: Any, scope: MemoryScope, child_prefix: str) -> None:
    """Delete intents ``scope`` covers, inside the transcript-erase transaction.

    An ancestor's intent is not this forget's to retire: its erase is wider.
    """
    if not await _present(conn):
        return
    rows = await conn.fetch(_DESCENDANTS, scope.path, child_prefix)
    covered = [row["scope_path"] for row in rows if covers(scope.path, row["scope_path"])]
    if covered:
        await conn.execute(
            "DELETE FROM native_conversation_erasures WHERE scope_path = ANY($1::text[])",
            covered,
        )

covers

covers(outer: str, inner: str) -> bool

Whether stored path outer covers inner by complete level.

Source code in src/symfonic/memory/backends/postgres_conversation_erase.py

def covers(outer: str, inner: str) -> bool:
    """Whether stored path ``outer`` covers ``inner`` by complete level."""
    try:
        return scope_from_path(outer).covers(scope_from_path(inner))
    except MemoryContractError:
        return False

lock_fresh async

lock_fresh(conn: Any, mode: str, *, joined: bool) -> None

Take the cursor-table lock so every later read in the block sees past it.

The block's own transaction issues the lock as its first statement, so its snapshot follows the lock at any isolation. A joined caller transaction may have fixed its snapshot before an intent committed; only READ COMMITTED re-reads per statement, so any other isolation is refused before a read. The refusal precedes the lock: a joined transaction keeps its locks to its end, and a caller that caught the refusal would otherwise hold every erase intent out. SHOW takes no snapshot.

The shadow check reads, so it follows the lock in the block's own transaction, which a refusal rolls back, and precedes it in a joined one, which is READ COMMITTED by then.

Source code in src/symfonic/memory/backends/postgres_conversation_erase.py

async def lock_fresh(conn: Any, mode: str, *, joined: bool) -> None:
    """Take the cursor-table lock so every later read in the block sees past it.

    The block's own transaction issues the lock as its first statement, so its
    snapshot follows the lock at any isolation. A joined caller transaction may
    have fixed its snapshot before an intent committed; only READ COMMITTED
    re-reads per statement, so any other isolation is refused before a read.
    The refusal precedes the lock: a joined transaction keeps its locks to its
    end, and a caller that caught the refusal would otherwise hold every erase
    intent out. ``SHOW`` takes no snapshot.

    The shadow check reads, so it follows the lock in the block's own
    transaction, which a refusal rolls back, and precedes it in a joined one,
    which is READ COMMITTED by then.
    """
    if joined:
        if await conn.fetchval("SHOW transaction_isolation") != "read committed":
            raise MemoryContractError(STALE_SNAPSHOT)
        await refuse_if_shadowed(conn)
    await conn.execute(f"LOCK TABLE native_conversation_cursors IN {mode} MODE")
    if not joined:
        await refuse_if_shadowed(conn)

refuse_if_pending async

refuse_if_pending(conn: Any, *paths: str) -> None

Refuse when a standing intent covers any of paths.

Readers and writers call this after :func:lock_fresh, whose lock conflicts with :func:request's, so an intent is either visible here or commits after the caller's transaction and its erase removes what it added.

Source code in src/symfonic/memory/backends/postgres_conversation_erase.py

async def refuse_if_pending(conn: Any, *paths: str) -> None:
    """Refuse when a standing intent covers any of ``paths``.

    Readers and writers call this after :func:`lock_fresh`, whose lock
    conflicts with :func:`request`'s, so an intent is either visible here or
    commits after the caller's transaction and its erase removes what it added.
    """
    if not await _present(conn):
        return
    for path in paths:
        rows = await conn.fetch(_ANCESTORS, path)
        if any(covers(row["scope_path"], path) for row in rows):
            raise MemoryContractError(ERASE_PENDING)

refuse_if_shadowed async

refuse_if_shadowed(conn: Any) -> None

Refuse when a conversation relation's schema is ambiguous on conn.

Checked per call, not once per connection: a shadow created after the connection prepared its statements leaves those plans on the old copy.

Source code in src/symfonic/memory/backends/postgres_conversation_erase.py

async def refuse_if_shadowed(conn: Any) -> None:
    """Refuse when a conversation relation's schema is ambiguous on ``conn``.

    Checked per call, not once per connection: a shadow created after the
    connection prepared its statements leaves those plans on the old copy.
    """
    if await conn.fetchval(_SHADOWED) is not None:
        raise MemoryContractError(SHADOWED)

request async

request(pool: Any, scope: MemoryScope) -> None

Commit an intent for scope; idempotent for a resumed forget.

Refuses first if the erase cannot finish: its transcript erase advances the erase generation (postgres_conversation_generation).

Source code in src/symfonic/memory/backends/postgres_conversation_erase.py

async def request(pool: Any, scope: MemoryScope) -> None:
    """Commit an intent for ``scope``; idempotent for a resumed forget.

    Refuses first if the erase cannot finish: its transcript erase advances
    the erase generation (``postgres_conversation_generation``).
    """
    from symfonic.memory.backends.postgres_conversation_generation import require_installed

    if scope.identity_version != 1:
        raise MemoryContractError("transcript erase intents are kind-aware v1 only")
    try:
        async with pool.transaction():
            conn = await pool.acquire()
            try:
                # Wait out in-flight appends and forks; later ones see the intent.
                await require_installed(conn)
                await conn.execute(
                    "LOCK TABLE native_conversation_cursors IN SHARE ROW EXCLUSIVE MODE"
                )
                await conn.execute(
                    "INSERT INTO native_conversation_erasures (scope_path) VALUES ($1) "
                    "ON CONFLICT DO NOTHING",
                    scope.path,
                )
            finally:
                await pool.release(conn)
    except Exception as exc:  # noqa: BLE001 - one unavailable-store contract
        raise MemoryUnavailable("conversation erase could not be requested") from exc