From 6e1f555ab05ad195cd6182f51673d3cc48c23c2d Mon Sep 17 00:00:00 2001 From: colombod Date: Thu, 10 Sep 2026 03:26:53 +0000 Subject: [PATCH] fix: /status endpoint must not walk the spool on every request MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #101 claimed '/status is now O(1) regardless of spool size', but the claim was false. The fix moved spool_stats() off the request path but missed a second unbounded scan reached via Registry.pipeline_metrics() -> QueueManager.derive_all_stats(). This path walked every worker key inline, opening and reading each .offset file and .dead.jsonl tail on each request, gated only by a 1-second READ-SIDE TTL that spared the second caller but never the first. On a healthy replica with 5000+ worker keys, /status timed out (60s/120s/150s+) while other endpoints responded normally. This fix mirrors the split already applied to spool_stats: - refresh_all_stats() — async, performs the full scan, populates cache, no TTL gate, the only operation that touches disk for these metrics - derive_all_stats() — now SYNCHRONOUS and cache-only; never touches filesystem, never raises - all_stats_age_seconds() — reports snapshot age (None before first refresh) - _spool_stats_refresher now drives BOTH refreshes each cycle Cold-window correctness: before the first refresh, in_queue and dead stats are UNKNOWN (not zero), so naive reads would fabricate loss. The snapshot carries stats_available: False, and pipeline_metrics suppresses rather than invents: residual is None when unavailable, degraded is only computed when stats are available, and _residual_positive_since is not latched on unknown data. /status now reports metrics.stats_available and metrics.as_of_seconds for full visibility. SECONDARY: test_serve_still_fails_loud_for_a_real_misconfiguration flaked on main (15s TimeoutExpired). Not a guard regression — the test passed twice on branch and fires locally in 0.74s. On the runner, the subprocess stalled during gunicorn startup, never reaching worker boot. Subprocess budget raised to 90s (zero cost on healthy runs) and timeout now raises AssertionError carrying partial stdout/stderr and naming it as wall-clock timeout, not proof of guard failure. Version: 6.7.5 -> 6.7.6 Generated with Amplifier Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- context_intelligence_server/main.py | 5 ++ context_intelligence_server/queue_manager.py | 63 ++++++++++++++--- context_intelligence_server/registry.py | 28 ++++++-- pyproject.toml | 2 +- tests/neo4j/test_queues_actions.py | 4 ++ tests/test_lazy_asgi_app.py | 74 ++++++++++++++++---- tests/test_main.py | 1 + tests/test_queue_manager.py | 74 ++++++++++++++++---- tests/test_registry.py | 56 +++++++++++++++ 9 files changed, 266 insertions(+), 41 deletions(-) diff --git a/context_intelligence_server/main.py b/context_intelligence_server/main.py index 0fbaac57..6b8c2c37 100644 --- a/context_intelligence_server/main.py +++ b/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: diff --git a/context_intelligence_server/queue_manager.py b/context_intelligence_server/queue_manager.py index 282de398..d351fb8d 100644 --- a/context_intelligence_server/queue_manager.py +++ b/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. diff --git a/context_intelligence_server/registry.py b/context_intelligence_server/registry.py index b82e8e8a..e5c1886a 100644 --- a/context_intelligence_server/registry.py +++ b/context_intelligence_server/registry.py @@ -255,10 +255,19 @@ 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 ) @@ -266,7 +275,10 @@ async def pipeline_metrics(self) -> dict[str, Any]: # 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,7 +292,10 @@ 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"], @@ -288,8 +303,13 @@ async def pipeline_metrics(self) -> dict[str, Any]: "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( diff --git a/pyproject.toml b/pyproject.toml index f1494325..fe5ea48f 100644 --- a/pyproject.toml +++ b/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 = [ diff --git a/tests/neo4j/test_queues_actions.py b/tests/neo4j/test_queues_actions.py index 6276bbc1..39562412 100644 --- a/tests/neo4j/test_queues_actions.py +++ b/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 diff --git a/tests/test_lazy_asgi_app.py b/tests/test_lazy_asgi_app.py index b52f75bd..e28f506c 100644 --- a/tests/test_lazy_asgi_app.py +++ b/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 ''}\n" + f"--- partial stderr ---\n{exc.stderr or ''}" + ) 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 diff --git a/tests/test_main.py b/tests/test_main.py index abc2f0ee..1cb77fbf 100644 --- a/tests/test_main.py +++ b/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 diff --git a/tests/test_queue_manager.py b/tests/test_queue_manager.py index 54ef4af9..4e26592f 100644 --- a/tests/test_queue_manager.py +++ b/tests/test_queue_manager.py @@ -356,7 +356,8 @@ async def test_derive_all_stats_counts_pending_and_dead(qm): # s2: no pending log data, one dead letter. await qm.dead_letter("s2", b"poison", error="boom") - stats = await qm.derive_all_stats() + await qm.refresh_all_stats() + stats = qm.derive_all_stats() assert stats["in_queue_total"] == 2 assert stats["dead_total"] == 1 @@ -402,7 +403,15 @@ async def test_purge_dead_letters_rejects_unsafe_session_id(qm, bad_id): await qm.purge_dead_letters(bad_id) -async def test_derive_all_stats_caches_within_ttl(qm, monkeypatch): +async def test_refresh_all_stats_always_scans_no_ttl_gate(qm, monkeypatch): + """refresh_all_stats() has NO TTL gate -- every call re-scans the spool. + + INCIDENT 2026-09-10: the old derive_all_stats() scanned inline on every + /status call, gated only by a 1-second read-side TTL -- which spares the + SECOND caller and never the first. On a ~5000-key spool on Azure Files + SMB that made /status time out at 60-180s while /version answered in 0s + on the same replica. Cadence is now owned entirely by the background + refresher loop (main.py._spool_stats_refresher), not a read-side TTL.""" await qm.append("s1", b"a") calls = {"n": 0} @@ -414,16 +423,48 @@ def counting(): monkeypatch.setattr(qm, "_all_worker_keys", counting) - await qm.derive_all_stats() - await qm.derive_all_stats() # within TTL -> served from cache - assert calls["n"] == 1 - - # Age the cache past the TTL; the next call must recompute. - qm._stats_cache_at = time.monotonic() - (qm._stats_cache_ttl + 1.0) - await qm.derive_all_stats() + await qm.refresh_all_stats() + await qm.refresh_all_stats() # no TTL -- scans again every time assert calls["n"] == 2 +def test_derive_all_stats_is_synchronous_read_only_and_never_scans(qm, monkeypatch): + """derive_all_stats() must NEVER touch the filesystem: it is a pure, + synchronous cache read. An empty cache returns the unavailable sentinel + (stats_available False, zeroed aggregates, never scans, never raises); a + populated cache returns that snapshot plus stats_available True -- this + is the entire point of the refresh/read split that fixes /status + stalling on a large spool (see the module docstring / incident above).""" + import pathlib + + def _boom(self: pathlib.Path) -> None: + raise AssertionError("derive_all_stats() touched the filesystem via glob()") + + monkeypatch.setattr(pathlib.Path, "glob", _boom) + + # Cold cache: the sentinel, no scan, no raise. + assert qm.derive_all_stats() == { + "per_key": [], + "in_queue_total": 0, + "dead_total": 0, + "stats_available": False, + } + + # Populated cache: returns exactly that snapshot plus stats_available, + # still without scanning. + qm._stats_cache = { + "per_key": [{"worker_key": "s1", "in_queue": 2, "dead": 0}], + "in_queue_total": 2, + "dead_total": 0, + } + assert qm.derive_all_stats() == { + "per_key": [{"worker_key": "s1", "in_queue": 2, "dead": 0}], + "in_queue_total": 2, + "dead_total": 0, + "stats_available": True, + } + + # --- recovery_seed_counts: residual-0-by-construction crash-recovery seed --- @@ -476,7 +517,8 @@ async def test_recovery_seed_counts_residual_is_zero_mixed_shape(qm): await qm.dead_letter("c", b"poison", error="boom") accepted, written = await qm.recovery_seed_counts() - stats = await qm.derive_all_stats() + await qm.refresh_all_stats() + stats = qm.derive_all_stats() residual = accepted - written - stats["in_queue_total"] - stats["dead_total"] assert residual == 0 @@ -495,7 +537,8 @@ async def test_recovery_seed_counts_crash_window_residual_zero(qm): assert written == 0 # NOT -1 (the crash-window trap) assert accepted == 2 # written_seed(0) + pending(1) + dead(1) - stats = await qm.derive_all_stats() + await qm.refresh_all_stats() + stats = qm.derive_all_stats() residual = accepted - written - stats["in_queue_total"] - stats["dead_total"] assert residual == 0 @@ -549,7 +592,8 @@ async def test_recovery_reconcile_then_seed_keeps_residual_zero(qm): await qm.recovery_reconcile_dead() accepted, written = await qm.recovery_seed_counts() - stats = await qm.derive_all_stats() + await qm.refresh_all_stats() + stats = qm.derive_all_stats() residual = accepted - written - stats["in_queue_total"] - stats["dead_total"] assert residual == 0 @@ -569,7 +613,8 @@ async def test_recovery_seed_counts_replay_window_residual_zero(qm): # written_seed = max(0, 1-1)=0; accepted_seed = 0 + 1 + 1 = 2 assert (accepted, written) == (2, 0) - stats = await qm.derive_all_stats() + await qm.refresh_all_stats() + stats = qm.derive_all_stats() residual = accepted - written - stats["in_queue_total"] - stats["dead_total"] assert residual == 0 @@ -968,7 +1013,8 @@ async def test_recovery_seed_counts_residual_zero_after_trimming_dead_letter_ses assert qm._dead_path("s1").exists() accepted, written = await qm.recovery_seed_counts() - stats = await qm.derive_all_stats() + await qm.refresh_all_stats() + stats = qm.derive_all_stats() residual = accepted - written - stats["in_queue_total"] - stats["dead_total"] assert residual == 0 diff --git a/tests/test_registry.py b/tests/test_registry.py index 294098b4..3ae26aee 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -1608,6 +1608,7 @@ async def test_metrics_block_shape_and_residual( reg.record_replayed(1) reg.record_write_retry() + await reg.queue_manager.refresh_all_stats() metrics = await reg.pipeline_metrics() assert metrics["accepted_total"] == 5 @@ -1638,6 +1639,7 @@ async def test_nonzero_residual_is_degraded( monkeypatch.setattr(registry_module, "_RESIDUAL_DEGRADED_GRACE", 0.0) reg.seed_counters(accepted=4, written=1) + await reg.queue_manager.refresh_all_stats() metrics = await reg.pipeline_metrics() assert metrics["in_queue_total"] == 0 @@ -1657,6 +1659,7 @@ async def test_dead_letters_force_degraded_even_at_zero_residual( reg.seed_counters(accepted=1, written=0) + await reg.queue_manager.refresh_all_stats() metrics = await reg.pipeline_metrics() assert metrics["residual"] == 0 @@ -1664,6 +1667,55 @@ async def test_dead_letters_force_degraded_even_at_zero_residual( assert metrics["degraded"] is True +# --------------------------------------------------------------------------- +# Cold-window contract (incident 2026-09-10): before refresh_all_stats() has +# ever produced a snapshot, pipeline_metrics() must not fabricate numbers -- +# it must suppress residual/degraded entirely rather than treat unknown +# in_queue/dead as zero. Without this, a hanging endpoint is merely traded +# for a lying one. See QueueManager.derive_all_stats / registry.pipeline_metrics. +# --------------------------------------------------------------------------- + + +class TestPipelineMetricsColdWindow: + async def test_cold_cache_suppresses_residual_and_does_not_latch_grace_timer( + self, reg_qm: tuple[SessionRegistry, Any] + ) -> None: + """With a COLD stats cache (refresh_all_stats() never called), even + a live-counter state that would otherwise imply a large positive + residual must be suppressed entirely: stats_available False, + residual None, degraded False, as_of_seconds None. Critically, + _residual_positive_since must NOT be latched by this cold call -- + the grace-period timer must not start on unknown data.""" + reg, _qm = reg_qm + # Would imply residual == 100 if the cold aggregate were trusted. + reg.seed_counters(accepted=100, written=0) + + metrics = await reg.pipeline_metrics() + + assert metrics["stats_available"] is False + assert metrics["residual"] is None + assert metrics["degraded"] is False + assert metrics["as_of_seconds"] is None + assert reg._residual_positive_since is None + + async def test_after_refresh_reports_available_stats_and_age( + self, reg_qm: tuple[SessionRegistry, Any] + ) -> None: + """After await refresh_all_stats(), pipeline_metrics() reports + stats_available True, a real numeric residual, and as_of_seconds as + a float >= 0 -- the snapshot is no longer unknown.""" + reg, qm = reg_qm + reg.seed_counters(accepted=3, written=1) + + await qm.refresh_all_stats() + metrics = await reg.pipeline_metrics() + + assert metrics["stats_available"] is True + assert metrics["residual"] == 2 + assert isinstance(metrics["as_of_seconds"], float) + assert metrics["as_of_seconds"] >= 0.0 + + # --------------------------------------------------------------------------- # FIX B: degraded false-positive fix. A bare dead-letter purge (record_purged) # keeps the residual conserved at zero, a NEGATIVE residual (transient @@ -1696,6 +1748,7 @@ async def test_purge_decrements_accepted_residual_stays_zero( assert reg.pipeline_counters()["accepted_total"] == 0 + await qm.refresh_all_stats() metrics = await reg.pipeline_metrics() assert metrics["residual"] == 0 assert metrics["degraded"] is False @@ -1715,6 +1768,7 @@ async def test_transient_in_queue_not_degraded( reg.record_written(2) await qm.append(sid, _line("tool:pre", "/ws", {"session_id": sid})) + await qm.refresh_all_stats() metrics = await reg.pipeline_metrics() assert metrics["residual"] == -1 @@ -1731,6 +1785,7 @@ async def test_dead_letter_reports_degraded( await qm.dead_letter(sid, _line("bad", "/ws", {"session_id": sid}), "boom") reg.seed_counters(accepted=1, written=0) + await qm.refresh_all_stats() metrics = await reg.pipeline_metrics() assert metrics["dead_letter_total"] >= 1 @@ -1747,6 +1802,7 @@ async def test_sustained_drop_reports_degraded( monkeypatch.setattr(registry_module, "_RESIDUAL_DEGRADED_GRACE", 0.0) reg.seed_counters(accepted=2, written=1) + await reg.queue_manager.refresh_all_stats() metrics = await reg.pipeline_metrics() assert metrics["residual"] == 1