diff --git a/context_intelligence_server/main.py b/context_intelligence_server/main.py index c2faf78..e37d3dc 100644 --- a/context_intelligence_server/main.py +++ b/context_intelligence_server/main.py @@ -42,7 +42,6 @@ ) from context_intelligence_server.neo4j_store import ( build_bounded_neo4j_driver, - count_untagged_nodes, ensure_neo4j_schema, mark_schema_ready, ) @@ -245,105 +244,41 @@ async def _crash_recovery_sweep_loop(interval: int, respawn_limit: int) -> None: logger.warning("crash_recovery_sweep: tick failed, will retry: %s", exc) -@asynccontextmanager -async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: - """Manage application lifespan: configure logging and create shared Neo4j driver.""" - setup_logging() - _admin = _settings.resolve_neo4j_admin() - _query = _settings.resolve_neo4j_query() - logger.info( - "lifespan_startup: creating Neo4j drivers admin_url=%s query_url=%s query_access_mode=%s", - _admin.url, - _query.url, - _query.access_mode, - ) - # Admin (read/write): schema init + all mutation paths. Keep the existing - # app.state.neo4j_driver NAME so nothing that reads it silently breaks. - # 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. 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. - app.state.neo4j_query_access_mode = _query.access_mode - # Initialize schema (indexes + uniqueness constraints) BEFORE the server starts - # accepting requests. This ensures the Session uniqueness constraint is active - # before any concurrent flush() transactions execute MERGE, which prevents the - # duplicate-Session-node race condition observed under concurrent upload load. - logger.info( - "lifespan_startup: initializing Neo4j schema (indexes + uniqueness constraints)" - ) - # Cold start FAILS LOUD on schema/data corruption that requires - # `doctor --fix` -- an un-migrated graph (duplicate legacy nodes OR - # nodes lacking the universal :Node label). Nothing has been written yet - # at cold start, so refusing to boot loses no data: this is the safest - # possible moment to surface an impossible state as an un-missable - # signal rather than a log line someone greps for later. Contrast with - # the flush path (Neo4jGraphStore._ensure_schema), which must keep - # self-healing and never raise (Salil's blocker -- raising there would - # dead-letter real in-flight activity records). fail_on_data_conflict=True - # here mirrors run_repair's contract: a :Node constraint data conflict - # raises a RuntimeError naming `doctor --fix` instead of being logged - # and swallowed. - 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 -- - # nodes that simply lack the :Node label altogether, which violate no - # constraint and so raise nothing by themselves. O(1) via the counts - # store (see count_untagged_nodes) -- this must never regress into the - # AllNodesScan stall PR #67 removed from the write path. - # - # A connectivity/probe failure here is NOT the same as "confirmed - # un-migrated" -- it means graph state could not be determined, not that - # it was determined to be bad -- so it is logged at DEBUG and swallowed - # rather than treated as a corruption finding; the flush path's - # self-heal still covers a genuinely dirty graph once it becomes - # reachable. +async def _startup_recovery(app: FastAPI) -> None: + """Crash-recovery pass + deferred-backlog sweep, off the startup path. + + Runs as a background task created by ``lifespan`` so that startup can + complete -- and the HTTP surface open -- in ~1s regardless of how large + the durable spool is. Sets ``app.state.recovery_complete`` when the + one-shot recovery pass is done, so ``/status`` can say so honestly + rather than reporting un-seeded counters as though they were settled. + + Thin wrapper around ``_startup_recovery_body``: as a BACKGROUND task, an + unhandled exception here would otherwise be swallowed until garbage + collection ("Task exception was never retrieved") -- silently leaving the + counters unseeded and no drainers respawned, with a server that looks + perfectly healthy. On the old inline path the same failure aborted boot + loudly. Log it, record it for ``/status``, and always set the event so + ``/status`` cannot report "recovery in progress" forever. + """ try: - untagged = await count_untagged_nodes(app.state.neo4j_driver) - except Exception as exc: # noqa: BLE001 - connectivity probe, not a confirmed bad state - _LOG_MSG = "startup migration-health probe skipped (graph unreachable?): %s" - logger.debug(_LOG_MSG, exc) - untagged = 0 - if untagged: - raise RuntimeError( - f"Neo4j graph has {untagged} node(s) lacking the :Node label " - "(un-migrated). Cold start refuses to boot to avoid duplicating " - "them on write. Run: context-intelligence-server doctor --fix" + await _startup_recovery_body(app) + except asyncio.CancelledError: + raise # shutdown -- not a failure, and must propagate + except Exception as exc: # noqa: BLE001 - background task boundary + app.state.recovery_error = repr(exc) + logger.exception( + "startup_recovery: FAILED -- conservation counters may be unseeded " + "and recovered drainers may not have respawned. The server is still " + "serving; a later boot's recover() reports the same sessions again, " + "and a new event for a session spawns its drainer via get_or_create()." ) + finally: + app.state.recovery_complete.set() + + +async def _startup_recovery_body(app: FastAPI) -> None: + """The actual recovery pass. See ``_startup_recovery`` for why it is split.""" # Crash recovery: on startup, respawn one drainer per # session that still has an undrained, complete line. The workspace is # parsed from that session's FIRST log line so the respawned worker is @@ -405,7 +340,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: # silent, un-discoverable fact. Names the exact counts and the # setting to raise. logger.warning( - "lifespan_startup: crash-recovery respawn cap reached " + "startup_recovery: crash-recovery respawn cap reached " "(crash_recovery_respawn_limit=%d): %d/%d respawned this boot, " "%d session(s) deferred to a later boot (untouched on disk, " "still fully recoverable). Raise crash_recovery_respawn_limit " @@ -416,36 +351,163 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: deferred_count, ) logger.info( - "lifespan_startup: crash recovery respawned %d/%d drainers", + "startup_recovery: crash recovery respawned %d/%d drainers", respawned, len(recovered), ) + # The one-shot pass is done: counters are seeded and this boot's drainers + # are up. /status can now report settled numbers rather than mid-recovery + # ones. Set BEFORE the sweep loop below, which never returns. (The wrapper + # also sets it in a finally, so a failure cannot pin /status at + # "in progress" forever -- but on the success path it must be set HERE, or + # the sweep loop would delay it indefinitely.) + app.state.recovery_complete.set() # Periodic deferred-backlog sweep: only meaningful under a FINITE ceiling # (a deferred tail can exist). With the default unbounded ceiling - # (respawn_limit is None) there is no deferred tail, so NO background task - # is started -- existing deployments are completely unaffected. When a - # finite ceiling IS set, this drains the deferred tail over time instead of + # (respawn_limit is None) there is no deferred tail, so the loop is not + # entered -- existing deployments are completely unaffected. When a finite + # ceiling IS set, this drains the deferred tail over time instead of # stranding it until a restart or a new event (see _crash_recovery_sweep_loop # and config.crash_recovery_sweep_interval_seconds). - _sweep_task: asyncio.Task[None] | None = None + # + # AWAITED rather than wrapped in its own create_task: this coroutine is + # already a background task that lifespan cancels on shutdown, so awaiting + # the sweep here means one task to cancel instead of two -- and no way for + # the sweep to outlive its parent. _sweep_interval = _settings.crash_recovery_sweep_interval_seconds if respawn_limit is not None and _sweep_interval > 0: - _sweep_task = asyncio.create_task( - _crash_recovery_sweep_loop(_sweep_interval, respawn_limit) - ) logger.info( "crash_recovery_sweep: enabled (interval=%ds, ceiling=%d) -- " "deferred backlog will drain progressively, not just on restart", _sweep_interval, respawn_limit, ) + await _crash_recovery_sweep_loop(_sweep_interval, respawn_limit) + + +@asynccontextmanager +async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: + """Manage application lifespan: configure logging and create shared Neo4j driver.""" + setup_logging() + _admin = _settings.resolve_neo4j_admin() + _query = _settings.resolve_neo4j_query() + logger.info( + "lifespan_startup: creating Neo4j drivers admin_url=%s query_url=%s query_access_mode=%s", + _admin.url, + _query.url, + _query.access_mode, + ) + # Admin (read/write): schema init + all mutation paths. Keep the existing + # app.state.neo4j_driver NAME so nothing that reads it silently breaks. + # 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. 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. + app.state.neo4j_query_access_mode = _query.access_mode + # Initialize schema (indexes + uniqueness constraints) BEFORE the server starts + # accepting requests. This ensures the Session uniqueness constraint is active + # before any concurrent flush() transactions execute MERGE, which prevents the + # duplicate-Session-node race condition observed under concurrent upload load. + logger.info( + "lifespan_startup: initializing Neo4j schema (indexes + uniqueness constraints)" + ) + # Cold start FAILS LOUD on schema/data corruption that requires + # `doctor --fix` -- an un-migrated graph (duplicate legacy nodes OR + # nodes lacking the universal :Node label). Nothing has been written yet + # at cold start, so refusing to boot loses no data: this is the safest + # possible moment to surface an impossible state as an un-missable + # signal rather than a log line someone greps for later. Contrast with + # the flush path (Neo4jGraphStore._ensure_schema), which must keep + # self-healing and never raise (Salil's blocker -- raising there would + # dead-letter real in-flight activity records). fail_on_data_conflict=True + # here mirrors run_repair's contract: a :Node constraint data conflict + # raises a RuntimeError naming `doctor --fix` instead of being logged + # and swallowed. + 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)." + ) + # NOTE: the O(1) untagged-:Node boot guard that used to sit here has been + # REMOVED. It refused to boot when any node lacked the universal :Node + # label. Two things made it dead weight: + # + # 1. Nothing in this service can produce an untagged node. Every + # node-creating statement in neo4j_store is label-scoped -- the node + # MERGE, both edge-endpoint MERGEs, and the session write path all + # MERGE (n:Node {node_id, workspace}). + # 2. Since get_node()/get_edge() became :Node-scoped, an untagged node is + # also INVISIBLE to every read. Read and write paths now agree, so a + # stray untagged node is inert dead data, not a correctness hazard -- + # and refusing to serve over inert data is disproportionate. + # + # The capability is not gone: `context-intelligence-server doctor` still + # reports untagged nodes via diagnose()/count_untagged_nodes, and + # `doctor --fix` still repairs them. Detection moved to the operator path, + # where it belongs; it is no longer a boot gate. (Historically this check + # was itself an AllNodesScan and caused a 25-30s boot stall until PR #67 + # made it O(1) -- a second reason not to keep it on the startup path.) + + # Crash recovery runs as a BACKGROUND task -- it MUST NOT gate serving. + # + # uvicorn runs the ASGI lifespan to completion BEFORE it handles a single + # request (uvicorn/server.py: `await self.startup()` then + # `await self.main_loop()`), and gunicorn binds the socket before that. So + # any work done here is work during which the process ACCEPTS connections + # and answers none -- /version and /status included, however cheap they + # are. Incident 2026-09-09: a ~5000-file spool on Azure Files kept this + # loop busy for 9+ minutes; ACA's tcpSocket probes saw an open port and + # reported the replica Healthy while every request timed out at the APIM + # gateway (231s), and clients dropped events from full buffers. + # + # Moving it behind create_task lets startup complete in ~1s, so the HTTP + # surface is up regardless of spool size. This is safe because respawn is + # idempotent by construction -- see _crash_recovery_topup, which is already + # documented as safe to call repeatedly on a LIVE server, and which the + # sweep loop has always called against a serving process. + app.state.recovery_complete = asyncio.Event() + app.state.recovery_error = None + _recovery_task: asyncio.Task[None] = asyncio.create_task(_startup_recovery(app)) + try: yield finally: - if _sweep_task is not None: - _sweep_task.cancel() - with suppress(asyncio.CancelledError): - await _sweep_task + # Cancel background recovery FIRST: it spawns drainers, and shutdown + # below quiesces them. Letting it keep spawning into a shutting-down + # registry would race the quiesce and dead-letter healthy events. + _recovery_task.cancel() + with suppress(asyncio.CancelledError): + await _recovery_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 @@ -929,6 +991,15 @@ async def get_status(request: Request) -> dict[str, Any]: # above. Cheap by construction (stat-only, short-TTL cached); see # QueueManager.spool_stats() for why this holds even under a huge spool. response["spool"] = await registry.queue_manager.spool_stats() + # Crash recovery now runs in the BACKGROUND so the HTTP surface opens + # immediately (see _startup_recovery). While it is still running, the + # conservation counters above have not been seeded yet, so say so rather + # than letting a caller read mid-recovery numbers as settled ones. + _recovery_evt = getattr(request.app.state, "recovery_complete", None) + response["recovery_complete"] = bool(_recovery_evt and _recovery_evt.is_set()) + # "complete" alone would be a lie by omission when recovery FAILED: the + # event is set either way (see _startup_recovery's finally). + response["recovery_error"] = getattr(request.app.state, "recovery_error", None) # Surface auth mode and admin-API capability so operators can confirm # admin is enabled without tailing startup logs. /status is # unauthenticated — only config-level boolean flags are exposed here diff --git a/pyproject.toml b/pyproject.toml index 8dd6420..ab01119 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "context-intelligence-server" -version = "6.7.2" +version = "6.7.3" description = "Context Intelligence Server for Amplifier" requires-python = ">=3.11" dependencies = [ diff --git a/tests/neo4j/test_node_identity_migration.py b/tests/neo4j/test_node_identity_migration.py index 2358d0d..88501ad 100644 --- a/tests/neo4j/test_node_identity_migration.py +++ b/tests/neo4j/test_node_identity_migration.py @@ -289,8 +289,19 @@ async def _run_migration_assertions(neo4j_container: dict[str, Any]) -> None: async def test_cold_start_guard_detects_untagged_only_graph( neo4j_container: dict[str, Any], ) -> None: - """Reproduces main.py's lifespan cold-start guard, against a REAL Neo4j, - for the untagged-only shape the :Node constraint alone CANNOT see. + """Exercises ``count_untagged_nodes`` and ``ensure_neo4j_schema`` against + a REAL Neo4j, for the untagged-only shape the :Node constraint alone + CANNOT see. + + NOTE: main.py's ``lifespan()`` no longer calls ``count_untagged_nodes`` + at all -- the O(1) untagged-:Node boot guard this test used to describe + was REMOVED (nothing in this service can produce an untagged node, and + since get_node()/get_edge() became :Node-scoped, an untagged node is + inert dead data rather than a correctness hazard; detection moved + entirely to the `doctor` operator path). This test does not call + ``lifespan()`` and its assertions are unaffected by that removal -- it + verifies the two underlying primitives directly, still used by + ``doctor``/``run_repair``. Uses ``_seed_untagged_only_graph`` (NOT ``_seed_dirty_graph``, whose duplicate ``dup-1`` :Event nodes trip the separate, always fail-open @@ -298,22 +309,19 @@ async def test_cold_start_guard_detects_untagged_only_graph( ``test_run_repair_dedups_backfills_and_constrains``). Every node seeded here has a unique (node_id, workspace) and no label collision, so NO uniqueness constraint (Session/Event/Node) sees a conflict -- the ONLY - defect is the missing ``:Node`` label, which is exactly why the lifespan - guard needs its second, independent check (``count_untagged_nodes``): - the constraint step provides no signal for this case on its own. + defect is the missing ``:Node`` label, which is exactly why + ``count_untagged_nodes`` exists as a second, independent check: the + constraint step provides no signal for this case on its own. - Reproduces the lifespan's two ordered steps directly against the live - container (the guard logic is inline in ``main.py``'s ``lifespan()``, - not its own importable function): + Exercises the two primitives directly against the live container: 1. ``ensure_neo4j_schema(driver, fail_on_data_conflict=True)`` -- succeeds (``True``), no constraint conflict to raise on. 2. ``count_untagged_nodes(driver)`` -- reports > 0. - Together, (1) succeeding and (2) being > 0 is precisely the condition - under which ``lifespan()`` raises ``RuntimeError`` naming - ``doctor --fix`` -- i.e. the un-migrated (untagged-only) graph IS - detected and cold start WOULD refuse to boot. + Together, (1) succeeding and (2) being > 0 is the un-migrated + (untagged-only) graph signature ``doctor``/``run_repair`` uses to detect + and repair this shape -- no longer a boot-time condition. """ _wipe(neo4j_container) try: @@ -324,10 +332,10 @@ async def test_cold_start_guard_detects_untagged_only_graph( auth=(neo4j_container["user"], neo4j_container["password"]), ) try: - # Step 1 (lifespan): fail-loud schema init does NOT raise here -- - # none of the seeded nodes carry :Node (or collide under any - # OTHER constraint), so no constraint sees a conflict. This is - # the case the constraint check alone misses. + # Step 1: fail-loud schema init does NOT raise here -- none of + # the seeded nodes carry :Node (or collide under any OTHER + # constraint), so no constraint sees a conflict. This is the + # case the constraint check alone misses. established = await ensure_neo4j_schema(driver, fail_on_data_conflict=True) assert established is True, ( "ensure_neo4j_schema(fail_on_data_conflict=True) must succeed " @@ -335,13 +343,14 @@ async def test_cold_start_guard_detects_untagged_only_graph( "dirty graph -- there is no constraint conflict to raise on." ) - # Step 2 (lifespan): the O(1) untagged guard DOES catch it. + # Step 2: the O(1) untagged counter DOES catch it (no longer + # called by lifespan() -- see the doctor operator path). untagged = await count_untagged_nodes(driver) assert untagged > 0, ( "count_untagged_nodes must report the seeded untagged legacy " - "nodes (legacy-sess-only + legacy-bare) -- this is the exact " - "signal main.py's lifespan() uses to raise RuntimeError and " - "refuse to boot on an un-migrated graph." + "nodes (legacy-sess-only + legacy-bare) -- this is the " + "signal `doctor`/`run_repair` use to detect and repair an " + "un-migrated graph." ) finally: await driver.close() diff --git a/tests/test_main.py b/tests/test_main.py index a5ab80a..bcc17cc 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -797,6 +797,196 @@ def _patched_lifespan_deps() -> Any: return mock_driver +# --------------------------------------------------------------------------- +# Background recovery task: startup must not block HTTP (fix/startup-must- +# not-block-http) +# --------------------------------------------------------------------------- + + +async def test_status_and_version_respond_while_recovery_is_still_running( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """THE load-bearing regression test: HTTP must be answerable while the + background recovery task is still running. + + uvicorn runs the ASGI lifespan to completion BEFORE it starts + main_loop (i.e. before it accepts/answers a single request), so any + startup work that runs INLINE in lifespan means the process accepts TCP + connections and answers nothing until that work finishes -- this is + exactly the 2026-09-09 incident (a ~5000-file spool kept lifespan busy + for 9+ minutes while ACA's tcpSocket probe reported the replica healthy). + Moving recovery behind ``asyncio.create_task`` lets lifespan yield + almost immediately, so /version and /status must respond -- and report + recovery as still in progress -- while recovery is deliberately held + open by a test-controlled gate. + + Against the OLD inline-recovery code this test CANNOT pass. Verified + directly (git stash on context_intelligence_server/main.py only): the + pre-fix module has no ``_startup_recovery_body`` at all -- recovery ran + fully inline inside ``lifespan`` -- so the monkeypatch.setattr below + fails fast with AttributeError. Had that seam existed but still run + synchronously before the ``yield``, this test would instead hang + forever (lifespan never yields while the gate is unset) until the + suite's timeout fired. Either way, the old code cannot make this test + pass -- that failure IS the regression this test guards against. + """ + recovery_gate = asyncio.Event() + + async def _blocked_recovery(app: Any) -> None: + await recovery_gate.wait() + + monkeypatch.setattr( + main_module, + "_startup_recovery_body", + AsyncMock(side_effect=_blocked_recovery), + ) + + mock_driver = _patched_lifespan_deps() + with ( + patch("context_intelligence_server.main.setup_logging"), + patch( + "context_intelligence_server.neo4j_store.AsyncGraphDatabase.driver", + return_value=mock_driver, + ), + patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), + ): + async with lifespan(main_module.app): + # Recovery is genuinely still running: lifespan already yielded + # without waiting for it. + assert main_module.app.state.recovery_complete.is_set() is False + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=main_module.app), + base_url="http://test", + ) as c: + version_response = await c.get("/version") + assert version_response.status_code == 200 + assert "version" in version_response.json() + + status_response = await c.get("/status") + assert status_response.status_code == 200 + assert status_response.json()["recovery_complete"] is False + + # Release the gate and wait for the background pass to finish. + recovery_gate.set() + await asyncio.wait_for( + main_module.app.state.recovery_complete.wait(), timeout=5 + ) + + settled_response = await c.get("/status") + assert settled_response.status_code == 200 + settled_data = settled_response.json() + assert settled_data["recovery_complete"] is True + assert settled_data["recovery_error"] is None + + +async def test_recovery_failure_is_logged_and_does_not_take_down_the_server( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """A failure inside the background recovery pass must be logged and + recorded on app.state -- never swallowed. + + As a bare background task (no ``_startup_recovery`` wrapper) an + unhandled exception here would vanish until garbage collection ("Task + exception was never retrieved"), leaving the conservation counters + unseeded and no drainers respawned behind a server that otherwise looks + perfectly healthy. On the OLD inline path the same failure aborted boot + loudly instead. The wrapper must catch it, log it, and stash it on + ``app.state.recovery_error`` so /status can surface it too. + """ + boom = RuntimeError("boom-test-recovery-failure") + monkeypatch.setattr( + main_module, "_startup_recovery_body", AsyncMock(side_effect=boom) + ) + + mock_driver = _patched_lifespan_deps() + with ( + patch("context_intelligence_server.main.setup_logging"), + patch( + "context_intelligence_server.neo4j_store.AsyncGraphDatabase.driver", + return_value=mock_driver, + ), + patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), + caplog.at_level(logging.ERROR, logger="context_intelligence_server"), + ): + async with lifespan(main_module.app): + await asyncio.wait_for( + main_module.app.state.recovery_complete.wait(), timeout=5 + ) + assert "boom-test-recovery-failure" in main_module.app.state.recovery_error + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=main_module.app), + base_url="http://test", + ) as c: + response = await c.get("/status") + assert response.status_code == 200 + assert "boom-test-recovery-failure" in response.json()["recovery_error"] + + error_records = [r for r in caplog.records if r.levelno == logging.ERROR] + assert any("startup_recovery" in r.getMessage() for r in error_records), ( + f"expected an ERROR log mentioning startup_recovery; " + f"got {[r.getMessage() for r in error_records]}" + ) + + +async def test_recovery_task_is_cancelled_on_shutdown() -> None: + """Shutdown must cancel the background recovery task -- not leave it running. + + Recovery spawns drainers and shutdown quiesces them (see + ``test_lifespan_quiesces_drain_workers_before_closing_shared_driver``); a + recovery task that survives lifespan exit would keep racing that + quiesce, spawning new drainers into a registry that is being torn down + and risking dead-lettered healthy events. + """ + never_set = asyncio.Event() + + async def _blocked_forever(app: Any) -> None: + await never_set.wait() + + mock_driver = _patched_lifespan_deps() + with ( + patch("context_intelligence_server.main.setup_logging"), + patch( + "context_intelligence_server.neo4j_store.AsyncGraphDatabase.driver", + return_value=mock_driver, + ), + patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), + patch( + "context_intelligence_server.main._startup_recovery_body", + new=AsyncMock(side_effect=_blocked_forever), + ), + ): + async with lifespan(main_module.app): + # Let the recovery task actually get scheduled and start blocking. + await asyncio.sleep(0) + assert main_module.app.state.recovery_complete.is_set() is False + + leftover = [ + t + for t in asyncio.all_tasks() + if not t.done() and "_startup_recovery" in repr(t) + ] + assert leftover == [], f"recovery task(s) still pending after shutdown: {leftover}" + + +async def test_status_reports_recovery_state_before_lifespan_has_run( + client: httpx.AsyncClient, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """/status must degrade gracefully -- not 500 -- when app.state has never + had recovery_complete/recovery_error set (lifespan hasn't run yet).""" + monkeypatch.delattr(main_module.app.state, "recovery_complete", raising=False) + monkeypatch.delattr(main_module.app.state, "recovery_error", raising=False) + + response = await client.get("/status") + assert response.status_code == 200 + data = response.json() + assert data["recovery_complete"] is False + assert data["recovery_error"] is None + + async def test_lifespan_recovers_and_respawns_drainers( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -833,7 +1023,11 @@ async def test_lifespan_recovers_and_respawns_drainers( ), ): async with lifespan(main_module.app): - pass + # Recovery now runs in a background task; wait for the one-shot + # pass to finish before asserting on its effects. + await asyncio.wait_for( + main_module.app.state.recovery_complete.wait(), timeout=5 + ) assert (sid, "/recovered-ws") in spawned @@ -925,7 +1119,9 @@ async def test_lifespan_default_respawns_all_recovered_sessions_unbounded( patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), ): async with lifespan(main_module.app): - pass + await asyncio.wait_for( + main_module.app.state.recovery_complete.wait(), timeout=5 + ) assert {s for s, _w in spawned} == set(sids) @@ -959,7 +1155,9 @@ async def test_lifespan_respawn_cap_defers_remainder_and_logs_warning( caplog.at_level(logging.WARNING, logger="context_intelligence_server"), ): async with lifespan(main_module.app): - pass + await asyncio.wait_for( + main_module.app.state.recovery_complete.wait(), timeout=5 + ) # Exactly the cap's worth of sessions were respawned -- never more. assert len(spawned) == 2 @@ -1004,7 +1202,9 @@ async def test_lifespan_deferred_sessions_untouched_and_recoverable_next_boot( patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), ): async with lifespan(main_module.app): - pass + await asyncio.wait_for( + main_module.app.state.recovery_complete.wait(), timeout=5 + ) assert len(spawned_boot1) == 1 deferred_sids = set(sids) - {s for s, _w in spawned_boot1} @@ -1116,7 +1316,14 @@ async def test_lifespan_enables_sweep_under_finite_limit( caplog.at_level(logging.INFO, logger="context_intelligence_server"), ): async with lifespan(main_module.app): - pass # task is created on entry and cancelled cleanly on exit + # The sweep is awaited inside the same background task, entered + # right after recovery_complete is set -- waiting for the event + # is enough for the "enabled" log line to have already fired by + # the time we resume (it's emitted synchronously before the + # sweep loop's first suspending await). + await asyncio.wait_for( + main_module.app.state.recovery_complete.wait(), timeout=5 + ) assert any( "crash_recovery_sweep: enabled" in r.getMessage() for r in caplog.records @@ -1183,15 +1390,16 @@ async def test_lifespan_no_sweep_when_interval_zero( # --------------------------------------------------------------------------- # Cold start FAILS LOUD on schema/data corruption that requires `doctor # --fix` (design decision, reversing the lifespan half of f4d8bab): an -# un-migrated graph -- duplicate legacy nodes (caught by the :Node -# constraint via fail_on_data_conflict=True) OR nodes lacking the :Node -# label altogether (caught by the O(1) count_untagged_nodes guard) -- must -# refuse to boot. Nothing has been written yet at cold start, so refusing to -# boot loses no data. The migration itself still lives ONLY in `doctor -# --fix` (run_repair); the flush path still self-heals (see -# tests/neo4j/test_node_identity_migration.py). A connectivity/probe -# failure (graph unreachable) is NOT treated as "confirmed un-migrated" and -# must not crash boot. +# un-migrated graph with duplicate legacy nodes (caught by the :Node +# constraint via fail_on_data_conflict=True) must refuse to boot. Nothing +# has been written yet at cold start, so refusing to boot loses no data. +# The migration itself still lives ONLY in `doctor --fix` (run_repair); the +# flush path still self-heals (see tests/neo4j/test_node_identity_migration.py). +# +# The O(1) untagged-:Node boot guard that used to sit alongside this check +# has been REMOVED (see test_lifespan_does_not_gate_boot_on_untagged_nodes +# below for why) -- detection of untagged nodes moved entirely to the +# `doctor` operator path. # --------------------------------------------------------------------------- @@ -1212,10 +1420,6 @@ async def test_lifespan_calls_ensure_schema_with_fail_on_data_conflict() -> None "context_intelligence_server.main.ensure_neo4j_schema", new=mock_ensure_schema, ), - patch( - "context_intelligence_server.main.count_untagged_nodes", - new=AsyncMock(return_value=0), - ), ): async with lifespan(main_module.app): pass @@ -1254,68 +1458,32 @@ async def test_lifespan_raises_on_ensure_schema_data_conflict() -> None: pass -async def test_lifespan_raises_on_untagged_nodes() -> None: - """On an un-migrated graph (untagged :Node count > 0), startup MUST - raise a RuntimeError naming `doctor --fix` -- boot refuses to start - rather than silently risking write-path duplication.""" - mock_driver = _patched_lifespan_deps() - with ( - patch("context_intelligence_server.main.setup_logging"), - patch( - "context_intelligence_server.neo4j_store.AsyncGraphDatabase.driver", - return_value=mock_driver, - ), - patch( - "context_intelligence_server.main.ensure_neo4j_schema", - new=AsyncMock(return_value=True), - ), - patch( - "context_intelligence_server.main.count_untagged_nodes", - new=AsyncMock(return_value=42), - ), - pytest.raises(RuntimeError, match="doctor --fix") as exc_info, - ): - async with lifespan(main_module.app): - pass +async def test_lifespan_does_not_gate_boot_on_untagged_nodes() -> None: + """The O(1) untagged-:Node boot guard has been REMOVED entirely -- + startup no longer counts or checks untagged nodes at all, and no longer + raises when they exist. - assert "42" in str(exc_info.value), ( - f"Expected the untagged count in the error message, got: {exc_info.value}" - ) + Two things made the guard dead weight: (1) nothing in this service can + produce an untagged node -- every node-creating statement in + neo4j_store is a label-scoped MERGE (n:Node {node_id, workspace}); and + (2) since get_node()/get_edge() became :Node-scoped, an untagged node is + also invisible to every read, so a stray one is inert dead data, not a + correctness hazard -- refusing to serve over it would be disproportionate. + The capability is not gone: `context-intelligence-server doctor` still + reports and repairs untagged nodes via diagnose()/count_untagged_nodes; + detection simply moved to the operator path and is no longer a boot gate. -async def test_lifespan_does_not_raise_on_clean_graph() -> None: - """On a fully-migrated graph (untagged count == 0, no constraint - conflict), startup does NOT raise -- the fail-loud guards are silent - when there is nothing to report.""" - mock_driver = _patched_lifespan_deps() - with ( - patch("context_intelligence_server.main.setup_logging"), - patch( - "context_intelligence_server.neo4j_store.AsyncGraphDatabase.driver", - return_value=mock_driver, - ), - patch( - "context_intelligence_server.main.ensure_neo4j_schema", - new=AsyncMock(return_value=True), - ), - patch( - "context_intelligence_server.main.count_untagged_nodes", - new=AsyncMock(return_value=0), - ) as mock_count, - ): - async with lifespan(main_module.app): - pass - - mock_count.assert_awaited_once() - + This test asserts the removal is genuine (not just untested): boot + succeeds with no untagged-count patching of any kind, and the function + main.py used to import for this guard no longer exists on the module. + """ + assert not hasattr(main_module, "count_untagged_nodes"), ( + "count_untagged_nodes must no longer be imported into main -- the " + "untagged-node boot guard was removed; detection lives only on the " + "doctor operator path now." + ) -async def test_lifespan_does_not_raise_when_health_check_itself_fails( - caplog: pytest.LogCaptureFixture, -) -> None: - """A health-check probe failure (e.g. count_untagged_nodes raising due to - a transient connectivity blip) must NOT be treated as a confirmed - un-migrated graph -- it is logged at DEBUG and swallowed, and boot - proceeds. Connectivity failure != confirmed data corruption.""" mock_driver = _patched_lifespan_deps() with ( patch("context_intelligence_server.main.setup_logging"), @@ -1327,11 +1495,6 @@ async def test_lifespan_does_not_raise_when_health_check_itself_fails( "context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock(return_value=True), ), - patch( - "context_intelligence_server.main.count_untagged_nodes", - new=AsyncMock(side_effect=RuntimeError("transient connectivity blip")), - ), - caplog.at_level(logging.DEBUG), ): async with lifespan(main_module.app): # MUST NOT raise pass