Skip to content

symfonic.memory.backends.postgres_conversation_fork

postgres_conversation_fork

Atomic conversation snapshot and restart-stable fork receipt.

forget_positional async 

forget_positional(pool: Any, scope: MemoryScope) -> int

One transaction for positional transcript and owner receipt erasure.

Source code in src/symfonic/memory/backends/postgres_conversation_fork.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
async def forget_positional(pool: Any, scope: MemoryScope) -> int:
    """One transaction for positional transcript and owner receipt erasure."""
    try:
        async with pool.transaction():
            conn = await pool.acquire()
            try:
                # Acquire the same table lock before touching receipts.
                await conn.execute(
                    "LOCK TABLE native_conversation_cursors IN SHARE ROW EXCLUSIVE MODE"
                )
                await forget_receipts(conn, scope)
                rows = await conn.fetch(
                    "DELETE FROM native_conversation_cursors WHERE scope_path = $1 "
                    "OR left(scope_path, length($1) + 1) = $1 || '/' RETURNING scope_path",
                    scope.path,
                )
                return len(rows)
            finally:
                await pool.release(conn)
    except Exception as exc:  # noqa: BLE001 - one unavailable-store contract
        raise MemoryUnavailable("conversation history could not be erased") from exc

forget_receipts async 

forget_receipts(conn: Any, scope: MemoryScope, native_child_prefix: str | None = None) -> None

Erase owner metadata; session-only erasure retains retry tombstones.

Source and destination share the complete owner, so matching the receipt owner removes only requests whose entire identity is covered by this erase.

Source code in src/symfonic/memory/backends/postgres_conversation_fork.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
async def forget_receipts(
    conn: Any, scope: MemoryScope, native_child_prefix: str | None = None,
) -> None:
    """Erase owner metadata; session-only erasure retains retry tombstones.

    Source and destination share the complete owner, so matching the receipt
    owner removes only requests whose entire identity is covered by this erase.
    """
    if scope.identity_version == 1 and native_child_prefix is None:
        raise MemoryContractError("native receipt erase requires the canonical child prefix")
    prefix = native_child_prefix if scope.identity_version == 1 else scope.path + "/"
    candidates = await conn.fetch(
        'SELECT DISTINCT owner_path FROM native_conversation_forks '
        'WHERE owner_path COLLATE "C" = $1 '
        'OR left(owner_path, length($2)) COLLATE "C" = $2', scope.path, prefix,
    )
    paths = []
    for row in candidates:
        try:
            if scope.covers(scope_from_path(row["owner_path"])):
                paths.append(row["owner_path"])
        except MemoryContractError:
            continue
    if paths:
        await conn.execute("DELETE FROM native_conversation_forks "
                           "WHERE owner_path = ANY($1::text[])", paths)
    destinations = await conn.fetch(
        'SELECT DISTINCT destination_path FROM native_conversation_forks '
        'WHERE destination_path COLLATE "C" = $1 '
        'OR left(destination_path, length($2)) COLLATE "C" = $2', scope.path, prefix,
    )
    paths = []
    for row in destinations:
        try:
            if scope.covers(scope_from_path(row["destination_path"])):
                paths.append(row["destination_path"])
        except MemoryContractError:
            continue
    if paths:
        await conn.execute("UPDATE native_conversation_forks SET destination_forgotten=TRUE "
                           "WHERE destination_path = ANY($1::text[])", paths)