fix(neo4j): reuse one bounded driver across per-session graph stores

context_intelligence_server/config.py

@@ -797,6 +797,26 @@ def resolve_neo4j_query(self) -> Neo4jClientConfig:
             access_mode="READ",
         )
 
+    # Upper bound on concurrent bolt connections for a driver shared across many
+    # logical callers (the lifespan admin driver, the lifespan query driver, the
+    # registry's shared per-session driver). Well under the server's default
+    # bolt thread-pool size so a driver leak can no longer starve it. The neo4j
+    # driver's own default is 100; 50 is a deliberate reduction, since every
+    # session now shares one pool instead of holding a private one.
+    #
+    # No companion max_connection_lifetime knob: the driver already recycles
+    # pooled connections at 3600 s by default, so a setting whose default equals
+    # the library default would change nothing.
+    neo4j_max_connection_pool_size: int = 50
+
+    @field_validator("neo4j_max_connection_pool_size")
+    @classmethod
+    def _validate_neo4j_max_connection_pool_size(cls, v: int) -> int:
+        """Fail loud on a non-positive pool size."""
+        if v <= 0:
+            raise ValueError(f"neo4j_max_connection_pool_size must be > 0, got {v}")
+        return v
+
     # -------------------------------------------------------------------------
     # Storage paths
     # -------------------------------------------------------------------------

context_intelligence_server/main.py

@@ -16,7 +16,7 @@
 
 from fastapi import Depends, FastAPI, HTTPException, Request
 from fastapi.responses import JSONResponse, Response
-from neo4j import READ_ACCESS, WRITE_ACCESS, AsyncGraphDatabase
+from neo4j import READ_ACCESS, WRITE_ACCESS
 
 from context_intelligence_server import __version__
 from context_intelligence_server.auth import (
@@ -41,6 +41,7 @@
     EventResponse,
 )
 from context_intelligence_server.neo4j_store import (
+    build_bounded_neo4j_driver,
     count_untagged_nodes,
     ensure_neo4j_schema,
 )
@@ -61,13 +62,18 @@ def _neo4j_access_const(mode: str) -> str:
 
 
 def build_neo4j_driver(config: Neo4jClientConfig) -> Any:
-    """Construct an AsyncGraphDatabase driver from a resolved Neo4j client config.
+    """Construct the pool-bounded admin AsyncGraphDatabase driver.
 
     Shared by ``lifespan()`` (the admin driver, on every server boot) and
     ``doctor.run_doctor()`` (the CLI), so the two entry points can never
-    construct the connection differently.
+    construct the connection differently. Delegates the actual driver
+    construction to ``build_bounded_neo4j_driver`` so the pool-bounding kwargs
+    have one source of truth, shared with ``SessionRegistry``'s driver.
     """
-    return AsyncGraphDatabase.driver(config.url, auth=config.auth)
+    return build_bounded_neo4j_driver(
+        config,
+        max_connection_pool_size=_settings.neo4j_max_connection_pool_size,
+    )
 
 
 # ---------------------------------------------------------------------------
@@ -255,9 +261,11 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     # build_neo4j_driver() is the SAME helper doctor.run_doctor() uses, so the
     # server and the doctor CLI can never construct this connection differently.
     app.state.neo4j_driver = build_neo4j_driver(_admin)
-    # Cypher-query (read-intent): /cypher + dashboard reads.
-    app.state.neo4j_query_driver = AsyncGraphDatabase.driver(
-        _query.url, auth=_query.auth
+    # Cypher-query (read-intent): /cypher + dashboard reads. Bounded through the
+    # same helper as the admin driver so every process-wide pool shares one cap.
+    app.state.neo4j_query_driver = build_bounded_neo4j_driver(
+        _query,
+        max_connection_pool_size=_settings.neo4j_max_connection_pool_size,
     )
     # Stash the resolved query access_mode so /cypher opens READ sessions without
     # re-resolving settings on every request.
@@ -411,9 +419,25 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
             _sweep_task.cancel()
             with suppress(asyncio.CancelledError):
                 await _sweep_task
+        # ORDER IS LOAD-BEARING. Quiesce the drainers FIRST. Every session's
+        # graph store now shares ONE driver, so closing it under a live drainer
+        # is no longer a per-session concern: the drainer's batch fails, it
+        # spends its max_delivery_attempts budget in ~250 ms, and
+        # _handle_exhausted_batch dead-letters each line AND commits the offset
+        # past it -- discarding healthy events that merely happened to be
+        # queued at shutdown, with no replay on the next boot. Cancelling first
+        # routes each drainer through CancelledError -> _safe_close -> a final
+        # flush while the driver is still open.
+        logger.info("lifespan_shutdown: quiescing drain workers")
+        await registry.shutdown_workers()
         logger.info("lifespan_shutdown: closing Neo4j drivers")
         await app.state.neo4j_driver.close()
         await app.state.neo4j_query_driver.close()
+        # The registry's shared per-session driver is independent of the two
+        # above (its own pool, built from settings.resolve_neo4j_admin() the
+        # first time a session is created) -- close it here too so no bolt
+        # connection outlives the process.
+        await registry.close_neo4j_driver()
 
 
 app = FastAPI(

context_intelligence_server/neo4j_store.py

@@ -22,8 +22,58 @@
 from neo4j import unit_of_work as _unit_of_work
 from neo4j.exceptions import DriverError, Neo4jError
 
+from context_intelligence_server.config import Neo4jClientConfig
+
 _LOG = logging.getLogger(__name__)
 
+
+def build_bounded_neo4j_driver(
+    config: Neo4jClientConfig,
+    *,
+    max_connection_pool_size: int,
+    connection_acquisition_timeout: float | None = None,
+) -> Any:
+    """Construct an AsyncGraphDatabase driver with a bounded connection pool.
+
+    Single source of truth for the pool-bounding kwargs applied to any driver
+    meant to be shared across many logical callers (the lifespan admin driver,
+    the lifespan query driver, the registry's shared per-session driver). All
+    three construct through here so they can never diverge.
+
+    Args:
+        max_connection_pool_size:
+            Hard cap on concurrent bolt connections for this driver.
+        connection_acquisition_timeout:
+            How long a caller waits for a free pooled connection before
+            failing. ``None`` (the lifespan drivers) leaves the driver default
+            in place, matching what those two did before they were routed
+            through this helper. ``SessionRegistry`` passes
+            ``settings.neo4j_lock_timeout`` so the shared per-session driver
+            keeps the acquisition budget the per-session drivers it replaced
+            carried -- and it matters more now, not less: one bounded pool is
+            shared by every session, so acquisition can genuinely queue.
+
+    ``max_connection_lifetime`` is deliberately not set: the neo4j driver
+    already recycles pooled connections at 3600 s by default, so passing it
+    would be a knob that changes nothing.
+    """
+    kwargs: dict[str, Any] = {
+        "max_connection_pool_size": max_connection_pool_size,
+        # Explicit auto-retry budget for transient errors (e.g. deadlocks) so
+        # the managed-transaction retry window is deliberate and reviewable
+        # rather than relying on the driver default implicitly. Carried over
+        # verbatim from the per-session driver construction this helper
+        # subsumed.
+        "max_transaction_retry_time": 30.0,
+    }
+    if (
+        connection_acquisition_timeout is not None
+        and connection_acquisition_timeout > 0
+    ):
+        kwargs["connection_acquisition_timeout"] = connection_acquisition_timeout
+    return AsyncGraphDatabase.driver(config.url, auth=config.auth, **kwargs)
+
+
 # ---------------------------------------------------------------------------
 # Cypher identifier validation
 # ---------------------------------------------------------------------------
@@ -1179,12 +1229,15 @@ def __init__(
         flush_chunk_rows: int = 100,
         flush_chunk_bytes: int = 4_194_304,
         neo4j_lock_timeout: float | None = None,
+        driver: Any | None = None,
     ) -> None:
-        """Initialise the store and create the async Neo4j driver.
+        """Initialise the store, reusing or creating the async Neo4j driver.
 
         Args:
             uri:               Bolt/neo4j URI, e.g. ``bolt://localhost:7687``.
+                               Ignored when ``driver`` is provided.
             auth:              ``(username, password)`` tuple, or ``None`` for no-auth.
+                               Ignored when ``driver`` is provided.
             database:          Target Neo4j database name (default: ``"neo4j"``).
             workspace:         Workspace to scope writes to.  ``None`` resolves to
                                ``"default"`` via the ``workspace`` property.
@@ -1195,18 +1248,29 @@ def __init__(
                                so a blocked flush raises ``Neo4jError``
                                instead of parking forever.  ``None`` disables
                                the timeout (default: no per-transaction limit).
-                               Also sets ``connection_acquisition_timeout`` on
-                               the driver to the same value so pool-exhaustion
-                               failures also surface quickly.
+                               When the store builds its own driver (``driver``
+                               not provided), this also sets
+                               ``connection_acquisition_timeout`` on it to the
+                               same value so pool-exhaustion failures surface
+                               quickly.
+            driver:            A pre-built async driver to reuse instead of
+                               constructing a new one. When provided, this store
+                               does not own the driver's lifecycle: ``close()``
+                               flushes and no-ops on the driver itself, leaving
+                               it open for other stores sharing it.
         """
-        # Explicit auto-retry budget for transient errors (e.g. deadlocks) so the
-        # managed-transaction retry window is deliberate and reviewable rather than
-        # relying on the driver default implicitly. 30.0s is a working default;
-        # design Open Question #3 — verify driver 6.1.0 backoff constants before tuning.
-        driver_kwargs: dict[str, Any] = {"max_transaction_retry_time": 30.0}
-        if neo4j_lock_timeout is not None and neo4j_lock_timeout > 0:
-            driver_kwargs["connection_acquisition_timeout"] = neo4j_lock_timeout
-        self._driver = AsyncGraphDatabase.driver(uri, auth=auth, **driver_kwargs)
+        if driver is not None:
+            self._driver = driver
+            self._owns_driver = False
+        else:
+            # Explicit auto-retry budget for transient errors (e.g. deadlocks) so
+            # the managed-transaction retry window is deliberate and reviewable
+            # rather than relying on the driver default implicitly.
+            driver_kwargs: dict[str, Any] = {"max_transaction_retry_time": 30.0}
+            if neo4j_lock_timeout is not None and neo4j_lock_timeout > 0:
+                driver_kwargs["connection_acquisition_timeout"] = neo4j_lock_timeout
+            self._driver = AsyncGraphDatabase.driver(uri, auth=auth, **driver_kwargs)
+            self._owns_driver = True
         self._database = database
         self._workspace = workspace
         self._created_by: str | None = None
@@ -1224,6 +1288,19 @@ def __init__(
             else None
         )
 
+    # ------------------------------------------------------------------
+    # owns_driver property
+    # ------------------------------------------------------------------
+
+    @property
+    def owns_driver(self) -> bool:
+        """True when this store built its own driver; False when injected.
+
+        Governs ``close()``: a store that does not own its driver must never
+        close it, since other stores may still be using it.
+        """
+        return self._owns_driver
+
     # ------------------------------------------------------------------
     # workspace property
     # ------------------------------------------------------------------
@@ -1633,10 +1710,15 @@ async def _ensure_schema(self) -> None:
         # once Neo4j is reachable / duplicates are cleared by the dedup pass).
 
     async def close(self) -> None:
-        """Flush pending writes, await any background task, and close the driver.
+        """Flush pending writes and close the driver, if this store owns it.
 
         Handles event-loop mismatch gracefully when closing the driver from a
         different loop context.  Sets ``_closed`` on completion.
+
+        When the driver was injected (``owns_driver`` is False), the driver is
+        left open: it is shared with other stores/callers and closing it here
+        would break them out from under their own in-flight work. The shared
+        driver's owner is responsible for closing it exactly once.
         """
         # Final flush to persist remaining buffer contents
         try:
@@ -1646,11 +1728,12 @@ async def close(self) -> None:
                 "Final flush failed during close; buffered writes may be lost"
             )
 
-        # Close the driver, ignoring event-loop mismatch errors
-        try:
-            await self._driver.close()
-        except RuntimeError:
-            pass
+        if self._owns_driver:
+            # Close the driver, ignoring event-loop mismatch errors
+            try:
+                await self._driver.close()
+            except RuntimeError:
+                pass
 
         self._closed = True
 

context_intelligence_server/registry.py

@@ -12,11 +12,14 @@
 
 from context_intelligence_server.blob_store import AsyncDiskBlobStore
 from context_intelligence_server.config import get_settings
-from context_intelligence_server.status import EventRecord, ring_buffer
-from context_intelligence_server.neo4j_store import Neo4jGraphStore
+from context_intelligence_server.neo4j_store import (
+    Neo4jGraphStore,
+    build_bounded_neo4j_driver,
+)
 from context_intelligence_server.pipeline import process_event, setup_handlers
 from context_intelligence_server.queue_manager import Batch, QueueManager
 from context_intelligence_server.services import HookStateService
+from context_intelligence_server.status import EventRecord, ring_buffer
 
 logger = logging.getLogger("context_intelligence_server")
 
@@ -76,6 +79,11 @@ def __init__(self) -> None:
         self._queue_manager: QueueManager | None = None
         self._write_semaphore: asyncio.Semaphore | None = None
         self._max_delivery_attempts: int = 0
+        # Shared, pool-bounded Neo4j driver for every per-session Neo4jGraphStore
+        # (see _ensure_neo4j_driver). Built lazily for the same reason as
+        # _queue_manager; kept separate from _ensure_infra so the two concerns
+        # can evolve independently.
+        self._neo4j_driver: Any | None = None
         # Live conservation counters surfaced via /status (accepted/written/
         # replayed/write_retries) so silently-dropped events are observable.
         self._accepted_total: int = 0
@@ -106,6 +114,72 @@ def queue_manager(self) -> QueueManager:
         assert self._queue_manager is not None
         return self._queue_manager
 
+    def _ensure_neo4j_driver(self) -> Any:
+        """Build the shared, pool-bounded Neo4j driver on first use.
+
+        Lazy for the same reason as ``_ensure_infra``. Kept as its own method
+        (not folded into ``_ensure_infra``) so the two constructions stay
+        independent edits.
+        """
+        if self._neo4j_driver is None:
+            settings = get_settings()
+            admin = settings.resolve_neo4j_admin()
+            self._neo4j_driver = build_bounded_neo4j_driver(
+                admin,
+                max_connection_pool_size=settings.neo4j_max_connection_pool_size,
+                # Parity with the per-session driver this one replaces: a
+                # blocked acquisition must surface on the SAME budget as a
+                # blocked transaction. Load-bearing now in a way it was not
+                # before -- every session shares this one bounded pool, so
+                # acquisition can actually queue.
+                connection_acquisition_timeout=settings.neo4j_lock_timeout,
+            )
+        return self._neo4j_driver
+
+    @property
+    def neo4j_driver(self) -> Any:
+        """The single shared, pool-bounded driver used by every per-session
+        Neo4jGraphStore -- never closed by a per-session finalize."""
+        return self._ensure_neo4j_driver()
+
+    async def shutdown_workers(self) -> None:
+        """Quiesce every drain worker BEFORE the shared driver is closed.
+
+        Ordering invariant (must run before ``close_neo4j_driver``): a live
+        drainer that meets a closed shared driver fails its batch, spends its
+        ``max_delivery_attempts`` budget in ~250 ms (5 attempts x the 50 ms
+        ``_DRAIN_POLL_INTERVAL`` backoff), and falls into
+        ``_handle_exhausted_batch`` -- which dead-letters each line AND commits
+        the offset past it. Those are healthy events that merely happened to be
+        queued at shutdown, and once dead-lettered they never replay.
+
+        Cancelling instead routes each drainer through its ``CancelledError``
+        handler -> ``_safe_close(worker)`` -> a final flush on a driver that is
+        still open. Anything left uncommitted stays in the durable queue and
+        replays on the next boot, which is the pre-shared-driver behaviour.
+
+        Exceptions are collected, not raised: shutdown must not be derailed by
+        one failing worker.
+        """
+        tasks = [w.task for w in self._workers.values() if w.task is not None]
+        if not tasks:
+            return
+        for task in tasks:
+            task.cancel()
+        await asyncio.gather(*tasks, return_exceptions=True)
+
+    async def close_neo4j_driver(self) -> None:
+        """Close the shared driver exactly once, at process shutdown.
+
+        Call ``shutdown_workers()`` first -- see its docstring for why closing
+        this driver under a live drainer dead-letters good events.
+
+        No-op if the driver was never built (no session has run yet).
+        """
+        if self._neo4j_driver is not None:
+            await self._neo4j_driver.close()
+            self._neo4j_driver = None
+
     @property
     def write_semaphore(self) -> asyncio.Semaphore:
         """The single shared global cap on concurrent Neo4j-write flushes."""
@@ -742,6 +816,7 @@ def get_or_create(
             neo4j_store = Neo4jGraphStore(
                 uri=_admin.url,
                 auth=_admin.auth,
+                driver=self.neo4j_driver,
                 flush_chunk_rows=settings.neo4j_flush_chunk_rows,
                 flush_chunk_bytes=settings.neo4j_flush_chunk_bytes,
                 neo4j_lock_timeout=settings.neo4j_lock_timeout,