fix(neo4j): scope get_node's fallback MATCH to :Node so it seeks instead of scanning

context_intelligence_server/main.py

@@ -44,6 +44,7 @@
     build_bounded_neo4j_driver,
     count_untagged_nodes,
     ensure_neo4j_schema,
+    mark_schema_ready,
 )
 from context_intelligence_server.registry import SessionRegistry
 from context_intelligence_server.routers.admin import router as admin_router
@@ -289,8 +290,34 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     # here mirrors run_repair's contract: a :Node constraint data conflict
     # raises a RuntimeError naming `doctor --fix` instead of being logged
     # and swallowed.
-    await ensure_neo4j_schema(app.state.neo4j_driver, fail_on_data_conflict=True)
-    logger.info("lifespan_startup: Neo4j schema initialized")
+    schema_fully_established = await ensure_neo4j_schema(
+        app.state.neo4j_driver, fail_on_data_conflict=True
+    )
+    # Seed the PROCESS-wide schema latch, but ONLY on a fully-established pass.
+    #
+    # fail_on_data_conflict=True makes this call fail closed on a :Node
+    # constraint DATA conflict -- but a CONNECTIVITY failure on any individual
+    # index/constraint is deliberately swallowed and reported through the
+    # return value instead (see ensure_neo4j_schema's docstring). Latching
+    # unconditionally would therefore mark a HALF-BUILT schema as ready and
+    # permanently disable the per-flush self-heal for the whole process --
+    # exactly the "constraint created once, never retried" gap
+    # Neo4jGraphStore._ensure_schema exists to close.
+    #
+    # On the happy path this seed is what stops every per-session store from
+    # re-running the same ~11-statement catalog pass on its first flush -- and,
+    # whenever that pass cannot complete, on EVERY subsequent flush -- competing
+    # for the very bolt pool it needs. See neo4j_store._SCHEMA_READY.
+    if schema_fully_established:
+        mark_schema_ready()
+        logger.info("lifespan_startup: Neo4j schema initialized")
+    else:
+        logger.warning(
+            "lifespan_startup: Neo4j schema NOT fully established (indexes or "
+            "constraints missing); leaving the process-wide latch unset so the "
+            "flush path retries schema init (rate-limited by "
+            "neo4j_store._SCHEMA_RETRY_BACKOFF_SECONDS)."
+        )
     # Fail-loud migration-health guard: duplicate nodes are already caught
     # above by the :Node constraint (fail_on_data_conflict=True); this catches
     # the OTHER un-migrated shape the constraint can't see on its own --

context_intelligence_server/neo4j_store.py

@@ -14,6 +14,7 @@
 import json
 import logging
 import re
+import time
 from collections.abc import Generator
 from datetime import datetime
 from typing import Any, LiteralString, cast
@@ -188,6 +189,55 @@ def build_bounded_neo4j_driver(
     f"MATCH (n:{_UNIVERSAL_NODE_LABEL} {{node_id: $node_id, workspace: $workspace}})"
 )
 
+# Single-node read-by-identity, built on the shared _NODE_MATCH_BY_ID prefix so
+# it can never drift back to a label-free MATCH.  This is the get_node() Neo4j
+# fallback, which runs on the hot per-event path (touch_session -> get_node), so
+# it MUST plan as a NodeUniqueIndexSeek: as a label-free MATCH it planned as an
+# AllNodesScan over the whole graph and, at 12.09M nodes, took ~60s per call
+# while holding its pooled connection for the duration.
+_NODE_GET_BY_ID_CYPHER = (
+    f"{_NODE_MATCH_BY_ID} RETURN properties(n) AS props, labels(n) AS lbls"
+)
+
+# Single-edge read-by-endpoints, anchored on BOTH :Node endpoints so the
+# planner seeks the :Node(node_id, workspace) unique index twice and then walks
+# only that node's own relationships.
+#
+# The previous form was ``MATCH ()-[r]->() WHERE r.src_id = ... AND r.dst_id =
+# ... AND r.workspace = ...``: relationship property indexes in Neo4j are
+# TYPE-scoped, and this pattern names no type, so no index could back it and it
+# planned as an AllRelationshipsScan -- the same defect class as the label-free
+# get_node() MATCH, over a set that is larger still (a graph has more
+# relationships than nodes).
+#
+# The relationship-property predicates are deliberately RETAINED after the
+# anchor. They are now evaluated against the handful of relationships incident
+# to one node rather than every relationship in the graph, so they cost nothing
+# -- and keeping them makes this a strictly NARROWING change: any edge the old
+# form would have returned and the new one would not is one whose endpoints do
+# not exist as :Node, which the edge writer (_edge_merge_cypher, which MERGEs
+# both :Node endpoints) structurally cannot produce.
+_EDGE_GET_BY_ENDPOINTS_CYPHER = (
+    f"MATCH (src:{_UNIVERSAL_NODE_LABEL} "
+    "{node_id: $src_id, workspace: $workspace})-[r]->"
+    f"(dst:{_UNIVERSAL_NODE_LABEL} "
+    "{node_id: $dst_id, workspace: $workspace}) "
+    "WHERE r.src_id = $src_id AND r.dst_id = $dst_id "
+    "AND r.workspace = $workspace "
+    "RETURN properties(r) AS props"
+)
+
+# Delegation lookup by the session it spawned. Named here (rather than inlined
+# at the call site) so it sits beside the composite index that backs it --
+# idx_delegation_sub_session on :Delegation(sub_session_id, workspace), created
+# in ensure_neo4j_schema. The query shape was already correct; before that index
+# existed it simply had nothing to seek and planned as a NodeByLabelScan across
+# every :Delegation node.
+_DELEGATION_BY_SUB_SESSION_CYPHER = (
+    "MATCH (d:Delegation {sub_session_id: $sid, workspace: $workspace}) "
+    "RETURN properties(d) AS props"
+)
+
 
 def _edge_merge_cypher(edge_type: str) -> str:
     """Return the UNWIND edge-MERGE query for *edge_type* — self-healing endpoints.
@@ -619,6 +669,96 @@ async def run_repair(driver: Any, database: str = "neo4j") -> dict[str, int]:
     }
 
 
+# ---------------------------------------------------------------------------
+# Process-wide schema latch.
+#
+# ``ensure_neo4j_schema`` has two callers whose SCOPES differ, and conflating
+# them is what turned a connectivity blip into a per-flush DDL storm:
+#
+#   - The FastAPI lifespan (``main.py``) runs it ONCE PER PROCESS, before the
+#     server accepts a single request, with ``fail_on_data_conflict=True`` --
+#     it refuses to boot unless the schema is fully established.
+#   - ``Neo4jGraphStore._ensure_schema`` runs it from inside ``_flush_body``,
+#     latched on ``self._schema_initialized`` -- a PER-STORE flag, and a store
+#     is constructed PER SESSION (``registry.get_or_create``).
+#
+# So the per-store latch made every new session pay a full ~11-statement DDL
+# pass on its first flush, and -- whenever that pass could not complete -- pay
+# it again on EVERY subsequent flush of EVERY session, with no backoff. Each
+# attempt opens a session and issues ~11 sequential catalog statements, every
+# one of which must acquire a pooled bolt connection; under pool starvation
+# each acquisition burns the full acquisition timeout before failing. That is
+# a large, futile, self-reinforcing load applied at exactly the moment the
+# pool is already exhausted, and it runs BEFORE any data is written
+# (``_flush_body`` awaits ``_ensure_schema`` ahead of Phase 1).
+#
+# Two changes fix that without weakening the self-heal:
+#
+#   1. ``_SCHEMA_READY`` is PROCESS-wide. The lifespan seeds it via
+#      ``mark_schema_ready()`` once its own fail-closed pass succeeds, so in
+#      the server the per-flush call becomes a single boolean read forever --
+#      it can never discover anything cold start did not already establish.
+#      Contexts with no lifespan (tests, CLI tools, direct store use) never
+#      seed it and keep today's behaviour: the first store to fully establish
+#      the schema sets it for the process.
+#   2. A failed pass no longer retries on the very next flush.
+#      ``_SCHEMA_RETRY_BACKOFF_SECONDS`` puts a floor between attempts, so a
+#      connectivity blip can no longer become a per-flush storm. The self-heal
+#      survives -- it just stops competing with the writes it is meant to
+#      enable.
+# ---------------------------------------------------------------------------
+_SCHEMA_RETRY_BACKOFF_SECONDS = 30.0
+
+_SCHEMA_READY: bool = False
+_SCHEMA_LAST_ATTEMPT: float | None = None
+
+
+def mark_schema_ready() -> None:
+    """Record that the Neo4j schema is fully established for THIS process.
+
+    Called by the FastAPI lifespan after its own fail-closed
+    ``ensure_neo4j_schema(..., fail_on_data_conflict=True)`` pass succeeds, and
+    by any store whose own pass fully establishes the schema. Once set, every
+    per-flush ``_ensure_schema`` call short-circuits to a boolean read.
+    """
+    global _SCHEMA_READY
+    _SCHEMA_READY = True
+
+
+def schema_ready() -> bool:
+    """Whether the schema has been fully established in this process."""
+    return _SCHEMA_READY
+
+
+def reset_schema_state() -> None:
+    """Clear the process-wide schema latch and backoff clock.
+
+    Exists for tests: the latch is deliberately process-global, so without an
+    explicit reset one test that establishes the schema would silently suppress
+    the schema path in every test that ran after it.
+    """
+    global _SCHEMA_READY, _SCHEMA_LAST_ATTEMPT
+    _SCHEMA_READY = False
+    _SCHEMA_LAST_ATTEMPT = None
+
+
+def _schema_retry_allowed() -> bool:
+    """Whether enough time has passed since the last un-latched schema attempt.
+
+    Records the attempt time as a side effect when it returns True, so two
+    concurrent flushes cannot both decide to run a full DDL pass.
+    """
+    global _SCHEMA_LAST_ATTEMPT
+    now = time.monotonic()
+    if (
+        _SCHEMA_LAST_ATTEMPT is not None
+        and now - _SCHEMA_LAST_ATTEMPT < _SCHEMA_RETRY_BACKOFF_SECONDS
+    ):
+        return False
+    _SCHEMA_LAST_ATTEMPT = now
+    return True
+
+
 async def ensure_neo4j_schema(
     driver: Any,
     database: str = "neo4j",
@@ -777,6 +917,20 @@ async def _create_index(statement: str) -> bool:
             )
             and fully_established
         )
+        # Backs find_delegation_by_sub_session's lookup
+        # (_DELEGATION_BY_SUB_SESSION_CYPHER). Without it that MATCH names a
+        # label but no indexed property, so it plans as a NodeByLabelScan over
+        # EVERY :Delegation node in the graph. It is not a rare path -- the
+        # self-delegation resolver hits it on live ingest whenever a parent
+        # Delegation has already flushed out of the in-memory buffer -- and it
+        # grows without bound as delegation volume grows.
+        fully_established = (
+            await _create_index(
+                "CREATE INDEX idx_delegation_sub_session IF NOT EXISTS "
+                "FOR (n:Delegation) ON (n.sub_session_id, n.workspace)"
+            )
+            and fully_established
+        )
         # NOTE: relationship created_by is intentionally NOT indexed in v1 — no consumer
         # yet; querying edges by contributor is unsupported until indexed in a future phase.
 
@@ -1416,9 +1570,8 @@ async def get_node(self, node_id: str) -> dict[str, Any] | None:
         # Neo4j fallback
         try:
             result = await self._driver.execute_query(
-                "MATCH (n) WHERE n.node_id = $id AND n.workspace = $workspace "
-                "RETURN properties(n) AS props, labels(n) AS lbls",
-                {"id": node_id, "workspace": self.workspace},
+                cast(LiteralString, _NODE_GET_BY_ID_CYPHER),
+                {"node_id": node_id, "workspace": self.workspace},
                 database_=self._database,
             )
             records = result.records
@@ -1464,8 +1617,7 @@ async def find_delegation_by_sub_session(
         # Neo4j fallback
         try:
             result = await self._driver.execute_query(
-                "MATCH (d:Delegation {sub_session_id: $sid, workspace: $workspace}) "
-                "RETURN properties(d) AS props",
+                cast(LiteralString, _DELEGATION_BY_SUB_SESSION_CYPHER),
                 {"sid": sub_session_id, "workspace": workspace},
                 database_=self._database,
             )
@@ -1495,10 +1647,7 @@ async def get_edge(self, src_id: str, dst_id: str) -> dict[str, Any] | None:
         # Neo4j fallback
         try:
             result = await self._driver.execute_query(
-                "MATCH ()-[r]->() "
-                "WHERE r.src_id = $src_id AND r.dst_id = $dst_id "
-                "AND r.workspace = $workspace "
-                "RETURN properties(r) AS props",
+                cast(LiteralString, _EDGE_GET_BY_ENDPOINTS_CYPHER),
                 {"src_id": src_id, "dst_id": dst_id, "workspace": self.workspace},
                 database_=self._database,
             )
@@ -1719,15 +1868,42 @@ async def _ensure_schema(self) -> None:
         any duplicates.  This closes the "constraint created once, never retried"
         data-integrity gap, where a missing uniqueness constraint would otherwise let
         concurrent MERGE accrue duplicate Session/Event nodes until process restart.
+
+        Two latches, checked in order, both O(1):
+
+        - ``self._schema_initialized`` -- THIS store already established it.
+        - ``schema_ready()`` -- ANY caller in this PROCESS already established
+          it, including the FastAPI lifespan's own fail-closed cold-start pass.
+          In the server that is the branch which always fires: cold start
+          already ran ``ensure_neo4j_schema(..., fail_on_data_conflict=True)``
+          and refused to boot otherwise, so a per-flush pass here can only
+          re-confirm what is already true -- at a cost of ~11 catalog
+          statements per flush, per session. See the ``_SCHEMA_READY`` block
+          above for why that mattered so much under pool starvation.
+
+        When neither latch is set, the attempt is additionally rate-limited by
+        ``_schema_retry_allowed()``: a pass that could not complete (e.g. Neo4j
+        unreachable, connectivity errors swallowed so real events are not
+        dead-lettered) no longer retries on the very NEXT flush. Skipping a
+        retry is safe precisely because the pass is a no-op in the healthy case
+        and a storm in the degraded one -- the writes that follow it have never
+        depended on it having run on this particular flush.
         """
-        if self._schema_initialized:
+        if self._schema_initialized or schema_ready():
+            return
+
+        if not _schema_retry_allowed():
             return
 
         fully_established = await ensure_neo4j_schema(self._driver, self._database)
         if fully_established:
             self._schema_initialized = True
-        # else: leave the flag False so the NEXT flush retries schema init (self-heals
-        # once Neo4j is reachable / duplicates are cleared by the dedup pass).
+            # "Fully established" is a PROCESS-wide fact, not a per-store one --
+            # latch it so sibling stores (one per session) never repeat the pass.
+            mark_schema_ready()
+        # else: leave both flags False so a LATER flush retries schema init, subject
+        # to the backoff above (self-heals once Neo4j is reachable / duplicates are
+        # cleared by the dedup pass).
 
     async def close(self) -> None:
         """Flush pending writes and close the driver, if this store owns it.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "context-intelligence-server"
-version = "6.7.1"
+version = "6.7.2"
 description = "Context Intelligence Server for Amplifier"
 requires-python = ">=3.11"
 dependencies = [

tests/conftest.py

@@ -20,6 +20,36 @@
 from context_intelligence_server.main import app, registry  # noqa: E402
 from context_intelligence_server.services import HookStateService  # noqa: E402
 
+# Guarded import so the suite still COLLECTS against unfixed source (where the
+# process-wide schema latch does not exist yet). Without this, reverting the
+# fix would break collection for every test in the repo, and the new guards
+# below would "fail" at import time rather than for the reason they exist.
+try:  # pragma: no cover - import resolution differs pre/post fix
+    from context_intelligence_server.neo4j_store import (  # noqa: E402
+        reset_schema_state,
+    )
+except ImportError:  # pragma: no cover - exercised only against unfixed code
+
+    def reset_schema_state() -> None:  # type: ignore[misc]
+        """No-op stand-in: unfixed code has no process-wide latch to reset."""
+
+
+@pytest.fixture(autouse=True)
+def _reset_neo4j_schema_state() -> Generator[None, None, None]:
+    """Clear the process-wide Neo4j schema latch around EVERY test.
+
+    ``neo4j_store._SCHEMA_READY`` is deliberately process-global: it is what
+    stops every per-session store from re-running the same ~11-statement
+    catalog pass on its first flush. Process-global state is also test-order
+    poison -- without this reset, the first test that fully establishes the
+    schema would silently short-circuit the schema path in every test that ran
+    after it, and those tests would pass or fail depending on collection order.
+    Resetting on both sides of the yield keeps each test's view independent.
+    """
+    reset_schema_state()
+    yield
+    reset_schema_state()
+
 
 # ---------------------------------------------------------------------------
 # Shared Neo4j mock helpers (used by POST /cypher tests)