symfonic.platform.conversation_pg_schema

conversation_pg_schema

Explicit PostgreSQL schema for the shared native HTTP conversation authority.

install_conversation_schema is a migration step run by the schema owner, never by a request or by authority start-up. The runtime authority only attests that the installed tables carry this version's marker and that its login holds the row privileges it needs; a missing or foreign table fails closed. These tables are separate from the graph-memory conversation log (native_conversation_*) and share nothing with it.

ConversationTables dataclass

ConversationTables(schema: str)

Fully qualified table names under one validated schema identifier.

PostgresConversationUnavailable

Bases: SymfonicError

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

attest_conversation_schema async

attest_conversation_schema(conn: Any, tables: ConversationTables) -> None

Refuse unless every table is installed at this version and usable by this login.

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

async def attest_conversation_schema(conn: Any, tables: ConversationTables) -> None:
    """Refuse unless every table is installed at this version and usable by this login."""
    if not await conn.fetchval(
        "SELECT has_schema_privilege(current_user, $1, 'USAGE')", tables.schema
    ):
        raise PostgresConversationUnavailable(
            "shared conversation login lacks USAGE on the conversation schema; run "
            "install_conversation_schema for this runtime role",
        )
    for table in tables.all():
        marker = await conn.fetchval(
            "SELECT obj_description(to_regclass($1)::oid, 'pg_class')",
            table,
        )
        if marker != MARKER:
            raise PostgresConversationUnavailable(
                "shared conversation schema is not installed; run "
                "install_conversation_schema as the schema owner",
            )
        for privilege in ("SELECT", "INSERT", "UPDATE", "DELETE"):
            if not await conn.fetchval(
                "SELECT has_table_privilege(current_user, $1, $2)",
                table,
                privilege,
            ):
                raise PostgresConversationUnavailable(
                    "shared conversation login lacks a required table privilege",
                )

attest_pool async

attest_pool(pool: Any, tables: ConversationTables) -> None

Attest through one pooled connection; a driver refusal is Unavailable.

A login without schema USAGE or a table privilege, an unreachable server, or a refused connection must reach the caller as the typed 503 refusal, not as a raw driver error at the first request.

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

async def attest_pool(pool: Any, tables: ConversationTables) -> None:
    """Attest through one pooled connection; a driver refusal is ``Unavailable``.

    A login without schema ``USAGE`` or a table privilege, an unreachable
    server, or a refused connection must reach the caller as the typed
    503 refusal, not as a raw driver error at the first request.
    """
    try:
        async with pool.acquire() as conn:
            await attest_conversation_schema(conn, tables)
    except PostgresConversationUnavailable:
        raise
    except Exception as exc:
        if not (isinstance(exc, OSError) or type(exc).__module__.startswith("asyncpg")):
            raise
        raise PostgresConversationUnavailable(
            f"shared conversation store refused start-up ({type(exc).__name__})"
        ) from exc

install_conversation_schema async

install_conversation_schema(conn: Any, *, schema: str, runtime_role: str) -> None

Create the tables in an existing schema and grant the runtime login row access.

Run by the schema owner as an explicit migration. Idempotent: an existing table keeps its rows. The runtime login gets USAGE on the schema (it cannot name the tables without it) and row privileges on the tables; PUBLIC gets nothing on the tables.

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

async def install_conversation_schema(conn: Any, *, schema: str, runtime_role: str) -> None:
    """Create the tables in an existing schema and grant the runtime login row access.

    Run by the schema owner as an explicit migration. Idempotent: an existing
    table keeps its rows. The runtime login gets ``USAGE`` on the schema (it
    cannot name the tables without it) and row privileges on the tables;
    ``PUBLIC`` gets nothing on the tables.
    """
    tables = ConversationTables(schema)
    role = _identifier(runtime_role, "runtime role")
    async with conn.transaction():
        await conn.execute(
            "SELECT pg_advisory_xact_lock(hashtextextended('symfonic-http-conversation-install',0))"
        )
        for statement in _ddl(tables):
            await conn.execute(statement)
        await conn.execute(f'GRANT USAGE ON SCHEMA "{tables.schema}" TO "{role}"')
        for table in tables.all():
            await conn.execute(f"COMMENT ON TABLE {table} IS '{MARKER}'")
            await conn.execute(f"REVOKE ALL ON {table} FROM PUBLIC")
            await conn.execute(f'GRANT {_RUNTIME_PRIVILEGES} ON {table} TO "{role}"')