symfonic.memory.backends.postgres_graph_runtime

postgres_graph_runtime

Opt-in graph runtime without bootstrap DDL; no deployment or index swap.

Import the adapter through the public backends facade. Owner bootstrap remains PostgresGraphBackend; this adapter only validates and serves an already installed graph schema. The graph role contract does not cover vectors, journals, leases or app tables. It covers the CORE-199 erase job table when one exists: SELECT only, and nothing on its receipts, which only the separate eraser principal writes. And the CORE-199 memory erase generations when installed: read and advance, never DELETE or TRUNCATE; the table's trigger refuses a lowering UPDATE.

GraphRuntimeNotReady

Bases: RuntimeError

The graph runtime role or installed schema violates the contract.

PostgresGraphRuntime

PostgresGraphRuntime(pool: Any, *, owner_login: str, edge_upsert_mode: str = 'tenant_guarded', eraser: Any = None)

Bases: PostgresGraphBackend

Explicit nonowner graph adapter; ensure_schema performs read-only checks.

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

def __init__(
    self,
    pool: Any,
    *,
    owner_login: str,
    edge_upsert_mode: str = "tenant_guarded",
    eraser: Any = None,
) -> None:
    super().__init__(pool, edge_upsert_mode=edge_upsert_mode, eraser=eraser)
    self._owner_login = owner_login

attest_eraser async

attest_eraser() -> Any

The runtime role contract, then the eraser's proof for this runtime.

One runtime connection proves the role contract and resolves the session_user and memory_nodes the eraser must match (CORE-194). Read-only. Any refusal withdraws the standing attestation.

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

async def attest_eraser(self) -> Any:
    """The runtime role contract, then the eraser's proof for this runtime.

    One runtime connection proves the role contract and resolves the
    ``session_user`` and ``memory_nodes`` the eraser must match
    (CORE-194). Read-only. Any refusal withdraws the standing attestation.
    """
    from symfonic.memory.backends.postgres_native_eraser import runtime_identity

    if self.eraser is None:
        raise MemoryContractError("no native eraser is composed (CORE-199)")
    self._runtime_identity = None
    connection = await self.pool.acquire()
    try:
        await require_runtime(connection, self._edge_upsert_mode, self._owner_login)
        identity = await runtime_identity(_Pinned(connection))
    finally:
        await self.pool.release(connection)
    attestation = await self.eraser.attest(runtime_login=identity[0], binding=identity[1])
    self._runtime_identity = identity
    return attestation

ensure_schema async

ensure_schema() -> None

Read-only startup; with an eraser composed, also attest_eraser.

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

async def ensure_schema(self) -> None:
    """Read-only startup; with an eraser composed, also ``attest_eraser``."""
    if self.eraser is not None:
        await self.attest_eraser()
        return
    connection = await self.pool.acquire()
    try:
        await require_runtime(connection, self._edge_upsert_mode, self._owner_login)
    finally:
        await self.pool.release(connection)

require_runtime async

require_runtime(connection: Any, mode: str, owner_login: str) -> None

Reject owner, elevated or role-switch-capable logins before startup.

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

async def require_runtime(connection: Any, mode: str, owner_login: str) -> None:
    """Reject owner, elevated or role-switch-capable logins before startup."""
    role = await connection.fetchrow(
        "SELECT oid, session_user=current_user AS direct_login, "
        "rolcanlogin, rolsuper, rolcreatedb, rolcreaterole, "
        "rolreplication, rolbypassrls FROM pg_roles WHERE rolname=current_user"
    )
    if (
        not role
        or not role["direct_login"]
        or not role["rolcanlogin"]
        or any(
            role[key]
            for key in (
                "rolsuper",
                "rolcreatedb",
                "rolcreaterole",
                "rolreplication",
                "rolbypassrls",
            )
        )
    ):
        raise GraphRuntimeNotReady("graph runtime requires an unprivileged login")
    if await connection.fetchval(
        "SELECT EXISTS(SELECT 1 FROM pg_auth_members WHERE member=$1)", role["oid"]
    ):
        raise GraphRuntimeNotReady("graph runtime must have no role memberships")
    if await connection.fetchval(
        "SELECT datdba=$1 FROM pg_database WHERE datname=current_database()", role["oid"]
    ):
        raise GraphRuntimeNotReady("graph runtime must not own the database")
    namespace = None
    for table in TABLES:
        row = await connection.fetchrow(
            "SELECT c.relkind::text AS kind, c.relnamespace AS namespace, "
            "pg_get_userbyid(c.relowner)=$3 AND n.nspowner=c.relowner AS expected_owner, "
            "c.relowner=$2 AS owns_table, "
            "n.nspowner=$2 AS owns_schema, "
            "has_schema_privilege(current_user,n.oid,'CREATE') AS can_create, "
            "has_schema_privilege(current_user,n.oid,'USAGE') AS can_use "
            "FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace "
            "WHERE c.oid=to_regclass($1)",
            table,
            role["oid"],
            owner_login,
        )
        if (
            not row
            or row["kind"] != "r"
            or any(row[key] for key in ("owns_table", "owns_schema", "can_create"))
            or not row["can_use"]
            or not row["expected_owner"]
        ):
            raise GraphRuntimeNotReady("graph runtime requires owner-installed tables")
        if namespace is not None and namespace != row["namespace"]:
            raise GraphRuntimeNotReady("graph runtime tables must share one schema")
        namespace = row["namespace"]
        for privilege in ("SELECT", "INSERT", "UPDATE", "DELETE"):
            if not await connection.fetchval(
                "SELECT has_table_privilege(current_user,$1,$2)", table, privilege
            ):
                raise GraphRuntimeNotReady("graph runtime is missing a graph DML grant")
        for privilege in (
            "TRUNCATE",
            "TRIGGER",
            "REFERENCES",
            "SELECT WITH GRANT OPTION",
            "INSERT WITH GRANT OPTION",
            "UPDATE WITH GRANT OPTION",
            "DELETE WITH GRANT OPTION",
        ):
            # Column-level REFERENCES or grant options escape has_table_privilege.
            if await connection.fetchval(
                "SELECT CASE WHEN split_part($2,' ',1) IN ('SELECT','INSERT','UPDATE',"
                "'REFERENCES') THEN has_any_column_privilege(current_user,$1,$2) "
                "ELSE has_table_privilege(current_user,$1,$2) END",
                table,
                privilege,
            ):
                raise GraphRuntimeNotReady("graph runtime has excess graph privileges")
    await _require_jobs_read_only(connection)
    # Validate installed columns without issuing CREATE/ALTER, including upgrades.
    for table in TABLES:
        columns = {
            "memory_nodes": "id,tenant_id,layer,label,properties,embedding,created_at,"
            "updated_at,access_count,importance,scope_path",
            "memory_edges": "id,tenant_id,source,target,relationship,weight,created_at,"
            "uses,provenance,source_episodic_ids,scope_path",
            "memory_pending_claims": "token,tenant_id,record_id,scope_path,state",
            "memory_workset_progress": "scope_path,cycle,revision,payload",
        }[table]
        await connection.fetch(f"SELECT {columns} FROM {table} LIMIT 0")
    if mode == "scoped":
        from symfonic.memory.backends.postgres_edge_scoped_writer import require_scoped_edge_index

        await require_scoped_edge_index(connection)
    else:
        from symfonic.memory.backends.postgres_edge_index_preflight import require_direct_edge_index

        await require_direct_edge_index(connection, mode)