`symfonic.platform.conversation_pg`¶

conversation_pg ¶

Shared native HTTP conversations on PostgreSQL, safe for many live workers.

Every published exchange is one transaction whose first statement is a revision compare-and-set on the conversation row: UPDATE ... WHERE revision = <prepared revision>. Under READ COMMITTED a concurrent writer blocks on the row lock and then re-checks the predicate against the committed version, so of two workers that prepared the same revision exactly one updates a row and the other publishes nothing and raises :class:ConversationConflict. Message rows are written only inside that transaction.

Ownership is a stored column checked in the same statement as every read or write; a session id from the request is a lookup key, never proof. The composition root owns the pool and the schema migration (conversation_pg_schema.install_conversation_schema); this authority runs no DDL and refuses to start until the installed schema attests.

PostgresConversationAuthority ¶

PostgresConversationAuthority(pool: Any, *, schema: str)

Revision-checked native conversation owner shared by every worker.

close ends this authority's use of the pool; the pool itself belongs to the composition root that created it.

Source code in src/symfonic/platform/conversation_pg.py

def __init__(self, pool: Any, *, schema: str) -> None:
    _require_pool(pool)
    self._tables = ConversationTables(schema)
    self._pool = pool
    self._ready = False
    self._closed = False
    self._start_lock = asyncio.Lock()
    self._reader: PostgresTranscriptReader | None = None

pauses `property` ¶

pauses: Any

Selected-root pause records over the same tables and pool.

append_exchange `async` ¶

append_exchange(principal: Any, session_id: str, revision: int, query: str, answer: str, *, conn: Any = None) -> int

Publish one exchange iff the conversation is still at revision.

Pass conn to join a caller's open transaction; the caller then owns the commit. Returns the new revision.

Source code in src/symfonic/platform/conversation_pg.py

async def append_exchange(
    self,
    principal: Any,
    session_id: str,
    revision: int,
    query: str,
    answer: str,
    *,
    conn: Any = None,
) -> int:
    """Publish one exchange iff the conversation is still at ``revision``.

    Pass ``conn`` to join a caller's open transaction; the caller then owns
    the commit. Returns the new revision.
    """
    owner = await self._open(principal)
    if conn is None:
        async with self._pool.acquire() as own, own.transaction():
            return await self._append(own, owner, session_id, revision, query, answer)
    return await self._append(conn, owner, session_id, revision, query, answer)

authorize `async` ¶

authorize(principal: Any, session_id: str) -> None

Refuse exactly as transcript does, reading no message row.

Source code in src/symfonic/platform/conversation_pg.py

async def authorize(self, principal: Any, session_id: str) -> None:
    """Refuse exactly as ``transcript`` does, reading no message row."""
    owner = await self._open(principal)
    async with self._pool.acquire() as conn:
        await self._owned(conn, owner, session_id)

create `async` ¶

create(principal: Any) -> str

Issue a new native-family session after verification.

Source code in src/symfonic/platform/conversation_pg.py

async def create(self, principal: Any) -> str:
    """Issue a new native-family session after verification."""
    owner = await self._open(principal)
    session_id = uuid.uuid4().hex
    async with self._pool.acquire() as conn:
        created = await conn.fetchval(
            f"INSERT INTO {self._tables.conversations} (tenant_id, sub_tenant_id, "
            "namespace, session_id, principal_id, family, revision, turns, created_at, "
            "updated_at) VALUES ($1, $2, $3, $4, $5, $6, 0, 0, clock_timestamp(), "
            "clock_timestamp()) ON CONFLICT DO NOTHING RETURNING session_id",
            *session_key(owner, session_id),
            owner.principal_id,
            NATIVE_FAMILY,
        )
    if created is None:
        raise ConversationConflict("session issuance collision")
    return session_id

discard_empty `async` ¶

discard_empty(principal: Any, session_id: str) -> None

Remove a session only if nothing was published and no live pause is pending.

A pending pause past its signed expiry no longer pins the session: its token cannot be redeemed, so nothing can publish to it.

Source code in src/symfonic/platform/conversation_pg.py

async def discard_empty(self, principal: Any, session_id: str) -> None:
    """Remove a session only if nothing was published and no live pause is pending.

    A pending pause past its signed expiry no longer pins the session: its
    token cannot be redeemed, so nothing can publish to it.
    """
    owner = await self._open(principal)
    async with self._pool.acquire() as conn, conn.transaction():
        row = await self._owned(conn, owner, session_id, lock=True)
        if row["turns"] != 0:
            return
        await conn.execute(
            f"DELETE FROM {self._tables.conversations} c WHERE {_OWNED} "
            f"AND NOT EXISTS (SELECT 1 FROM {self._tables.pauses} p WHERE "
            "p.tenant_id=c.tenant_id AND p.sub_tenant_id=c.sub_tenant_id AND "
            "p.namespace=c.namespace AND p.session_id=c.session_id AND p.state='pending' "
            "AND p.expires_at >= extract(epoch FROM clock_timestamp()))",
            *session_key(owner, session_id),
            owner.principal_id,
        )

list_sessions `async` ¶

list_sessions(principal: Any) -> list[dict[str, Any]]

List only this verified principal's native sessions.

Source code in src/symfonic/platform/conversation_pg.py

async def list_sessions(self, principal: Any) -> list[dict[str, Any]]:
    """List only this verified principal's native sessions."""
    owner = await self._open(principal)
    async with self._pool.acquire() as conn:
        return await owner_sessions(conn, self._tables, owner)

prepare `async` ¶

prepare(principal: Any, session_id: str, query: str) -> RevisionedConversation

Read authorized history; publish only against the revision read here.

Source code in src/symfonic/platform/conversation_pg.py

async def prepare(self, principal: Any, session_id: str, query: str) -> RevisionedConversation:
    """Read authorized history; publish only against the revision read here."""
    from symfonic import Message

    revision, rows = await self.snapshot(principal, session_id)
    history = tuple(Message(role=row["role"], content=row["content"]) for row in rows)
    lock = asyncio.Lock()
    published = False

    async def publish(answer: str) -> None:
        nonlocal published
        if not isinstance(answer, str) or not answer.strip():
            raise ValueError("empty conversation answer")
        async with lock:
            if published:
                return
            await self.append_exchange(principal, session_id, revision, query, answer)
            published = True

    return RevisionedConversation(history=history, publish=publish, revision=revision)

snapshot `async` ¶

snapshot(principal: Any, session_id: str) -> tuple[int, Any]

One consistent owned read of the revision and its messages.

Source code in src/symfonic/platform/conversation_pg.py

async def snapshot(self, principal: Any, session_id: str) -> tuple[int, Any]:
    """One consistent owned read of the revision and its messages."""
    owner = await self._open(principal)
    async with (
        self._pool.acquire() as conn,
        conn.transaction(
            isolation="repeatable_read",
            readonly=True,
        ),
    ):
        row = await self._owned(conn, owner, session_id)
        return int(row["revision"]), await message_rows(conn, self._tables, owner, session_id)

transcript `async` ¶

transcript(principal: Any, session_id: str) -> list[dict[str, Any]]

Authorize and read in one snapshot.

Source code in src/symfonic/platform/conversation_pg.py

async def transcript(self, principal: Any, session_id: str) -> list[dict[str, Any]]:
    """Authorize and read in one snapshot."""
    _, rows = await self.snapshot(principal, session_id)
    return transcript_rows(rows)

transcript_reader ¶

transcript_reader() -> PostgresTranscriptReader

The reader the selected root composes; there is no legacy facade here.

Source code in src/symfonic/platform/conversation_pg.py

def transcript_reader(self) -> PostgresTranscriptReader:
    """The reader the selected root composes; there is no legacy facade here."""
    if self._reader is not None:
        raise ValueError("conversation authority already has a transcript reader")
    self._reader = PostgresTranscriptReader(self._pool, self._tables, self.astart)
    return self._reader

PostgresConversationUnavailable ¶

Bases: SymfonicError

The PostgreSQL conversation store is not installed, reachable, or open.

PostgresTranscriptReader ¶

PostgresTranscriptReader(pool: Any, tables: ConversationTables, start: Callable[[], Awaitable[None]])

The selected root's transcript-reader slot, read from the same tables.

Like the compatibility facade's reader it is keyed by scope and session; principal ownership is the authority's transcript check, made first.

Source code in src/symfonic/platform/conversation_pg_reader.py

def __init__(
    self, pool: Any, tables: ConversationTables, start: Callable[[], Awaitable[None]]
) -> None:
    self._pool, self._tables, self._start = pool, tables, start

RevisionedConversation `dataclass` ¶

RevisionedConversation(history: tuple[Any, ...], publish: Callable[[str], Awaitable[None]], revision: int = -1)

Bases: PreparedConversation

Prepared history plus the revision its publication is checked against.

symfonic.platform.conversation_pg¶

conversation_pg ¶

PostgresConversationAuthority ¶

pauses property ¶

append_exchange async ¶

authorize async ¶

create async ¶

discard_empty async ¶

list_sessions async ¶

prepare async ¶

snapshot async ¶

transcript async ¶

transcript_reader ¶

PostgresConversationUnavailable ¶

PostgresTranscriptReader ¶

RevisionedConversation dataclass ¶

`symfonic.platform.conversation_pg`¶

pauses `property` ¶

append_exchange `async` ¶

authorize `async` ¶

create `async` ¶

discard_empty `async` ¶

list_sessions `async` ¶

prepare `async` ¶

snapshot `async` ¶

transcript `async` ¶

RevisionedConversation `dataclass` ¶