fix: /status off the spool walk, and trim drained queue files as they are processed

context_intelligence_server/config.py

@@ -911,6 +911,31 @@ def _validate_crash_recovery_sweep_interval(cls, v: int) -> int:
             )
         return v
 
+    # How often the background loop (_spool_stats_refresher in main.py) walks
+    # the queue directory to refresh the snapshot QueueManager.spool_stats()
+    # serves to /status. Exists because that scan is NOT cheap enough to run
+    # on the request path: incident 2026-09-09 -- /status called the
+    # equivalent of today's refresh_spool_stats() inline, per request, and on
+    # a ~5000-file spool on Azure Files SMB that meant a stat() round trip
+    # per file (plus an open+read+close of every session's .offset), taking
+    # minutes while /version on the same server returned in under a second.
+    # Moving the scan to a periodic background refresh makes /status's read
+    # O(1) regardless of spool size; this setting controls how stale that
+    # snapshot is allowed to get between refreshes (surfaced honestly via
+    # /status's spool.as_of_seconds, never presented as though it were live).
+    spool_stats_refresh_interval_seconds: float = 60.0
+
+    @field_validator("spool_stats_refresh_interval_seconds")
+    @classmethod
+    def _validate_spool_stats_refresh_interval(cls, v: float) -> float:
+        """Fail loud on a non-positive interval; there is no "disabled" value."""
+        if v <= 0:
+            raise ValueError(
+                "spool_stats_refresh_interval_seconds must be a positive "
+                f"number of seconds, got {v}"
+            )
+        return v
+
     # -------------------------------------------------------------------------
     # Logging
     # -------------------------------------------------------------------------

context_intelligence_server/main.py

@@ -355,6 +355,34 @@ async def _startup_recovery_body(app: FastAPI) -> None:
         respawned,
         len(recovered),
     )
+    # Reclaim orphaned drained files. The drain loop's per-session trim only
+    # fires for a session that HAS a live worker, and crash recovery respawns a
+    # worker only for a session with UNDRAINED data -- so a session that fully
+    # drained and then went away keeps its .log/.offset forever and is re-walked
+    # by every later boot's recovery. Measured 2026-09-09 on a real spool: 25
+    # fully-drained logs, ~1.7 GiB, including a 691 MB and a 600 MB file, none
+    # of which the per-session trim could ever see.
+    #
+    # Runs AFTER recovery_seed_counts (above) so the conservation baseline is
+    # read from disk before anything is removed, and after the respawn loop so
+    # a session that DOES have pending data already has its drainer. Safe by
+    # delegation: delete_drained refuses while any uncommitted byte remains, so
+    # this can never take a log a drainer still needs. Never raises.
+    try:
+        (
+            _reclaimed_keys,
+            _reclaimed_bytes,
+        ) = await registry.queue_manager.reclaim_drained_orphans()
+    except Exception:  # noqa: BLE001 - disk reclaim must never fail a boot
+        logger.exception("startup_recovery: orphan reclaim failed; continuing")
+    else:
+        if _reclaimed_keys:
+            logger.info(
+                "startup_recovery: reclaimed %d fully-drained session file(s), "
+                "%d byte(s) of spool",
+                _reclaimed_keys,
+                _reclaimed_bytes,
+            )
     # 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
@@ -385,6 +413,36 @@ async def _startup_recovery_body(app: FastAPI) -> None:
         await _crash_recovery_sweep_loop(_sweep_interval, respawn_limit)
 
 
+async def _spool_stats_refresher(app: FastAPI) -> None:
+    """Periodically refresh the spool snapshot ``/status`` reads.
+
+    Started by ``lifespan`` as a background task, same as ``_startup_recovery``
+    -- this is the ONLY place that calls ``QueueManager.refresh_spool_stats()``
+    (the actual directory scan). ``/status`` (via
+    ``QueueManager.spool_stats()``) only ever reads the snapshot this loop
+    produces, so its response time no longer depends on spool size (incident
+    2026-09-09: an inline scan of a ~5000-file Azure Files SMB spool made
+    ``/status`` time out at 180s while ``/version`` answered instantly).
+
+    A single failed refresh must not kill this loop -- the snapshot simply
+    goes stale, which ``/status``'s ``spool.as_of_seconds`` field makes
+    honestly visible, rather than the health probe losing spool visibility
+    outright. CancelledError (shutdown) always propagates.
+    """
+    while True:
+        try:
+            await registry.queue_manager.refresh_spool_stats()
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            # A failed refresh must not kill the loop -- the snapshot simply
+            # goes stale (see spool.as_of_seconds); log and retry next tick.
+            logger.exception(
+                "spool_stats_refresh_failed: snapshot is now stale; will retry"
+            )
+        await asyncio.sleep(_settings.spool_stats_refresh_interval_seconds)
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     """Manage application lifespan: configure logging and create shared Neo4j driver."""
@@ -498,6 +556,13 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     app.state.recovery_complete = asyncio.Event()
     app.state.recovery_error = None
     _recovery_task: asyncio.Task[None] = asyncio.create_task(_startup_recovery(app))
+    # Background spool-stats refresher (see _spool_stats_refresher): same
+    # create_task/cancel-on-shutdown pattern as _recovery_task above, so
+    # /status's spool block is served from a periodically-refreshed snapshot
+    # instead of scanning the queue directory inline on every request.
+    _spool_stats_task: asyncio.Task[None] = asyncio.create_task(
+        _spool_stats_refresher(app)
+    )
 
     try:
         yield
@@ -508,6 +573,12 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         _recovery_task.cancel()
         with suppress(asyncio.CancelledError):
             await _recovery_task
+        # Cancel the spool-stats refresher too -- it holds no drainer-related
+        # state, but it must not keep touching a queue directory that a
+        # shutting-down process may be tearing down around it.
+        _spool_stats_task.cancel()
+        with suppress(asyncio.CancelledError):
+            await _spool_stats_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
@@ -986,11 +1057,16 @@ async def get_status(request: Request) -> dict[str, Any]:
     # this block must NOT carry the per-key table or the dead-letter
     # listing — both are authenticated-only.
     response["metrics"] = await registry.pipeline_metrics()
-    # Aggregate-only spool footprint: two integers only, no session ids, no
-    # workspace names, no per-key table -- same privacy contract as `metrics`
-    # 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()
+    # Aggregate-only spool footprint: no session ids, no workspace names, no
+    # per-key table -- same privacy contract as `metrics` above. spool_stats()
+    # is a synchronous, O(1) cache read -- it never scans the queue
+    # directory (see QueueManager.spool_stats() / _spool_stats_refresher).
+    # `as_of_seconds` tells callers how stale that snapshot is (None before
+    # the background refresher has produced one), so /status never presents
+    # a cached number as though it were live.
+    _spool: dict[str, Any] = dict(registry.queue_manager.spool_stats())
+    _spool["as_of_seconds"] = registry.queue_manager.spool_stats_age_seconds()
+    response["spool"] = _spool
     # 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

context_intelligence_server/queue_manager.py

@@ -140,15 +140,17 @@ def __init__(self, queues_dir: Path):
         self._stats_cache: dict[str, Any] | None = None
         self._stats_cache_at: float = 0.0
         self._stats_cache_ttl: float = 1.0
-        # Separate cache for spool_stats(). A longer TTL than _stats_cache_ttl
-        # is fine here: spool_stats() is an
-        # operator-facing "is the backlog growing" signal, not a
-        # correctness-sensitive value, so a few extra seconds of staleness is
-        # an acceptable trade for fewer directory scans under frequent
-        # /status polling.
+        # 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
+        # (config.spool_stats_refresh_interval_seconds) -- NOT gated by a
+        # read-side TTL. spool_stats() itself is a pure cache read and never
+        # scans the directory: a ~5000-file spool on Azure Files SMB
+        # previously made the inline per-request scan take minutes
+        # (incident 2026-09-09), because /status called the equivalent of
+        # today's refresh_spool_stats() directly, on every request.
         self._spool_cache: dict[str, int] | None = None
         self._spool_cache_at: float = 0.0
-        self._spool_cache_ttl: float = 5.0
         # One _KeyGuard per worker key that has been appended to and
         # not yet finalized-and-deleted. Created lazily by _guard(); removed
         # ONLY by delete_drained, under the admission lock, gated on identity
@@ -614,6 +616,54 @@ def _count_dead(self, worker_key: str) -> int:
         """
         return self._stream_newlines(self._dead_path(worker_key))
 
+    async def reclaim_drained_orphans(self) -> tuple[int, int]:
+        """Delete the ``.log``/``.offset`` of every FULLY-DRAINED key on disk.
+
+        Returns ``(keys_reclaimed, bytes_reclaimed)``.
+
+        WHY THIS EXISTS (measured 2026-09-09, local box): the drain loop's
+        per-session trim only fires for a session that currently HAS a live
+        drain worker. Crash recovery respawns a worker only for a session with
+        UNDRAINED data, so a session that was fully drained and then went away
+        gets no worker, never reaches the idle branch, and keeps its files
+        forever -- and every later boot's recovery re-walks them. On a real
+        spool that was 25 fully-drained logs totalling ~1.7 GiB, including a
+        691 MB and a 600 MB file, none of which the per-session trim could
+        ever see.
+
+        Safe by construction, because it delegates to ``delete_drained``:
+        idempotent, takes the key's ``file_lock``, REFUSES (cheaply, returning
+        False) while any uncommitted byte remains, keeps ``.dead.jsonl``, and
+        unlinks a stale ``.offset`` so a log recreated later cannot read past
+        its own end. A key with a live worker mid-drain is therefore skipped
+        by ``delete_drained`` itself -- no separate exclusion list, and no way
+        for this to race a drainer into losing data.
+
+        Cost is O(keys), never O(bytes): one ``stat`` for the size (so the
+        reclaimed total can be reported) plus ``delete_drained``'s own
+        stat + offset read. Never raises: a per-key failure is logged and
+        skipped so one unreadable file cannot abort the pass.
+        """
+        keys_reclaimed = 0
+        bytes_reclaimed = 0
+        for key in self._all_worker_keys():
+            try:
+                try:
+                    size = self._log_path(key).stat().st_size
+                except FileNotFoundError:
+                    size = 0
+                if await self.delete_drained(key):
+                    # Only count a key that actually had a log to reclaim --
+                    # delete_drained also returns True for a key whose log was
+                    # already absent (dead-letter-only), which reclaims nothing.
+                    if size:
+                        keys_reclaimed += 1
+                        bytes_reclaimed += size
+            except (OSError, ValueError):
+                logger.warning("reclaim_drained_orphans_key_failed key=%s", key)
+                continue
+        return keys_reclaimed, bytes_reclaimed
+
     def _all_worker_keys(self) -> list[str]:
         """Return the sorted union of ``.log`` and ``.dead.jsonl`` stems.
 
@@ -679,26 +729,32 @@ def _all() -> dict[str, Any]:
         self._stats_cache_at = now
         return stats
 
-    async def spool_stats(self) -> dict[str, int]:
-        """Cheap, aggregate-only spool footprint for the unauthenticated /status.
+    async def refresh_spool_stats(self) -> dict[str, int]:
+        """Scan the spool directory and refresh the snapshot ``spool_stats()`` serves.
 
         Returns two integers: ``pending_sessions`` (keys whose committed offset
         is below the ``.log`` size) and ``spool_bytes_total`` (bytes across all
-        queue files). Sized via ``stat()`` per file -- O(file count), never
-        O(bytes) -- and cached for ``_spool_cache_ttl`` seconds. No identifiers
-        are returned or derivable.
-
-        Must not raise (/status is the unauthenticated health probe): a
-        directory-level failure returns the uncached sentinel ``{-1, -1}`` and a
-        per-file failure skips that entry. ``-1`` means "temporarily
-        unavailable", distinct from a real ``0``.
+        queue files), plus ``corrupt_offsets`` (a non-numeric ``.offset``
+        count). Sized via ``stat()`` per file -- O(file count), never
+        O(bytes). No identifiers are returned or derivable.
+
+        This is the ONLY method that walks the queue directory for spool
+        stats. On success it stores the result as the snapshot
+        ``spool_stats()`` reads (``_spool_cache`` / ``_spool_cache_at``) --
+        the sole write path for that cache. Intended to be called
+        periodically off the request path (see ``_spool_stats_refresher`` in
+        main.py), on ``config.spool_stats_refresh_interval_seconds``; never
+        from ``/status`` directly, which is exactly what made a ~5000-file
+        spool on Azure Files SMB stall the unauthenticated health probe for
+        minutes (incident 2026-09-09).
+
+        Must not raise (called from a background loop that must keep
+        running): a per-file failure skips that entry, and a directory-level
+        failure returns the uncached sentinel ``{-1, -1, -1}`` WITHOUT
+        touching ``_spool_cache`` -- a prior good snapshot survives a
+        transient scan failure rather than being wiped by it. ``-1`` means
+        "temporarily unavailable", distinct from a real ``0``.
         """
-        now = time.monotonic()
-        if (
-            self._spool_cache is not None
-            and (now - self._spool_cache_at) < self._spool_cache_ttl
-        ):
-            return self._spool_cache
 
         def _scan() -> dict[str, int]:
             spool_bytes_total = 0
@@ -742,19 +798,61 @@ def _scan() -> dict[str, int]:
             stats = await asyncio.to_thread(_scan)
         except (OSError, ValueError):
             # Queue dir missing/unavailable, or a transient FS error mid-scan.
-            # /status must return 200: degrade to an uncached sentinel so the
-            # next poll retries once the filesystem recovers. -1 means
-            # "temporarily unavailable", distinct from a real 0.
+            # The refresher must keep running regardless: degrade to the
+            # uncached sentinel so the next refresh retries once the
+            # filesystem recovers, and so /status's `as_of_seconds` shows the
+            # existing snapshot (if any) going stale rather than this failure
+            # silently overwriting it. -1 means "temporarily unavailable",
+            # distinct from a real 0.
             return {
                 "pending_sessions": -1,
                 "spool_bytes_total": -1,
                 "corrupt_offsets": -1,
             }
 
         self._spool_cache = stats
-        self._spool_cache_at = now
+        self._spool_cache_at = time.monotonic()
         return stats
 
+    def spool_stats(self) -> dict[str, int]:
+        """Cheap, read-only, aggregate-only spool footprint for the unauthenticated /status.
+
+        Returns the snapshot last produced by ``refresh_spool_stats()`` (see
+        ``_spool_stats_refresher`` in main.py), or the sentinel
+        ``{"pending_sessions": -1, "spool_bytes_total": -1,
+        "corrupt_offsets": -1}`` when no snapshot has been produced yet
+        (``-1`` means "temporarily unavailable", distinct from a real ``0``).
+        No identifiers are returned or derivable.
+
+        Synchronous and NEVER touches the filesystem -- that is the entire
+        point of the refresh/read split (see ``refresh_spool_stats``).
+        O(1) regardless of spool size, on every call. Never raises.
+
+        Returns a FRESH dict each call (not a reference into the cache), so a
+        caller (e.g. ``/status``) can add fields of its own -- such as
+        ``as_of_seconds`` via ``spool_stats_age_seconds()`` -- without
+        mutating the snapshot the next caller sees.
+        """
+        if self._spool_cache is not None:
+            return dict(self._spool_cache)
+        return {
+            "pending_sessions": -1,
+            "spool_bytes_total": -1,
+            "corrupt_offsets": -1,
+        }
+
+    def spool_stats_age_seconds(self) -> float | None:
+        """Seconds since the cached spool snapshot was last refreshed.
+
+        Returns ``None`` when ``refresh_spool_stats()`` has never produced a
+        snapshot yet. Lets ``/status`` distinguish a fresh reading from a
+        stale one instead of presenting whatever is cached as though it were
+        live. Pure arithmetic: never touches the filesystem, never raises.
+        """
+        if self._spool_cache is None:
+            return None
+        return round(time.monotonic() - self._spool_cache_at, 1)
+
     async def dead_letter_keys(self) -> list[str]:
         """Return sorted worker keys that have a ``.dead.jsonl`` file.