symfonic.capabilities.memory.graph_native_admission

graph_native_admission

Fail-closed admission for direct kind-aware graph memory operations.

admit_raw_pg_mutation

admit_raw_pg_mutation(scope: MemoryScope, store: Any, verb: str | None) -> bool

Exact graph-only PostgreSQL v1 with an attested CORE-199 eraser (CORE-194).

verb None is composition, which needs the whole lifecycle. False hands the decision to require_native_backends, which refuses PostgreSQL: a writable v1 scope whose only forget is the bounded inline delete could hold rows that no forget can erase.

Source code in src/symfonic/capabilities/memory/graph_native_admission.py

def admit_raw_pg_mutation(scope: MemoryScope, store: Any, verb: str | None) -> bool:
    """Exact graph-only PostgreSQL v1 with an attested CORE-199 eraser (CORE-194).

    ``verb`` None is composition, which needs the whole lifecycle. False hands
    the decision to ``require_native_backends``, which refuses PostgreSQL: a
    writable v1 scope whose only forget is the bounded inline delete could
    hold rows that no forget can erase.
    """
    if verb is not None and verb not in _RAW_PG_VERBS:
        return False
    if not _raw_pg_shape(scope, store) or _eraser_problem(store._graph) is not None:
        return False
    from symfonic.memory.storage_scope import NativeStorageScope

    NativeStorageScope.from_scope(scope)
    return True

exact_pg_graph

exact_pg_graph(graph: Any) -> bool

Exactly the owner backend or the nonowner runtime; never a subclass.

The runtime only replaces startup with read-only checks, so both serve the same lifecycle SQL (CORE-194). The module test keeps the optional PostgreSQL stack off ordinary in-memory paths.

Source code in src/symfonic/capabilities/memory/graph_native_admission.py

def exact_pg_graph(graph: Any) -> bool:
    """Exactly the owner backend or the nonowner runtime; never a subclass.

    The runtime only replaces startup with read-only checks, so both serve
    the same lifecycle SQL (CORE-194). The module test keeps the optional
    PostgreSQL stack off ordinary in-memory paths.
    """
    if type(graph).__module__ not in PG_GRAPH_MODULES:
        return False
    from symfonic.memory.backends.postgres_graph import PostgresGraphBackend
    from symfonic.memory.backends.postgres_graph_runtime import PostgresGraphRuntime

    return type(graph) in (PostgresGraphBackend, PostgresGraphRuntime)

native_quick_writer_scope

native_quick_writer_scope(graph: Any, window: Any, scope: MemoryScope) -> Any

Validate the writing graph before reading the native QUICK window.

Source code in src/symfonic/capabilities/memory/graph_native_admission.py

def native_quick_writer_scope(graph: Any, window: Any, scope: MemoryScope) -> Any:
    """Validate the writing graph before reading the native QUICK window."""
    if not scope.identity_version:
        from symfonic.capabilities.memory.graph_rows import tenant_scope

        return tenant_scope(scope)
    require_native_quick_graph(graph)
    if graph is not getattr(window, "_graph", None):
        raise MemoryContractError("native QUICK requires the same journalled graph")
    from symfonic.memory.storage_scope import NativeStorageScope

    return NativeStorageScope.from_scope(scope)

require_composition

require_composition(factory: Any, scope: MemoryScope) -> None

Refuse an unadmitted v1 mutation shape before any Agent can exist.

Source code in src/symfonic/capabilities/memory/graph_native_admission.py

def require_composition(factory: Any, scope: MemoryScope) -> None:
    """Refuse an unadmitted v1 mutation shape before any Agent can exist."""
    from symfonic.capabilities.memory.graph_store import GraphBackedHms

    store = factory._store
    if not isinstance(store, GraphBackedHms):
        return
    try:
        store._check_mutation(scope)
    except MemoryContractError as error:
        problem = _raw_pg_shape(scope, store) and _eraser_problem(store._graph)
        if problem:
            raise MemoryContractError(f"{error}; {problem}") from error
        raise
    if admit_raw_pg_mutation(scope, store, None):
        for name in _RAW_PG_UNPROVEN:
            if getattr(factory, name) is not None:
                raise MemoryContractError(
                    f"graph-only PostgreSQL v1 memory does not admit {name[1:]}: "
                    "its forget would not reach that state (CORE-194)")
        conversation = factory._conversation
        if conversation is not None:
            from symfonic.capabilities.memory.turn_generation import fenced

            if not fenced(conversation) or conversation.scope.path != scope.path:
                raise MemoryContractError(_UNFENCED_CONVERSATION)
            from symfonic.capabilities.memory.admin_erase import ERASER_POOL

            log = conversation.store
            if log.pool is store._graph.eraser.pool:
                raise MemoryContractError(ERASER_POOL)
            # CORE-171: served history checks out only behind an anchor: the
            # log's own when configured (any database), else memory's.
            if log.proof is None:
                store._graph.eraser.guarding(log.pool)

require_native_backends

require_native_backends(scope: MemoryScope, graph: Any, vectors: Any) -> None

Only concrete backends that explicitly prove v1 storage may serve it.

Source code in src/symfonic/capabilities/memory/graph_native_admission.py

def require_native_backends(scope: MemoryScope, graph: Any, vectors: Any) -> None:
    """Only concrete backends that explicitly prove v1 storage may serve it."""
    if not scope.identity_version:
        return
    if type(graph).__module__ == "symfonic.memory.backends.mongo_native_receipt":
        from symfonic.capabilities.memory.receipt_read_dispatch import read_snapshot_for

        read_snapshot_for(graph, vectors)
        return
    from symfonic.capabilities.memory.journal import JournalledGraph, current_journal

    if type(graph) is JournalledGraph:
        # The cycle overlay has not established kind-aware journal identity.
        from symfonic.memory.backends.in_memory import InMemoryGraphBackend

        durable = graph.durable
        durable_type = type(durable)
        durable_exact = durable_type is InMemoryGraphBackend
        if (not durable_exact
                and durable_type.__module__ == "symfonic.memory.backends.postgres_graph"):
            # Ordinary in-memory admission must not load the optional PG stack.
            from symfonic.memory.backends.postgres_graph import PostgresGraphBackend

            durable_exact = durable_type is PostgresGraphBackend
        graph_proven = (
            current_journal() is None
            and durable_exact
            and durable_type.__dict__.get("supports_native_storage_scope") is True
        )
    elif isinstance(graph, JournalledGraph):
        # A subclass flag must not bypass the exact wrapper's durable and
        # active-journal checks.
        graph_proven = False
    else:
        graph_proven = _declares_native_storage(graph)
    vector_proven = (
        vectors is None
        or _declares_native_storage(vectors)
    )
    if not graph_proven or not vector_proven:
        raise MemoryContractError(_REFUSED)

require_native_quick_graph

require_native_quick_graph(graph: Any) -> None

Admit only the in-memory cycle's fenced, journalled reference graph.

Source code in src/symfonic/capabilities/memory/graph_native_admission.py

def require_native_quick_graph(graph: Any) -> None:
    """Admit only the in-memory cycle's fenced, journalled reference graph."""
    from symfonic.capabilities.memory.fencing import FencedGraph
    from symfonic.capabilities.memory.journal import JournalledGraph, current_journal
    from symfonic.memory.backends.in_memory import InMemoryGraphBackend
    from symfonic.memory.graph.store import GraphMemoryStore

    if type(graph) is not FencedGraph or current_journal() is None:
        raise MemoryContractError("native QUICK requires a fenced in-memory journal cycle")
    store = graph.unfenced
    if type(store) is not GraphMemoryStore or type(store.backend) is not JournalledGraph:
        raise MemoryContractError("native QUICK requires an exact journalled graph store")
    durable = store.backend.durable
    if (type(durable) is InMemoryGraphBackend
            and type(durable).__dict__.get("supports_native_storage_scope") is True):
        return
    from symfonic.memory.backends.postgres_receipt_views import (
        PostgresReceiptGraph,
        receipt_pair_domain,
    )

    if type(durable) is PostgresReceiptGraph:
        receipt_pair_domain(durable, None)
        return
    raise MemoryContractError("native QUICK requires exact InMemoryGraphBackend or receipt graph")

require_native_read

require_native_read(scope: MemoryScope, store: Any) -> None

Raw graph-only PostgreSQL proves v1 reads; mutation is admitted apart.

Reads need no eraser. Mutation goes through admit_raw_pg_mutation, which also requires an attested CORE-199 eraser (CORE-194). Exact HMS/backend types exclude custom implementations and cycle overlays. Keep the optional PostgreSQL import off ordinary in-memory read paths.

Source code in src/symfonic/capabilities/memory/graph_native_admission.py

def require_native_read(scope: MemoryScope, store: Any) -> None:
    """Raw graph-only PostgreSQL proves v1 reads; mutation is admitted apart.

    Reads need no eraser. Mutation goes through ``admit_raw_pg_mutation``,
    which also requires an attested CORE-199 eraser (CORE-194).
    Exact HMS/backend types exclude custom implementations and cycle overlays.
    Keep the optional PostgreSQL import off ordinary in-memory read paths.
    """
    graph = store._graph
    if (type(scope) is MemoryScope and scope.identity_version == 1
            and store._vectors is None and exact_pg_graph(graph)):
        from symfonic.capabilities.memory.graph_store import GraphBackedHms

        if type(store) is GraphBackedHms:
            from symfonic.memory.storage_scope import NativeStorageScope

            NativeStorageScope.from_scope(scope)
            return
    require_native_backends(scope, graph, store._vectors)