Projection Backend Contract Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Extract a backend-neutral projection contract and backend factory so Neo4j remains the default while pgGraph can be evaluated behind the same retrieval and projection surface.

Architecture: Eventloom remains the source of truth. GraphStore becomes the first concrete implementation of a typed ProjectionStore contract, and all high-level construction paths use a backend factory instead of instantiating Neo4j directly. The pgGraph path is explicitly experimental and unavailable until a real adapter passes the same contract and benchmark gates.

Tech Stack: Python 3.11+, typing Protocol, Pydantic settings, existing Neo4j GraphStore, pytest/ruff/mypy, benchmark guardrail CLI.

---

File Structure

Task 1: Typed Projection Contract

Files:

Add tests/test_projection.py:

from __future__ import annotations

from typing import assert_type

from zaxy.extract import ExtractionResult
from zaxy.graph import GraphEdge, GraphEntity, GraphEventProjectionStatus, GraphInferredEdgeStatus, SearchResult
from zaxy.projection import ProjectionStore


class FakeProjectionStore:
    async def connect(self) -> None:
        pass

    async def close(self) -> None:
        pass

    async def init_schema(self) -> None:
        pass

    async def upsert_extraction(self, result: ExtractionResult, session_id: str = "default") -> None:
        pass

    async def invalidate_entity(
        self,
        name: str,
        entity_type: str,
        invalid_at: str,
        session_id: str = "default",
    ) -> None:
        pass

    async def search_exact(
        self,
        name: str,
        entity_type: str | None = None,
        temporal_point: str | None = None,
        session_id: str = "default",
    ) -> list[GraphEntity]:
        return []

    async def search_keyword(
        self,
        query: str,
        limit: int = 10,
        temporal_point: str | None = None,
        session_id: str = "default",
    ) -> list[SearchResult]:
        return []

    async def search_traversal(
        self,
        start_name: str,
        relation_type: str | None = None,
        depth: int = 2,
        temporal_point: str | None = None,
        session_id: str = "default",
    ) -> list[GraphEntity]:
        return []

    async def search_vector(
        self,
        embedding: list[float],
        limit: int = 10,
        temporal_point: str | None = None,
        session_id: str = "default",
    ) -> list[SearchResult]:
        return []

    async def inspect_event_projection_status(
        self,
        session_id: str,
        *,
        eventloom_latest_seq: int | None = None,
        eventloom_latest_hash: str | None = None,
    ) -> GraphEventProjectionStatus:
        raise NotImplementedError

    async def inspect_inferred_edge_status(
        self,
        session_id: str,
        *,
        limit: int = 10,
    ) -> GraphInferredEdgeStatus:
        raise NotImplementedError

    async def get_overview_graph(self, session_id: str, *, limit: int = 100) -> tuple[list[GraphEntity], list[GraphEdge]]:
        return [], []


def test_fake_projection_store_satisfies_contract() -> None:
    store: ProjectionStore = FakeProjectionStore()
    assert store is not None


def test_projection_store_search_types_are_concrete(store: ProjectionStore) -> None:
    assert_type(store, ProjectionStore)

Run:

PYTHONPATH=src pytest tests/test_projection.py -q --no-cov
PYTHONPATH=src mypy tests/test_projection.py

Expected: pytest may pass structurally, but mypy fails because ProjectionStore still returns object and lacks the extended methods.

Update src/zaxy/projection.py:

from __future__ import annotations

from typing import Protocol

from zaxy.extract import ExtractionResult
from zaxy.graph import GraphEdge, GraphEntity, GraphEventProjectionStatus, GraphInferredEdgeStatus, SearchResult


class ProjectionStore(Protocol):
    """Backend contract for projecting Eventloom facts into a queryable memory index."""

    async def connect(self) -> None: ...
    async def close(self) -> None: ...
    async def init_schema(self) -> None: ...

    async def upsert_extraction(self, result: ExtractionResult, session_id: str = "default") -> None: ...

    async def invalidate_entity(
        self,
        name: str,
        entity_type: str,
        invalid_at: str,
        session_id: str = "default",
    ) -> None: ...

    async def search_exact(
        self,
        name: str,
        entity_type: str | None = None,
        temporal_point: str | None = None,
        session_id: str = "default",
    ) -> list[GraphEntity]: ...

    async def search_keyword(
        self,
        query: str,
        limit: int = 10,
        temporal_point: str | None = None,
        session_id: str = "default",
    ) -> list[SearchResult]: ...

    async def search_traversal(
        self,
        start_name: str,
        relation_type: str | None = None,
        depth: int = 2,
        temporal_point: str | None = None,
        session_id: str = "default",
    ) -> list[GraphEntity]: ...

    async def search_vector(
        self,
        embedding: list[float],
        limit: int = 10,
        temporal_point: str | None = None,
        session_id: str = "default",
    ) -> list[SearchResult]: ...

    async def inspect_event_projection_status(
        self,
        session_id: str,
        *,
        eventloom_latest_seq: int | None = None,
        eventloom_latest_hash: str | None = None,
    ) -> GraphEventProjectionStatus: ...

    async def inspect_inferred_edge_status(
        self,
        session_id: str,
        *,
        limit: int = 10,
    ) -> GraphInferredEdgeStatus: ...

    async def get_overview_graph(
        self,
        session_id: str,
        *,
        limit: int = 100,
    ) -> tuple[list[GraphEntity], list[GraphEdge]]: ...

Run:

PYTHONPATH=src pytest tests/test_projection.py tests/test_graph.py -q --no-cov -k "projection_store or graph_store_satisfies"
PYTHONPATH=src mypy src/zaxy/projection.py tests/test_projection.py

Expected: tests pass and mypy succeeds.

git add src/zaxy/projection.py tests/test_projection.py
git commit -m "refactor: type projection store contract"

Task 2: Backend Factory And Setting

Files:

Append to tests/test_projection.py:

import pytest

from zaxy.graph import GraphStore
from zaxy.projection_backends import ProjectionBackendConfig, build_projection_store


def test_build_projection_store_defaults_to_neo4j() -> None:
    store = build_projection_store(
        ProjectionBackendConfig(
            backend="neo4j",
            neo4j_uri="bolt://localhost:7687",
            neo4j_user="neo4j",
            neo4j_password="testpassword",
            neo4j_ca_cert=None,
            neo4j_trust_all=False,
        )
    )

    assert isinstance(store, GraphStore)


def test_build_projection_store_rejects_pggraph_until_adapter_exists() -> None:
    with pytest.raises(NotImplementedError, match="pgGraph backend is experimental"):
        build_projection_store(
            ProjectionBackendConfig(
                backend="pggraph",
                neo4j_uri="bolt://localhost:7687",
                neo4j_user="neo4j",
                neo4j_password="testpassword",
                neo4j_ca_cert=None,
                neo4j_trust_all=False,
            )
        )

Add to tests/test_config.py:

def test_projection_backend_defaults_to_neo4j(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("PROJECTION_BACKEND", raising=False)
    get_settings.cache_clear()

    assert get_settings().projection_backend == "neo4j"

Run:

PYTHONPATH=src pytest tests/test_projection.py tests/test_config.py -q --no-cov -k "projection_backend or build_projection_store"

Expected: fails because projection_backends.py and projection_backend setting do not exist.

Create src/zaxy/projection_backends.py:

"""Projection backend construction.

Neo4j is the production default. pgGraph is exposed only as an explicit
experimental target until an adapter passes the same contract and benchmarks.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Literal

from zaxy.graph import GraphStore
from zaxy.projection import ProjectionStore

ProjectionBackendName = Literal["neo4j", "pggraph"]


@dataclass(frozen=True)
class ProjectionBackendConfig:
    backend: str
    neo4j_uri: str
    neo4j_user: str
    neo4j_password: str
    neo4j_ca_cert: str | None
    neo4j_trust_all: bool


def build_projection_store(config: ProjectionBackendConfig) -> ProjectionStore:
    backend = config.backend.casefold().strip()
    if backend == "neo4j":
        return GraphStore(
            config.neo4j_uri,
            config.neo4j_user,
            config.neo4j_password,
            ca_cert=config.neo4j_ca_cert,
            trust_all=config.neo4j_trust_all,
        )
    if backend == "pggraph":
        raise NotImplementedError(
            "pgGraph backend is experimental and has no adapter yet. "
            "Keep PROJECTION_BACKEND=neo4j until pgGraph passes the projection contract and benchmark gates."
        )
    raise ValueError("projection backend must be one of: neo4j, pggraph")

Add to Settings in src/zaxy/config.py:

projection_backend: str = Field(default="neo4j", validation_alias="PROJECTION_BACKEND")

Run:

PYTHONPATH=src pytest tests/test_projection.py tests/test_config.py -q --no-cov -k "projection_backend or build_projection_store"
ruff check src/zaxy/projection_backends.py src/zaxy/config.py tests/test_projection.py tests/test_config.py
PYTHONPATH=src mypy src/zaxy/projection_backends.py

Expected: tests pass, ruff clean, mypy succeeds.

git add src/zaxy/projection_backends.py src/zaxy/config.py tests/test_projection.py tests/test_config.py
git commit -m "feat: add projection backend factory"

Task 3: Route High-Level Construction Through The Factory

Files:

Add tests that patch zaxy.core.build_projection_store, zaxy.mcp_server.build_projection_store, and zaxy.live_benchmark.build_projection_store, then instantiate or run the existing constructor paths and assert the factory receives backend="neo4j" from settings.

Use this pattern in each file:

@patch("zaxy.core.build_projection_store")
def test_memory_fabric_constructs_projection_store_through_factory(mock_build: MagicMock, tmp_path: Path) -> None:
    mock_build.return_value = MagicMock()

    fabric = MemoryFabric(eventloom_path=str(tmp_path / ".eventloom"), tracer_disabled=True)

    assert fabric.graph is mock_build.return_value
    assert mock_build.call_args.args[0].backend == "neo4j"

Run:

PYTHONPATH=src pytest tests/test_core.py tests/test_mcp.py tests/test_live_benchmark.py -q --no-cov -k "projection_store_through_factory or projection_backend"

Expected: fails because construction still instantiates GraphStore directly.

Import and use:

from zaxy.projection_backends import ProjectionBackendConfig, build_projection_store

Build config from resolved settings:

ProjectionBackendConfig(
    backend=resolved_settings.projection_backend,
    neo4j_uri=neo4j_uri or resolved_settings.neo4j_uri,
    neo4j_user=neo4j_user or resolved_settings.neo4j_user,
    neo4j_password=neo4j_password or resolved_settings.neo4j_password,
    neo4j_ca_cert=neo4j_ca_cert if neo4j_ca_cert is not None else resolved_settings.neo4j_ca_cert,
    neo4j_trust_all=neo4j_trust_all if neo4j_trust_all is not None else resolved_settings.neo4j_trust_all,
)

Leave direct GraphStore construction in migration/schema commands that inspect Neo4j-only internals such as _driver, and document those as Neo4j-only until the pgGraph adapter has equivalent operations.

Run:

PYTHONPATH=src pytest tests/test_core.py tests/test_mcp.py tests/test_live_benchmark.py -q --no-cov -k "projection_store_through_factory or projection_backend"
PYTHONPATH=src pytest tests/test_core.py tests/test_mcp.py tests/test_query.py -q --no-cov
ruff check src/zaxy/core.py src/zaxy/mcp_server.py src/zaxy/live_benchmark.py src/zaxy/__main__.py
PYTHONPATH=src mypy src/zaxy/core.py src/zaxy/mcp_server.py src/zaxy/live_benchmark.py

Expected: tests pass, ruff clean, mypy succeeds.

git add src/zaxy/core.py src/zaxy/mcp_server.py src/zaxy/live_benchmark.py src/zaxy/__main__.py tests/test_core.py tests/test_mcp.py tests/test_live_benchmark.py
git commit -m "refactor: construct projection stores through backend factory"

Task 4: Documentation And Roadmap State

Files:

Add to tests/test_docs_site.py:

def test_pggraph_docs_keep_backend_experimental_and_contract_first() -> None:
    spec = Path("docs/superpowers/specs/2026-05-17-skill-memory-pggraph-evaluation-design.md").read_text(encoding="utf-8")
    agents = Path("AGENTS.md").read_text(encoding="utf-8")

    assert "PROJECTION_BACKEND=neo4j" in spec
    assert "pgGraph backend is experimental" in spec
    assert "Projection backend contract and Neo4j factory" in agents
    assert "Skill Memory procedural world-model layer" in agents

Run:

PYTHONPATH=src pytest tests/test_docs_site.py -q --no-cov -k pggraph_docs

Expected: fails because docs do not yet mention the factory state or updated roadmap completion.

Update AGENTS.md status:

- [x] Skill Memory procedural world-model layer with lifecycle events, checkout routing, MCP helper, docs, and full-set quality guardrail verification
- [x] Projection backend contract and Neo4j factory for pgGraph evaluation without changing the default backend

Move the Skill Memory next step out of the active list. Update pgGraph next step to:

Build the experimental pgGraph adapter behind `PROJECTION_BACKEND=pggraph` only after the backend-neutral Neo4j factory and contract tests are green.

Update the pgGraph spec with the current upstream posture from the docs checked on May 18, 2026:

As of May 18, 2026, pgGraph docs describe version 0.1.0, PostgreSQL 13-18 support, and alpha status for experimentation, demos, benchmarks, and early feedback. Zaxy therefore keeps `PROJECTION_BACKEND=neo4j` as the default and treats `PROJECTION_BACKEND=pggraph` as unavailable until an adapter passes the benchmark gates.

Run:

PYTHONPATH=src pytest tests/test_docs_site.py -q --no-cov

Expected: docs tests pass.

git add AGENTS.md docs/benchmarks.md docs/superpowers/specs/2026-05-17-skill-memory-pggraph-evaluation-design.md tests/test_docs_site.py
git commit -m "docs: update pggraph evaluation roadmap state"

Task 5: Final Verification

Files:

Run:

PYTHONPATH=src pytest tests/test_projection.py tests/test_config.py tests/test_core.py tests/test_mcp.py tests/test_query.py tests/test_docs_site.py -q --no-cov

Expected: PASS.

Run:

ruff check src/zaxy/projection.py src/zaxy/projection_backends.py src/zaxy/config.py src/zaxy/core.py src/zaxy/mcp_server.py src/zaxy/live_benchmark.py tests/test_projection.py tests/test_config.py tests/test_core.py tests/test_mcp.py tests/test_docs_site.py
PYTHONPATH=src mypy src/zaxy/projection.py src/zaxy/projection_backends.py src/zaxy/config.py src/zaxy/core.py src/zaxy/mcp_server.py src/zaxy/live_benchmark.py

Expected: ruff clean and mypy succeeds.

Run:

PYTHONPATH=src python -m zaxy benchmark-compare reports/benchmarks/longmemeval-500-hash/live-benchmark.json \
  --backend zaxy-checkout \
  --min-mean-score 0.626 \
  --min-answer-recall-at-5 0.608 \
  --min-recall-at-5 0.956 \
  --min-citation-coverage 1.0 \
  --max-p95-ms 15000 \
  --max-p99-ms 23000

Expected: PASS on the archived report. If a live run is performed, compare quality floors separately from noisy local latency and do not replace archived floors without an explicit new report.

Run:

rg -n "PROJECTION_BACKEND|pggraph|GraphStore\\(" src/zaxy tests docs AGENTS.md

Expected: pgGraph is documented as experimental, default backend remains Neo4j, and any remaining direct GraphStore( construction is either in the Neo4j adapter/factory, tests, or Neo4j-specific CLI/schema code.

git status --short

Expected: clean worktree. If verification caused edits, commit them with chore: finalize projection backend contract.

Self-Review