Orchestrata Ultimate › Plans › neo4j-seam-design

Operational-graph repository seam (backlog #163)

Source: plans/neo4j-seam-design.md at 8a2cb42ad · last changed 2026-07-27 · first committed 2026-07-08 · 118 lines.

Prose pending.

Plan text

Operational-graph repository seam (backlog #163)

Route all direct operational-Neo4j access through a repository/adapter seam mirroring semantic_graph's create_semantic_graph factory, so the operational store becomes swappable/droppable. Driver construction is already centralized (core/neo4j_factory, D12); this adds the usage seam. Evidence base: full driver-usage catalog of all 23+ touching files (session 224 Explore audit, 2026-07-08).

Goals / non-goals

• Goal: no subsystem file imports neo4j or calls create_neo4j / create_async_neo4j directly; they depend on a GraphRepository protocol obtained from one factory.
• Goal: ownership semantics (shared vs owned) made explicit — today ad-hoc _owns_driver flags are re-implemented in 4 files.
• Non-goal: touching semantic_graph/* (Memgraph, already behind its own seam) or changing any Cypher.
• Non-goal: dropping Neo4j. This seam is the prerequisite.

Package layout — core/graph/ (new)

Mirrors semantic_graph/'s shape. Lives in core because every subsystem already imports core, and the raw-driver factory (core/neo4j_factory) is already there — no new layering edges, no cycles.

core/graph/
  __init__.py        # re-export: protocols, config, factories, NO_AUTH
  interfaces.py      # SyncGraphRepository / AsyncGraphRepository / UnitOfWork protocols
  config.py          # GraphRepositoryConfig.from_env() (NEO4J_URI/USERNAME/PASSWORD/DATABASE)
  repository.py      # create_graph_repository() / create_async_graph_repository()
  adapters/
    __init__.py
    neo4j_sync.py    # Neo4jSyncRepository
    neo4j_async.py   # Neo4jAsyncRepository

Protocol surface (derived from the usage catalog — covers every real call)

class AsyncGraphRepository(Protocol):
    async def execute_read(self, cypher: str, params: Mapping[str, Any] | None = None,
                           *, database: str | None = None) -> list[dict[str, Any]]: ...
    async def execute_write(self, cypher: str, params: Mapping[str, Any] | None = None,
                            *, database: str | None = None) -> list[dict[str, Any]]: ...
    async def verify_connectivity(self) -> None: ...
    async def close(self) -> None: ...          # no-op when repository is shared

class SyncGraphRepository(Protocol):
    def execute_read(self, cypher: str, params: Mapping[str, Any] | None = None,
                     *, database: str | None = None) -> list[dict[str, Any]]: ...
    def execute_write(self, cypher: str, params: Mapping[str, Any] | None = None,
                      *, database: str | None = None) -> list[dict[str, Any]]: ...
    def unit_of_work(self, *, database: str | None = None) -> AbstractContextManager[UnitOfWork]: ...
    def verify_connectivity(self) -> None: ...
    def close(self) -> None: ...

class UnitOfWork(Protocol):                     # one transaction, N statements
    def run(self, cypher: str, params: Mapping[str, Any] | None = None) -> list[dict[str, Any]]: ...

Decisions locked by real usage:

• database= optional — only memory/cold_store, knowledge_graph/neo4j_client, session/kg_context, session/kg_sync/_base pass it; everyone else omits.
• Params: one params mapping (adapters normalize; callers today mix positional-dict and **kwargs).
• Return list[dict] — covers async for, .data(), comprehension styles.
• execute_write uses managed transactions (session.execute_write) in the adapter → free driver-level transient retry; execute_read uses session.execute_read. This upgrades the plain-session.run callers to retry-safe semantics without their code changing.
• unit_of_work (sync) replaces the multi-statement-per-transaction patterns: cxp_auditor (node+rel), pipeline_kg (node+link), neo4j_store txn functions, kg_context session-passing helpers.
• Ownership: adapters take driver=None (create + own) or an injected driver/shared=True (never close). One implementation, not 4 copies of _owns_driver.
• Escape hatch: adapters expose .driver (documented raw handle) for the exotic files below; the protocol does NOT include it — callers needing it type against the concrete adapter, making escapes greppable.

Migration order — one commit per phase, suite + Neo4j up for each

Status: COMPLETE (all phases landed + pushed to origin/master). phase 0 = 17e34088 (+ driver_kwargs edf93ab6), phase 1 = 460b6237, phase 2 = 7db35929, phase 3 = 64aa67eb, phase 4 = bed9e2f8, phase 5/5b = 62bc2ad1, phase 6 (D13 gate) = 219b83c3. The seam is enforced: scripts/architecture_checker.py D13 bans import neo4j and .session( outside the documented allowlist (_GRAPH_SEAM_ALLOWED); gate green at 0 violations over 653 files (session 225, 2026-07-11). The Neo4jSyncer._driver compat property flagged in phase 5b was removed.

knowledge_graph/neo4j_client.py (async, has its own retry loop) stays as-is initially — it is itself nearly a repository; phase 2+ may either retire it into Neo4jAsyncRepository or have it implement the protocol. Decide at phase 4 with the evidence from phases 1–3.

Test strategy

• New seam: real-Neo4j tests (no mocks per doctrine) — CRUD via both repos, database= routing, uow commit/rollback, ownership (shared repo close() is a no-op; owned closes), transient-retry behavior.
• Migrations: existing per-subsystem suites are the safety net (run per phase); add a regression test only where behavior changes (e.g. plain-run → managed-txn upgrade).

Related

• Plans
• Orchestrata Ultimate — project index
• Workflow
• Tasks