diff --git a/context_intelligence_server/config.py b/context_intelligence_server/config.py index 4e36ba0..85742be 100644 --- a/context_intelligence_server/config.py +++ b/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 # ------------------------------------------------------------------------- diff --git a/context_intelligence_server/main.py b/context_intelligence_server/main.py index e37d3dc..0fbaac5 100644 --- a/context_intelligence_server/main.py +++ b/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 diff --git a/context_intelligence_server/queue_manager.py b/context_intelligence_server/queue_manager.py index a52eca1..282de39 100644 --- a/context_intelligence_server/queue_manager.py +++ b/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,9 +798,12 @@ 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, @@ -752,9 +811,48 @@ def _scan() -> dict[str, int]: } 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. diff --git a/context_intelligence_server/registry.py b/context_intelligence_server/registry.py index 22ec953..b82e8e8 100644 --- a/context_intelligence_server/registry.py +++ b/context_intelligence_server/registry.py @@ -403,6 +403,29 @@ async def drain_worker( await self._safe_close(worker) self._deregister(session_id) return + # Trim as we go: don't wait for a clean session:end + # (_finalize_session) to reclaim disk. delete_drained + # is idempotent, takes the key's file_lock, REFUSES + # (returns False, cheaply) while any uncommitted + # bytes remain, keeps .dead.jsonl, and unlinks a + # stale .offset so a log recreated later cannot read + # past its own end -- the exact same call + # _finalize_session already makes, just triggered + # here too, earlier and repeatedly, so a session + # that goes quiet without ever finalizing cleanly + # (crash, orphan, indefinite idle) doesn't keep its + # drained .log/.offset on disk -- and off every + # later boot's recovery scan -- forever. + if await qm.delete_drained(session_id): + # INFO, not DEBUG: a reclaim that leaves no trace + # in the journal is one an operator cannot confirm + # is happening. Measured 2026-09-09: 16 files were + # trimmed on a live box and the log showed nothing. + logger.info( + "spool_trimmed session=%s", + session_id, + extra={"session_id": session_id}, + ) continue idle_elapsed = 0.0 diff --git a/pyproject.toml b/pyproject.toml index 44c5e64..f149432 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "context-intelligence-server" -version = "6.7.4" +version = "6.7.5" description = "Context Intelligence Server for Amplifier" requires-python = ">=3.11" dependencies = [ diff --git a/tests/test_main.py b/tests/test_main.py index bcc17cc..abc2f0e 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -971,6 +971,178 @@ async def _blocked_forever(app: Any) -> None: assert leftover == [], f"recovery task(s) still pending after shutdown: {leftover}" +# --------------------------------------------------------------------------- +# Startup recovery: reclaim orphaned drained spool files +# (fix/trim-spool-as-processed) +# --------------------------------------------------------------------------- + + +async def test_startup_recovery_logs_reclaimed_orphans( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """``_startup_recovery_body`` calls ``reclaim_drained_orphans()`` and, + when keys were reclaimed, logs an INFO line naming the exact counts -- + an operator signal that orphaned spool files were swept on this boot.""" + monkeypatch.setattr( + registry.queue_manager, + "reclaim_drained_orphans", + AsyncMock(return_value=(3, 4096)), + ) + + 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.INFO, logger="context_intelligence_server"), + ): + async with lifespan(main_module.app): + await asyncio.wait_for( + main_module.app.state.recovery_complete.wait(), timeout=5 + ) + assert main_module.app.state.recovery_error is None + + reclaimed_records = [ + r for r in caplog.records if "startup_recovery: reclaimed" in r.getMessage() + ] + assert reclaimed_records, ( + f"expected an INFO 'startup_recovery: reclaimed' log; got " + f"{[r.getMessage() for r in caplog.records]}" + ) + message = reclaimed_records[0].getMessage() + assert "3" in message + assert "4096" in message + + +async def test_startup_recovery_orphan_reclaim_failure_does_not_fail_boot( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """A raising ``reclaim_drained_orphans()`` must not fail startup recovery + -- it is caught and logged INSIDE ``_startup_recovery_body``, so the boot + completes normally: ``recovery_complete`` still gets set and + ``recovery_error`` stays None (the outer ``_startup_recovery`` wrapper + never sees an exception).""" + monkeypatch.setattr( + registry.queue_manager, + "reclaim_drained_orphans", + AsyncMock(side_effect=RuntimeError("boom-orphan-reclaim")), + ) + + 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 main_module.app.state.recovery_error is None + + error_records = [r for r in caplog.records if r.levelno == logging.ERROR] + assert any("orphan reclaim failed" in r.getMessage() for r in error_records), ( + f"expected an ERROR log mentioning 'orphan reclaim failed'; " + f"got {[r.getMessage() for r in error_records]}" + ) + + +# --------------------------------------------------------------------------- +# Background spool-stats refresher (fix/trim-spool-as-processed): /status +# must never scan the queue directory itself -- a background loop refreshes +# the snapshot it reads instead. +# --------------------------------------------------------------------------- + + +async def test_spool_stats_refresher_survives_a_failing_refresh( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """A single failed refresh must not kill the background loop -- it logs + at ERROR and keeps going (the snapshot just goes stale; see /status's + spool.as_of_seconds).""" + monkeypatch.setattr( + main_module._settings, "spool_stats_refresh_interval_seconds", 0.01 + ) + + calls = {"n": 0} + real_refresh = registry.queue_manager.refresh_spool_stats + + async def _flaky_refresh() -> dict[str, int]: + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("boom-spool-refresh") + return await real_refresh() + + monkeypatch.setattr(registry.queue_manager, "refresh_spool_stats", _flaky_refresh) + + with caplog.at_level(logging.ERROR, logger="context_intelligence_server"): + task = asyncio.create_task(main_module._spool_stats_refresher(main_module.app)) + for _ in range(200): + await asyncio.sleep(0.01) + if calls["n"] >= 2: + break + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + assert calls["n"] >= 2 # the loop kept going after the failure + error_records = [r for r in caplog.records if r.levelno == logging.ERROR] + assert any("spool_stats_refresh_failed" in r.getMessage() for r in error_records) + + +async def test_spool_stats_refresher_task_is_cancelled_on_shutdown() -> None: + """Shutdown must cancel the background spool-stats refresher task -- not + leave it running against a queue directory a shutting-down process may + be tearing down around it (mirrors + test_recovery_task_is_cancelled_on_shutdown).""" + 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): + # Let the refresher task actually get scheduled and complete at + # least one iteration (initial refresh, then its sleep). + await asyncio.sleep(0.05) + + # Confirm the task actually exists WHILE lifespan is running -- + # otherwise the "no leftover after shutdown" check below would + # pass vacuously if the task were never spawned at all. Match + # the background task's own coroutine call form (with the + # opening paren) -- not just the substring + # "_spool_stats_refresher", which this very test function's name + # also contains. + running = [ + t + for t in asyncio.all_tasks() + if not t.done() and "_spool_stats_refresher(" in repr(t) + ] + assert running, "spool stats refresher task was never scheduled" + + leftover = [ + t + for t in asyncio.all_tasks() + if not t.done() and "_spool_stats_refresher(" in repr(t) + ] + assert leftover == [], ( + f"spool stats refresher task(s) still pending after shutdown: {leftover}" + ) + + async def test_status_reports_recovery_state_before_lifespan_has_run( client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch, @@ -1724,7 +1896,9 @@ async def test_status_includes_metrics_block(client: httpx.AsyncClient) -> None: async def test_status_includes_spool_block(client: httpx.AsyncClient) -> None: """/status carries an additive, aggregate-only spool block so a growing on-disk backlog is never invisible (the 38 GB / two-day incident this - guards against had zero signal anywhere).""" + guards against had zero signal anywhere). Now also carries + `as_of_seconds` (Change 1) so a caller can tell a fresh reading from a + stale one instead of assuming the numbers are live.""" response = await client.get("/status") assert response.status_code == 200 data = response.json() @@ -1735,17 +1909,38 @@ async def test_status_includes_spool_block(client: httpx.AsyncClient) -> None: "pending_sessions", "spool_bytes_total", "corrupt_offsets", + "as_of_seconds", } assert isinstance(spool["pending_sessions"], int) assert isinstance(spool["spool_bytes_total"], int) assert isinstance(spool["corrupt_offsets"], int) -async def test_status_spool_block_reflects_real_backlog( +async def test_status_spool_as_of_seconds_none_before_any_refresh( + client: httpx.AsyncClient, +) -> None: + """Before the background refresher has ever populated the cache (the + `client` fixture never runs `lifespan`, so no refresh has happened), + /status reports the sentinel with `as_of_seconds` None -- /status itself + never scans, so there is nothing else it could honestly report.""" + response = await client.get("/status") + spool = response.json()["spool"] + + assert spool == { + "pending_sessions": -1, + "spool_bytes_total": -1, + "corrupt_offsets": -1, + "as_of_seconds": None, + } + + +async def test_status_spool_block_reflects_real_backlog_after_refresh( client: httpx.AsyncClient, ) -> None: - """The spool block's numbers move when there's real undrained data on - disk -- not a hardcoded placeholder.""" + """The spool block's numbers move once the background refresher has run + -- not a hardcoded placeholder. /status itself never scans (Change 1), so + the test drives the refresh explicitly, exactly as + main.py._spool_stats_refresher does on its own cadence.""" qm = registry.queue_manager body = json.dumps( { @@ -1756,13 +1951,16 @@ async def test_status_spool_block_reflects_real_backlog( ).encode("utf-8") await qm.append("sess-spool-visible", body) # get_or_create is bypassed here (raw append only) so this line stays - # undrained -- exactly the "pending" shape spool_stats() measures. + # undrained -- exactly the "pending" shape refresh_spool_stats() measures. + await qm.refresh_spool_stats() response = await client.get("/status") data = response.json() assert data["spool"]["pending_sessions"] >= 1 assert data["spool"]["spool_bytes_total"] > 0 + assert isinstance(data["spool"]["as_of_seconds"], float) + assert data["spool"]["as_of_seconds"] >= 0.0 async def test_status_spool_block_never_leaks_session_identifiers( @@ -1780,6 +1978,7 @@ async def test_status_spool_block_never_leaks_session_identifiers( } ).encode("utf-8") await qm.append(secret_sid, body) + await qm.refresh_spool_stats() response = await client.get("/status") raw_text = response.text @@ -1793,13 +1992,14 @@ async def test_status_corrupt_offset_returns_200_and_surfaces_count( client: httpx.AsyncClient, ) -> None: """Regression (ci_pr73-267): a corrupt .offset previously 500'd /status via - derive_all_stats() (which runs before spool_stats in get_status). /status - must now stay 200 AND surface the corruption as spool.corrupt_offsets -- the - only signal (no logging, so the polled health path is never flooded).""" + derive_all_stats() (which runs before the spool block in get_status). + /status must stay 200 AND surface the corruption as spool.corrupt_offsets + -- the only signal (no logging, so the polled health path is never + flooded) -- once a refresh has actually observed it.""" qm = registry.queue_manager await qm.append("sess-corrupt-offset", b"a") qm._offset_path("sess-corrupt-offset").write_text("not-a-number", encoding="utf-8") - qm._spool_cache = None # bypass TTL cache so the corruption is seen now + await qm.refresh_spool_stats() # the only thing that scans; sees the corruption response = await client.get("/status") @@ -1809,21 +2009,13 @@ async def test_status_corrupt_offset_returns_200_and_surfaces_count( async def test_status_returns_200_when_spool_dir_unavailable( client: httpx.AsyncClient, - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, ) -> None: - """Regression (ci_pr73-ueh): /status is the ACA health probe. spool_stats() - scans the queue dir with iterdir() (raises on a missing dir), unlike the - glob()-based sibling readers. A transiently-unavailable queue dir (e.g. an - Azure Files SMB remount) must NOT turn /status into a 500 -> failed probe - -> container restart loop. It must return 200 with a degraded sentinel.""" - qm = registry.queue_manager - # Point the scan at a directory that does not exist so iterdir() raises, - # exactly as it would during an SMB mount drop. Bypass the TTL cache so the - # scan actually runs on this call. - monkeypatch.setattr(qm, "_dir", tmp_path / "gone") - monkeypatch.setattr(qm, "_spool_cache", None) - + """Regression (ci_pr73-ueh): /status is the ACA health probe. It must + never turn a broken/unavailable queue dir into a 500 -> failed probe -> + container restart loop. Since Change 1, /status's spool block is a pure + cache read (QueueManager.spool_stats()) that never touches the + filesystem at all -- so this holds unconditionally, without needing to + break the directory for this specific request.""" response = await client.get("/status") assert response.status_code == 200 @@ -1831,6 +2023,7 @@ async def test_status_returns_200_when_spool_dir_unavailable( "pending_sessions": -1, "spool_bytes_total": -1, "corrupt_offsets": -1, + "as_of_seconds": None, } diff --git a/tests/test_queue_manager.py b/tests/test_queue_manager.py index e84c164..54ef4af 100644 --- a/tests/test_queue_manager.py +++ b/tests/test_queue_manager.py @@ -575,17 +575,20 @@ async def test_recovery_seed_counts_replay_window_residual_zero(qm): # --------------------------------------------------------------------------- -# spool_stats (Change 2): cheap, aggregate-only spool footprint for /status +# spool stats (Change 2, split for the /status-must-never-scan fix): +# refresh_spool_stats() does the ONE-AND-ONLY directory scan and populates +# the cache; spool_stats() is a synchronous, read-only, never-scans cache +# read for /status. See QueueManager.refresh_spool_stats/spool_stats. # --------------------------------------------------------------------------- -async def test_spool_stats_counts_pending_session_and_bytes(qm, tmp_path): +async def test_refresh_spool_stats_counts_pending_session_and_bytes(qm, tmp_path): """A session with unconsumed log data counts as pending; total bytes reflects every file on disk (.log + .offset + .dead.jsonl).""" await qm.append("s1", b"a") await qm.append("s1", b"b") - stats = await qm.spool_stats() + stats = await qm.refresh_spool_stats() assert stats["pending_sessions"] == 1 queues_dir = tmp_path / "queues" @@ -594,7 +597,7 @@ async def test_spool_stats_counts_pending_session_and_bytes(qm, tmp_path): assert expected_bytes > 0 -async def test_spool_stats_fully_committed_session_not_pending(qm): +async def test_refresh_spool_stats_fully_committed_session_not_pending(qm): """A session whose committed offset reaches EOF is NOT counted as pending, even though its .log/.offset files still occupy disk space (spool_bytes_total still reflects them).""" @@ -602,25 +605,25 @@ async def test_spool_stats_fully_committed_session_not_pending(qm): line = b"a\n" await qm.commit("s1", len(line)) - stats = await qm.spool_stats() + stats = await qm.refresh_spool_stats() assert stats["pending_sessions"] == 0 assert stats["spool_bytes_total"] > 0 # log + offset files still on disk -async def test_spool_stats_dead_letter_only_session_not_pending(qm): +async def test_refresh_spool_stats_dead_letter_only_session_not_pending(qm): """A dead-letter-only key (no .log) contributes bytes but is never counted as a pending session -- pending_sessions is defined purely over .log files with unconsumed data.""" await qm.dead_letter("s-dead", b"poison", error="boom") - stats = await qm.spool_stats() + stats = await qm.refresh_spool_stats() assert stats["pending_sessions"] == 0 assert stats["spool_bytes_total"] > 0 -async def test_spool_stats_multiple_sessions_aggregate(qm): +async def test_refresh_spool_stats_multiple_sessions_aggregate(qm): """pending_sessions counts sessions independently; bytes sum across all.""" await qm.append("s1", b"a") # pending await qm.append("s2", b"b") @@ -628,19 +631,20 @@ async def test_spool_stats_multiple_sessions_aggregate(qm): await qm.commit("s2", len(line)) # fully committed, not pending await qm.append("s3", b"c") # pending - stats = await qm.spool_stats() + stats = await qm.refresh_spool_stats() assert stats["pending_sessions"] == 2 -async def test_spool_stats_returns_only_aggregate_keys_no_identifiers(qm): - """/status is unauthenticated: spool_stats() must return ONLY the two - aggregate integers -- no session ids, workspace names, or per-key table - of any kind, so there's nothing to accidentally leak through /status.""" +async def test_refresh_spool_stats_returns_only_aggregate_keys_no_identifiers(qm): + """/status is unauthenticated: the spool snapshot must carry ONLY the + three aggregate fields -- no session ids, workspace names, or per-key + table of any kind, so there's nothing to accidentally leak through + /status.""" await qm.append("my-secret-session-id", b"a") await qm.dead_letter("another-session-id", b"poison", error="boom") - stats = await qm.spool_stats() + stats = await qm.refresh_spool_stats() assert set(stats.keys()) == { "pending_sessions", @@ -651,10 +655,14 @@ async def test_spool_stats_returns_only_aggregate_keys_no_identifiers(qm): assert "my-secret-session-id" not in serialized assert "another-session-id" not in serialized + # spool_stats() (the /status read path) exposes the exact same snapshot. + assert qm.spool_stats() == stats -async def test_spool_stats_caches_within_ttl(qm, monkeypatch): - """Repeated calls within the TTL window are served from cache -- the - directory is not re-scanned on every /status poll.""" + +async def test_refresh_spool_stats_always_scans_no_ttl_gate(qm, monkeypatch): + """refresh_spool_stats() has NO TTL gate -- every call re-scans the + directory. The old read-side TTL is gone; cadence is now owned entirely + by the background refresher loop (main.py._spool_stats_refresher).""" import pathlib await qm.append("s1", b"a") @@ -674,16 +682,44 @@ def counting_iterdir(self: pathlib.Path): monkeypatch.setattr(pathlib.Path, "iterdir", counting_iterdir) - await qm.spool_stats() - await qm.spool_stats() # within TTL -> served from cache - assert calls["n"] == 1 - - # Age the cache past the TTL; the next call must recompute. - qm._spool_cache_at = time.monotonic() - (qm._spool_cache_ttl + 1.0) - await qm.spool_stats() + await qm.refresh_spool_stats() + await qm.refresh_spool_stats() # no TTL -- scans again every time assert calls["n"] == 2 +def test_spool_stats_is_synchronous_read_only_and_never_scans(qm, monkeypatch): + """spool_stats() must NEVER touch the filesystem: it is a pure, + synchronous cache read. An empty cache returns the unavailable sentinel + (never scans, never raises); a populated cache returns exactly that + snapshot -- this is the entire point of the refresh/read split that + fixes /status stalling on a large spool (see the module docstring).""" + import pathlib + + def _boom(self: pathlib.Path) -> None: + raise AssertionError("spool_stats() touched the filesystem via iterdir()") + + monkeypatch.setattr(pathlib.Path, "iterdir", _boom) + + # Empty cache: the sentinel, no scan, no raise. + assert qm.spool_stats() == { + "pending_sessions": -1, + "spool_bytes_total": -1, + "corrupt_offsets": -1, + } + + # Populated cache: returns exactly that snapshot, still without scanning. + qm._spool_cache = { + "pending_sessions": 3, + "spool_bytes_total": 12345, + "corrupt_offsets": 0, + } + assert qm.spool_stats() == { + "pending_sessions": 3, + "spool_bytes_total": 12345, + "corrupt_offsets": 0, + } + + # --------------------------------------------------------------------------- # Streamed boot/stats scans (ci_pr73-xq2): _complete_data_end and # _count_newlines must be bounded-memory AND numerically identical to the old @@ -792,9 +828,9 @@ async def test_recovery_seed_counts_unchanged_under_streaming(qm): assert accepted == 2 # one written + one pending -async def test_spool_stats_empty_directory(qm): - """An empty spool directory reports zero for both aggregates.""" - stats = await qm.spool_stats() +async def test_refresh_spool_stats_empty_directory(qm): + """An empty spool directory reports zero for all three aggregates.""" + stats = await qm.refresh_spool_stats() assert stats == { "pending_sessions": 0, "spool_bytes_total": 0, @@ -803,25 +839,26 @@ async def test_spool_stats_empty_directory(qm): # --------------------------------------------------------------------------- -# spool_stats health-endpoint safety (regression, ci_pr73-ueh): -# /status is the unauthenticated ACA health probe and calls spool_stats() -# unconditionally. spool_stats() uses iterdir() (raises on a missing dir), -# unlike every sibling reader which uses glob() (empty on a missing dir), so -# a transiently-unavailable queue dir or a corrupt .offset MUST degrade to a -# sentinel, never raise -- an escape becomes a 500 -> failed probe -> restart. +# refresh_spool_stats health-endpoint safety (regression, ci_pr73-ueh): +# the background refresher (main.py._spool_stats_refresher) calls +# refresh_spool_stats() unconditionally. It uses iterdir() (raises on a +# missing dir), unlike every sibling reader which uses glob() (empty on a +# missing dir), so a transiently-unavailable queue dir or a corrupt .offset +# MUST degrade to a sentinel, never raise -- an escape would kill the +# refresher loop and (pre-split) would have 500'd /status directly. # --------------------------------------------------------------------------- -async def test_spool_stats_missing_directory_returns_sentinel(qm, tmp_path): - """A missing queue dir makes iterdir() raise FileNotFoundError; spool_stats - must return the degraded sentinel {-1, -1} rather than propagate (which - would 500 the /status health probe -- e.g. during an Azure Files remount).""" +async def test_refresh_spool_stats_missing_directory_returns_sentinel(qm, tmp_path): + """A missing queue dir makes iterdir() raise FileNotFoundError; + refresh_spool_stats() must return the degraded sentinel {-1, -1, -1} + rather than propagate (which would kill the background refresher loop -- + e.g. during an Azure Files remount).""" import shutil shutil.rmtree(tmp_path / "queues") - qm._spool_cache = None # bypass the TTL cache so the scan actually runs - stats = await qm.spool_stats() + stats = await qm.refresh_spool_stats() assert stats == { "pending_sessions": -1, @@ -830,65 +867,219 @@ async def test_spool_stats_missing_directory_returns_sentinel(qm, tmp_path): } -async def test_spool_stats_sentinel_is_not_cached(qm, tmp_path): - """The degraded sentinel is NOT cached: once the directory is healthy - again, the very next call recovers the real aggregate numbers.""" +async def test_refresh_spool_stats_sentinel_is_not_cached(qm, tmp_path): + """A failed refresh does NOT overwrite the cache: once the directory is + healthy again, the very next refresh recovers the real aggregate numbers + (and, meanwhile, spool_stats() never reports a fabricated sentinel as a + real snapshot).""" import shutil queues_dir = tmp_path / "queues" shutil.rmtree(queues_dir) - qm._spool_cache = None - assert await qm.spool_stats() == { + assert await qm.refresh_spool_stats() == { "pending_sessions": -1, "spool_bytes_total": -1, "corrupt_offsets": -1, } + assert qm._spool_cache is None # the sentinel was never cached - # Filesystem recovers; no manual cache reset -- the sentinel was never stored. + # Filesystem recovers. queues_dir.mkdir(parents=True, exist_ok=True) await qm.append("s1", b"a") - stats = await qm.spool_stats() + stats = await qm.refresh_spool_stats() assert stats["pending_sessions"] == 1 assert stats["spool_bytes_total"] > 0 + assert qm.spool_stats() == stats # now cached, and readable without a scan -async def test_spool_stats_corrupt_offset_does_not_sink_scan(qm): +async def test_refresh_spool_stats_corrupt_offset_does_not_sink_scan(qm): """A corrupt/unreadable .offset for one session must not fail the whole - scan (which would 500 /status): that file's bytes still count, only its - pending calc is skipped.""" + scan: that file's bytes still count, only its pending calc is skipped.""" await qm.append("s1", b"a") qm._offset_path("s1").write_text("not-a-number", encoding="utf-8") - qm._spool_cache = None - stats = await qm.spool_stats() + stats = await qm.refresh_spool_stats() assert stats["spool_bytes_total"] > 0 assert isinstance(stats["pending_sessions"], int) -async def test_spool_stats_counts_corrupt_offsets(qm): +async def test_refresh_spool_stats_counts_corrupt_offsets(qm): """A non-numeric .offset is surfaced as an aggregate corrupt_offsets count (the ONLY visibility signal -- no logging). A healthy session contributes 0.""" await qm.append("s-good", b"a") # valid: no .offset yet -> committed 0 await qm.append("s-bad", b"a") qm._offset_path("s-bad").write_text("not-a-number", encoding="utf-8") - qm._spool_cache = None - stats = await qm.spool_stats() + stats = await qm.refresh_spool_stats() assert stats["corrupt_offsets"] == 1 assert stats["spool_bytes_total"] > 0 # corrupt file's bytes still counted -async def test_spool_stats_healthy_offsets_report_zero_corrupt(qm): +async def test_refresh_spool_stats_healthy_offsets_report_zero_corrupt(qm): """corrupt_offsets is 0 when every .offset is a valid integer (it must not fire on the normal committed-offset path).""" await qm.append("s1", b"a") line = b"a\n" await qm.commit("s1", len(line)) # writes a valid numeric .offset - qm._spool_cache = None - stats = await qm.spool_stats() + stats = await qm.refresh_spool_stats() assert stats["corrupt_offsets"] == 0 + + +async def test_spool_stats_age_seconds_none_before_any_refresh(qm): + """spool_stats_age_seconds() is None until refresh_spool_stats() has + produced its first snapshot -- pure arithmetic, never touches disk.""" + assert qm.spool_stats_age_seconds() is None + + +async def test_spool_stats_age_seconds_reflects_time_since_refresh(qm): + """After a refresh, spool_stats_age_seconds() reports elapsed time (>= 0, + rounded to 1dp) and grows as time passes -- how /status tells a fresh + snapshot from a stale one.""" + await qm.refresh_spool_stats() + age0 = qm.spool_stats_age_seconds() + assert isinstance(age0, float) + assert age0 >= 0.0 + + qm._spool_cache_at = time.monotonic() - 5.0 + age1 = qm.spool_stats_age_seconds() + assert age1 >= 5.0 + + +async def test_recovery_seed_counts_residual_zero_after_trimming_dead_letter_session( + qm, +): + """Conservation holds after Change 2's early trim: a session whose + .log/.offset have already been reclaimed by delete_drained() (leaving + only its .dead.jsonl) still seeds a zero residual -- the dead record is + accounted for entirely through `dead`, not lost when its log disappears.""" + await qm.append("s1", b"a") + line = b"a\n" + await qm.commit("s1", len(line)) # fully drained/committed + await qm.dead_letter("s1", b"a", error="boom") # a dead-letter record too + + assert await qm.delete_drained("s1") is True # log/offset reclaimed + assert not qm._log_path("s1").exists() + assert not qm._offset_path("s1").exists() + assert qm._dead_path("s1").exists() + + accepted, written = await qm.recovery_seed_counts() + stats = await qm.derive_all_stats() + residual = accepted - written - stats["in_queue_total"] - stats["dead_total"] + assert residual == 0 + + +# --------------------------------------------------------------------------- +# reclaim_drained_orphans: reclaim .log/.offset for keys with no live worker +# (fix/trim-spool-as-processed) -- the per-session idle-branch trim only ever +# runs for a session with a live drain worker, so a session that fully +# drained and then went away keeps its files forever without this sweep. +# --------------------------------------------------------------------------- + + +async def test_reclaim_drained_orphans_removes_fully_drained_key(qm, tmp_path): + """A fully-drained (committed-to-EOF) key's .log/.offset are removed, and + the exact byte size of that log is reported back.""" + line = b"hello\n" + await qm.append("s1", line) + await qm.commit("s1", len(line)) + log_size = qm._log_path("s1").stat().st_size + assert log_size == len(line) + + result = await qm.reclaim_drained_orphans() + + assert result == (1, log_size) + assert not qm._log_path("s1").exists() + assert not qm._offset_path("s1").exists() + + +async def test_reclaim_drained_orphans_keeps_dead_letters(qm): + """A reclaimed key's .dead.jsonl survives -- only .log/.offset are removed.""" + line = b"hello\n" + await qm.append("s1", line) + await qm.commit("s1", len(line)) + await qm.dead_letter("s1", b"poison", error="boom") + + result = await qm.reclaim_drained_orphans() + + assert result == (1, len(line)) + assert not qm._log_path("s1").exists() + assert qm._dead_path("s1").exists() # retained + assert len(await qm.read_dead_letters("s1")) == 1 + + +async def test_reclaim_drained_orphans_skips_key_with_uncommitted_bytes(qm): + """A key whose log has bytes beyond the committed offset is left + entirely untouched -- delete_drained refuses it, so it is never counted.""" + await qm.append("s-pending", b"a") # no commit() -> committed offset is 0 + + result = await qm.reclaim_drained_orphans() + + assert result == (0, 0) + assert qm._log_path("s-pending").exists() + + +async def test_reclaim_drained_orphans_ignores_dead_letter_only_key(qm): + """A key with only a .dead.jsonl (no .log at all) reclaims nothing -- + delete_drained returns True for it (nothing to remove), but with a zero + log size it must not be counted as a reclaimed key.""" + await qm.dead_letter("s-dead-only", b"poison", error="boom") + + result = await qm.reclaim_drained_orphans() + + assert result == (0, 0) + assert qm._dead_path("s-dead-only").exists() + + +async def test_reclaim_drained_orphans_mixed_directory(qm): + """Two fully-drained keys and one pending key: only the drained keys are + reclaimed (count and byte total cover exactly those two), and the + pending key's files are left in place.""" + line_a = b"aaa\n" + line_b = b"bbbbb\n" + await qm.append("s-drained-a", line_a) + await qm.commit("s-drained-a", len(line_a)) + await qm.append("s-drained-b", line_b) + await qm.commit("s-drained-b", len(line_b)) + await qm.append("s-pending", b"c") # never committed + + result = await qm.reclaim_drained_orphans() + + assert result == (2, len(line_a) + len(line_b)) + assert not qm._log_path("s-drained-a").exists() + assert not qm._offset_path("s-drained-a").exists() + assert not qm._log_path("s-drained-b").exists() + assert not qm._offset_path("s-drained-b").exists() + assert qm._log_path("s-pending").exists() + + +async def test_reclaim_drained_orphans_per_key_failure_does_not_abort_pass( + qm, monkeypatch +): + """One key raising out of delete_drained() must not stop the sweep -- + the other drained key is still reclaimed and nothing propagates.""" + line_good = b"good\n" + line_bad = b"bad\n" + await qm.append("s-good", line_good) + await qm.commit("s-good", len(line_good)) + await qm.append("s-bad", line_bad) + await qm.commit("s-bad", len(line_bad)) + + original_delete_drained = qm.delete_drained + + async def _flaky_delete_drained(session_id: str) -> bool: + if session_id == "s-bad": + raise OSError("simulated failure") + return await original_delete_drained(session_id) + + monkeypatch.setattr(qm, "delete_drained", _flaky_delete_drained) + + result = await qm.reclaim_drained_orphans() + + assert result == (1, len(line_good)) + assert not qm._log_path("s-good").exists() + assert qm._log_path("s-bad").exists() # left untouched after the failure diff --git a/tests/test_registry.py b/tests/test_registry.py index 6be2bb4..294098b 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -1149,6 +1149,136 @@ async def test_stale_reap_graph_close_error_still_deregisters( assert sid not in reg._workers +class TestDurableSpoolTrim: + """Change 2 (fix/trim-spool-as-processed): the idle branch of the drain + loop attempts delete_drained() on every idle tick, not just on a clean + session:end -- so disk is reclaimed as sessions go quiet instead of only + at _finalize_session.""" + + async def test_idle_drained_session_trims_log_and_offset_keeps_dead_letters( + self, reg_qm: tuple[SessionRegistry, Any] + ) -> None: + """A fully-drained (committed-to-EOF) idle session gets its .log and + .offset reclaimed by the drain loop's idle branch, WITHOUT a + session:end ever being seen -- and its .dead.jsonl survives.""" + reg, qm = reg_qm + sid = "s-trim" + worker = SessionWorker( + session_id=sid, workspace="/ws", services=HookStateService(workspace="/ws") + ) + worker.services.graph.flush = AsyncMock() # type: ignore[method-assign] + reg._register_for_test(worker) + + # A dead-letter file already exists for this session (e.g. from an + # earlier failure) and must survive the trim -- delete_drained keeps + # .dead.jsonl. + await qm.dead_letter(sid, b"poison", error="boom") + + with patch( + "context_intelligence_server.registry.process_event", + new_callable=AsyncMock, + ): + await qm.append(sid, _line("tool:pre", "/ws", {"session_id": sid})) + # Tiny flush_timeout so the idle branch's reclaim attempt fires + # quickly once the appended line has drained. + task = asyncio.create_task(reg.drain_worker(worker, flush_timeout=0.05)) + for _ in range(200): + await asyncio.sleep(0.02) + if not qm._log_path(sid).exists(): + break + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + assert not qm._log_path(sid).exists() + assert not qm._offset_path(sid).exists() + assert qm._dead_path(sid).exists() # kept -- never touched by trim + + async def test_idle_uncommitted_bytes_are_never_trimmed( + self, reg_qm: tuple[SessionRegistry, Any] + ) -> None: + """A torn (non-newline-terminated) tail is uncommitted, unreadable + bytes that read_batch never surfaces as a record, so the idle branch + fires repeatedly with nothing to dispatch -- but delete_drained + refuses every time (size > committed), so the files are retained, + never silently dropped.""" + reg, qm = reg_qm + sid = "s-trim-uncommitted" + worker = SessionWorker( + session_id=sid, workspace="/ws", services=HookStateService(workspace="/ws") + ) + worker.services.graph.flush = AsyncMock() # type: ignore[method-assign] + reg._register_for_test(worker) + + # A torn tail: bytes with no trailing newline are never a complete + # record (read_batch's readline() breaks on it), so they sit + # uncommitted indefinitely without any drain_worker dispatch at all. + with open(qm._log_path(sid), "ab") as f: + f.write(b"torn-no-newline-yet") + + task = asyncio.create_task(reg.drain_worker(worker, flush_timeout=0.05)) + # Give the idle branch several chances to fire (and refuse) the trim. + await asyncio.sleep(0.3) + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + assert qm._log_path(sid).exists() # never dropped -- delete_drained refused + + async def test_idle_drained_session_trim_logs_at_info( + self, reg_qm: tuple[SessionRegistry, Any], caplog: pytest.LogCaptureFixture + ) -> None: + """The drain loop's idle-branch trim log ("spool_trimmed") fires at + INFO, not DEBUG -- a reclaim that leaves no trace in the journal is + one an operator cannot confirm is happening (fix/trim-spool-as- + processed: 16 files were trimmed on a live box and the log showed + nothing at the old DEBUG level).""" + reg, qm = reg_qm + sid = "s-trim-info" + worker = SessionWorker( + session_id=sid, workspace="/ws", services=HookStateService(workspace="/ws") + ) + worker.services.graph.flush = AsyncMock() # type: ignore[method-assign] + reg._register_for_test(worker) + + with ( + patch( + "context_intelligence_server.registry.process_event", + new_callable=AsyncMock, + ), + caplog.at_level(logging.INFO, logger="context_intelligence_server"), + ): + await qm.append(sid, _line("tool:pre", "/ws", {"session_id": sid})) + task = asyncio.create_task(reg.drain_worker(worker, flush_timeout=0.05)) + for _ in range(200): + await asyncio.sleep(0.02) + if not qm._log_path(sid).exists(): + break + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + assert not qm._log_path(sid).exists() # trim did happen + + info_records = [ + r + for r in caplog.records + if r.levelno == logging.INFO + and "spool_trimmed" in r.getMessage() + and getattr(r, "session_id", None) == sid + ] + assert info_records, ( + f"expected an INFO 'spool_trimmed' record for session={sid}; got " + f"{[(r.levelname, r.getMessage()) for r in caplog.records]}" + ) + + class _AccumBufferGraph: """FAITHFUL model of a real store's accumulating buffer (NOT a hollow mock).