symfonic.memory.backends.postgres_conversation

postgres_conversation

Durable, exact-conversation append/tail storage on the shared Postgres pool.

The cursor row is locked for the whole append transaction. Two processes publishing to one conversation therefore receive distinct monotonically increasing positions; UUIDs and wall-clock timestamps never order a window. The caller must authorize the conversation before passing its key here.

PostgresConversationLog

PostgresConversationLog(pool: Any, *, erase_anchor: Any = None, receipt_key: Any = None)

A durable log; all identities have already passed an authorizer.

Since 11.3.0 a log used on its own needs erase_anchor= (a PostgresEraseAnchor with a name unique to this database) and receipt_key=; composed beside raw PostgreSQL v1 memory, memory's eraser guards its pool instead. Otherwise every read and write refuses (postgres_conversation_anchor).

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

def __init__(self, pool: Any, *, erase_anchor: Any = None, receipt_key: Any = None) -> None:
    self._pool = pool
    self._proof = anchored(pool, erase_anchor, receipt_key)

pool property

pool: Any

The shared transaction domain when composed with other backends.

proof property

proof: Any

The conversation erase proof, or None when memory's anchor must guard the pool.

append_final async

append_final(key: ConversationKey, run_id: str, user: str, assistant: str, *, generation: tuple[tuple[str, int], ...] | None = None) -> int

Publish one exchange; generation is the turn's captured erase token.

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

async def append_final(
    self, key: ConversationKey, run_id: str, user: str, assistant: str,
    *, generation: tuple[tuple[str, int], ...] | None = None,
) -> int:
    """Publish one exchange; ``generation`` is the turn's captured erase token."""
    if not key.scope_path or not key.session_id or not run_id:
        raise MemoryContractError("conversation append requires an exact key and run_id")
    spoken = tuple(
        (speaker, body.strip())
        for speaker, body in (("user", user), ("assistant", assistant))
        if body.strip()
    )
    if not spoken:
        return 0
    self._proven()
    joined = bound_connection(self._pool) is not None
    try:
        async with self._pool.transaction() as locked:
            # No guard read may fix a repeatable-read snapshot before this lock.
            await lock_fresh(locked, "ROW EXCLUSIVE", joined=joined)
            conn = await self._pool.acquire()
            try:
                await refuse_if_pending(conn, key.scope_path)
                if generation is not None:
                    await require_on(conn, key.scope_path, generation)
                await conn.execute(
                    "INSERT INTO native_conversation_cursors "
                    "(scope_path, session_id) VALUES ($1, $2) ON CONFLICT DO NOTHING",
                    key.scope_path,
                    key.session_id,
                )
                position = await conn.fetchval(
                    "SELECT next_position FROM native_conversation_cursors "
                    "WHERE scope_path = $1 AND session_id = $2 FOR UPDATE",
                    key.scope_path,
                    key.session_id,
                )
                prior = await conn.fetch(
                    "SELECT speaker, body FROM native_conversation_turns "
                    "WHERE scope_path = $1 AND session_id = $2 AND run_id = $3",
                    key.scope_path,
                    key.session_id,
                    run_id,
                )
                if prior:
                    if {row["speaker"]: row["body"] for row in prior} != dict(spoken):
                        raise MemoryContractError(
                            "conversation run_id was already published with different text"
                        )
                    return 0
                for speaker, body in spoken:
                    position += 1
                    await conn.execute(
                        "INSERT INTO native_conversation_turns "
                        "(scope_path, session_id, position, run_id, speaker, body) "
                        "VALUES ($1, $2, $3, $4, $5, $6)",
                        key.scope_path,
                        key.session_id,
                        position,
                        run_id,
                        speaker,
                        body,
                    )
                await conn.execute(
                    "UPDATE native_conversation_cursors SET next_position = $3 "
                    "WHERE scope_path = $1 AND session_id = $2",
                    key.scope_path,
                    key.session_id,
                    position,
                )
            finally:
                await self._pool.release(conn)
    except MemoryContractError:
        raise
    except Exception as exc:  # noqa: BLE001 - one unavailable-store contract
        raise MemoryUnavailable("conversation history could not be published") from exc
    return len(spoken)

export async

export(scope: MemoryScope) -> tuple[dict[str, Any], ...]

Return rows owned by this scope or descendants for a privacy export.

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

async def export(self, scope: MemoryScope) -> tuple[dict[str, Any], ...]:
    """Return rows owned by this scope or descendants for a privacy export."""
    self._proven()
    return await export(self._pool, scope)

forget async

forget(scope: MemoryScope) -> int

Erase this scope's log and descendants; FK cascade removes turns.

An anchored log logs the erase in its transaction and returns only once the anchor has recorded it, else NOT_ANCHORED; inside a caller transaction it cannot, so it refuses UNANCHORABLE first.

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

async def forget(self, scope: MemoryScope) -> int:
    """Erase this scope's log and descendants; FK cascade removes turns.

    An anchored log logs the erase in its transaction and returns only once
    the anchor has recorded it, else ``NOT_ANCHORED``; inside a caller
    transaction it cannot, so it refuses ``UNANCHORABLE`` first.
    """
    if scope.identity_version not in (0, 1):
        raise MemoryContractError("unsupported conversation scope identity version")
    if self._proof is None:
        self._proven()
        return await self._erase(scope)
    if bound_connection(self._pool) is not None:
        raise ConversationAnchorError(UNANCHORABLE)
    count = await self._erase(scope)
    await self._proof.acknowledge(self._pool)
    return count

fork async

fork(request: ConversationForkRequest) -> ConversationForkResult

Atomically clone an authorized transcript into an unused session.

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

async def fork(self, request: ConversationForkRequest) -> ConversationForkResult:
    """Atomically clone an authorized transcript into an unused session."""
    from symfonic.memory.backends.postgres_conversation_fork import fork

    self._proven()
    return await fork(self._pool, request)