fix: /status must never walk the spool — the second unbounded scan PR #101 missed

context_intelligence_server/main.py

@@ -431,7 +431,12 @@ async def _spool_stats_refresher(app: FastAPI) -> None:
     """
     while True:
         try:
+            # Both disk-derived /status blocks refresh here, together: the
+            # spool footprint AND the queue/dead aggregate. Either one left on
+            # the request path re-creates the timeout this loop exists to
+            # prevent -- PR #101 moved only the first and /status stayed down.
             await registry.queue_manager.refresh_spool_stats()
+            await registry.queue_manager.refresh_all_stats()
         except asyncio.CancelledError:
             raise
         except Exception:

context_intelligence_server/queue_manager.py

@@ -139,7 +139,6 @@ def __init__(self, queues_dir: Path):
         self._dir.mkdir(parents=True, exist_ok=True)
         self._stats_cache: dict[str, Any] | None = None
         self._stats_cache_at: float = 0.0
-        self._stats_cache_ttl: float = 1.0
         # Snapshot for spool_stats() / spool_stats_age_seconds(). Written
         # ONLY by refresh_spool_stats(), which a background loop
         # (_spool_stats_refresher in main.py) calls on its own cadence
@@ -678,20 +677,29 @@ def _all_worker_keys(self) -> list[str]:
             keys.add(dead.name[: -len(".dead.jsonl")])
         return sorted(keys)
 
-    async def derive_all_stats(self) -> dict[str, Any]:
-        """Derive live queue stats purely from disk, with a short TTL cache.
+    async def refresh_all_stats(self) -> dict[str, Any]:
+        """Scan the spool and refresh the snapshot ``derive_all_stats()`` serves.
 
         Aggregates per-worker ``in_queue`` (complete uncommitted lines) and
         ``dead`` (dead-letter records). ``in_queue`` is a tail read from the
-        committed offset to EOF (the whole file is never read); results are
-        cached for ``_stats_cache_ttl`` seconds since ``/status`` polls often.
+        committed offset to EOF (the whole file is never read).
+
+        THE ONLY thing that touches disk for these figures, and it is called
+        exclusively by the background refresher (``_spool_stats_refresher`` in
+        main.py) -- NEVER from a request handler.
+
+        INCIDENT 2026-09-10: this scan used to run INLINE on every ``/status``
+        call, gated only by a 1-second read-side TTL. A read-side TTL spares
+        the SECOND caller and never the first, so on team-shared's ~5000-key
+        spool every poll past the TTL paid a full per-key walk -- an
+        ``open+read`` of each ``.offset``, a streamed newline count of each
+        undrained tail, and a read of each ``.dead.jsonl``, all over Azure
+        Files SMB. ``/status`` timed out at 60s/120s/150s/180s while
+        ``/version`` answered in 0s and ``POST /events`` in 1s on the SAME
+        replica. PR #101 fixed the identical flaw in ``spool_stats`` and
+        MISSED this second walk in the same handler.
         """
         now = time.monotonic()
-        if (
-            self._stats_cache is not None
-            and (now - self._stats_cache_at) < self._stats_cache_ttl
-        ):
-            return self._stats_cache
 
         def _all() -> dict[str, Any]:
             per_key: list[dict[str, Any]] = []
@@ -729,6 +737,41 @@ def _all() -> dict[str, Any]:
         self._stats_cache_at = now
         return stats
 
+    def derive_all_stats(self) -> dict[str, Any]:
+        """Return the last snapshot from ``refresh_all_stats()``. NEVER scans.
+
+        Synchronous, pure cache read, no filesystem access, never raises --
+        so ``/status``'s cost is O(1) no matter how large the spool is. See
+        ``refresh_all_stats`` for the incident that required this split.
+
+        Before the first refresh completes there is no snapshot. Rather than
+        scan (which is the whole bug) or invent numbers, this returns
+        ``stats_available: False`` with zeroed aggregates, and
+        ``Registry.pipeline_metrics`` uses that flag to SUPPRESS its
+        residual/degraded determination for that window -- otherwise
+        ``residual = accepted - written - in_queue - dead`` would read the
+        absent in_queue/dead as 0 and report a fabricated loss. A hanging
+        endpoint traded for a lying one is not a fix.
+        """
+        if self._stats_cache is None:
+            return {
+                "per_key": [],
+                "in_queue_total": 0,
+                "dead_total": 0,
+                "stats_available": False,
+            }
+        return {**self._stats_cache, "stats_available": True}
+
+    def all_stats_age_seconds(self) -> float | None:
+        """Seconds since ``refresh_all_stats()`` last produced a snapshot.
+
+        ``None`` when it never has. Lets ``/status`` state the snapshot's
+        staleness instead of presenting it as live.
+        """
+        if self._stats_cache is None:
+            return None
+        return max(0.0, time.monotonic() - self._stats_cache_at)
+
     async def refresh_spool_stats(self) -> dict[str, int]:
         """Scan the spool directory and refresh the snapshot ``spool_stats()`` serves.
 

context_intelligence_server/registry.py

@@ -255,18 +255,30 @@ async def pipeline_metrics(self) -> dict[str, Any]:
         deleted, but the in-memory counters persist across restarts via
         `seed_counters`.
         """
-        agg = await self.queue_manager.derive_all_stats()
+        # Cache-only read: NEVER scans the spool. See
+        # QueueManager.refresh_all_stats for the incident that required this
+        # (an inline per-key walk here made /status time out at 60-180s on a
+        # ~5000-key spool while /version answered in 0s on the same replica).
+        agg = self.queue_manager.derive_all_stats()
         counters = self.pipeline_counters()
         in_queue = agg["in_queue_total"]
         dead = agg["dead_total"]
+        # False only in the window before the background refresher has produced
+        # its first snapshot. in_queue/dead are then 0 because they are UNKNOWN,
+        # not because they are zero -- so the residual below would be inflated by
+        # exactly the backlog we cannot see yet.
+        stats_available = agg.get("stats_available", True)
         residual = (
             counters["accepted_total"] - counters["written_total"] - in_queue - dead
         )
         # A NEGATIVE residual is never data loss: written+in_queue+dead cannot
         # legitimately exceed accepted, so residual<0 is purely a sampling skew
         # between the fresh counters and the cached disk snapshot. Clamp the
         # loss signal at zero -- only a positive residual can mean real loss.
-        lost = max(0, residual)
+        # Suppress the loss signal entirely while the aggregates are unknown:
+        # reporting a fabricated residual (and latching _residual_positive_since
+        # from it) would trade a hanging endpoint for a lying one.
+        lost = max(0, residual) if stats_available else 0
         now = time.monotonic()
         if lost > 0:
             if self._residual_positive_since is None:
@@ -280,16 +292,24 @@ async def pipeline_metrics(self) -> dict[str, Any]:
         # dead>0 is an accounted-for loss and is degraded immediately (no grace).
         # A positive residual is degraded only once it has PERSISTED past the
         # grace window -- transient in-flight skew clears before then.
-        degraded = dead > 0 or sustained
+        # dead>0 cannot be trusted before the first snapshot either -- it is 0
+        # by absence, not by measurement -- so degraded stays False until the
+        # refresher has actually looked.
+        degraded = stats_available and (dead > 0 or sustained)
         return {
             "accepted_total": counters["accepted_total"],
             "written_total": counters["written_total"],
             "replayed_total": counters["replayed_total"],
             "write_retries_total": counters["write_retries_total"],
             "in_queue_total": in_queue,
             "dead_letter_total": dead,
-            "residual": residual,
+            "residual": residual if stats_available else None,
             "degraded": degraded,
+            # Callers must be able to tell "measured zero" from "not yet
+            # measured"; as_of_seconds carries the snapshot's age (None before
+            # the first refresh), mirroring /status's spool.as_of_seconds.
+            "stats_available": stats_available,
+            "as_of_seconds": self.queue_manager.all_stats_age_seconds(),
         }
 
     async def _process_one(

pyproject.toml

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

tests/neo4j/test_queues_actions.py

@@ -173,6 +173,10 @@ async def test_replay_rewrites_through_real_drainer(
     #    ingest of the replayed line is accounted for (accepted=1), then the
     #    drainer's write balances it. residual == 0, degraded False, no dead.
     reg.seed_counters(accepted=1, written=0)
+    # pipeline_metrics() is a CACHE-ONLY read now (the inline per-key walk it
+    # used to do made /status time out at 60-180s on a ~5000-key spool); the
+    # background refresher owns the scan, so drive it explicitly here.
+    await reg.queue_manager.refresh_all_stats()
     metrics = await reg.pipeline_metrics()
     assert metrics["dead_letter_total"] == 0, metrics
     assert metrics["residual"] == 0, metrics

tests/test_lazy_asgi_app.py

@@ -38,10 +38,48 @@
 PROJECT_ROOT = Path(__file__).parent.parent
 
 
+# Wall-clock budget for a subprocess that is expected to exit almost
+# immediately (`--help`, `--version`, an import). Generous already: these
+# never reach gunicorn.
+_FAST_CLI_TIMEOUT_S = 15
+
+# Budget for a `serve` invocation that must fail its startup guard. This one
+# pays for spawn + a cold import of the whole server package (fastapi, neo4j,
+# pydantic) + gunicorn arbiter start + fork + worker load() before the guard
+# can raise at all -- and it pays it on a shared CI runner with a cold page
+# cache, not on a warm dev box.
+#
+# CI FLAKE 2026-09-10 (run 34419935174, main @ 7c021bc): this test tripped a
+# 15s timeout and was reported as a guard regression. It was not. The same
+# tree had passed CI twice on its branch (e58bbcd, 1f60669), and locally the
+# guard fires in 0.74s with returncode 3 and a full traceback. The subprocess
+# had only reached "Starting gunicorn" -- it never got to worker boot, so the
+# guard never had the chance to fire. A tight wall-clock budget on a cold
+# runner is not evidence about the guard.
+#
+# Raising this costs NOTHING on a healthy run (the process exits in under a
+# second) and only spends real time when something is genuinely wrong -- which
+# is exactly when you want to wait and capture the output rather than SIGKILL
+# it and lose the buffered stderr.
+_SERVE_CLI_TIMEOUT_S = 90
+
+
 def _run_cli(
-    args: list[str], *, cwd: Path, env: dict[str, str]
+    args: list[str],
+    *,
+    cwd: Path,
+    env: dict[str, str],
+    timeout: int = _FAST_CLI_TIMEOUT_S,
 ) -> subprocess.CompletedProcess:
-    """Invoke the CLI's main() in a fresh subprocess (genuinely fresh import)."""
+    """Invoke the CLI's main() in a fresh subprocess (genuinely fresh import).
+
+    On timeout, re-raises with the partial stdout/stderr attached to the
+    message. A bare ``TimeoutExpired`` says only "15 seconds passed", which
+    tells the next reader nothing about WHERE the process was stuck -- and the
+    SIGKILL discards whatever the child had buffered. Surfacing the partial
+    output is the difference between "the guard regressed" and "the runner was
+    slow and never reached worker boot".
+    """
     code = (
         "import sys\n"
         "from context_intelligence_server.main import main\n"
@@ -50,15 +88,25 @@ def _run_cli(
         "except SystemExit as e:\n"
         "    sys.exit(e.code if e.code is not None else 0)\n"
     )
-    return subprocess.run(
-        [sys.executable, "-c", code, *args],
-        cwd=str(cwd),
-        env=env,
-        capture_output=True,
-        text=True,
-        timeout=15,
-        check=False,
-    )
+    try:
+        return subprocess.run(
+            [sys.executable, "-c", code, *args],
+            cwd=str(cwd),
+            env=env,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+            check=False,
+        )
+    except subprocess.TimeoutExpired as exc:
+        raise AssertionError(
+            f"CLI subprocess {args!r} did not exit within {timeout}s.\n"
+            f"This is a WALL-CLOCK timeout, NOT proof that a guard failed to "
+            f"fire -- check the partial output below for how far it got "
+            f"before it was killed.\n"
+            f"--- partial stdout ---\n{exc.stdout or '<empty>'}\n"
+            f"--- partial stderr ---\n{exc.stderr or '<empty>'}"
+        ) from exc
 
 
 def _clean_env(tmp_path: Path) -> dict[str, str]:
@@ -190,7 +238,9 @@ def test_serve_still_fails_loud_for_a_real_misconfiguration(
         # construction (inside the worker's load()), before ever binding
         # matters for the assertion.
         env["AMPLIFIER_CONTEXT_INTELLIGENCE_SERVER_SERVER_PORT"] = "18321"
-        result = _run_cli(["serve"], cwd=PROJECT_ROOT, env=env)
+        result = _run_cli(
+            ["serve"], cwd=PROJECT_ROOT, env=env, timeout=_SERVE_CLI_TIMEOUT_S
+        )
         assert result.returncode != 0
         assert "neo4j_require_explicit_clients" in result.stdout + result.stderr
 

tests/test_main.py

@@ -1708,6 +1708,7 @@ async def test_lifespan_seeds_counters_from_disk(tmp_path: Path) -> None:
     accepted_seed, written_seed = await reg.queue_manager.recovery_seed_counts()
     reg.seed_counters(accepted_seed, written_seed)
 
+    await reg.queue_manager.refresh_all_stats()
     metrics = await reg.pipeline_metrics()
     assert metrics["accepted_total"] == 2
     assert metrics["written_total"] == 1