diff --git a/context_intelligence_server/main.py b/context_intelligence_server/main.py
index 0fbaac5..6b8c2c3 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 282de39..d351fb8 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 b82e8e8..e5c1886 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 f149432..fe5ea48 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 6276bbc..3956241 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 b52f75b..e28f506 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 '<empty>'}\n"
+            f"--- partial stderr ---\n{exc.stderr or '<empty>'}"
+        ) from exc
 
 
 def _clean_env(tmp_path: Path) -> dict[str, str]:
@@ -190,7 +238,9 @@ def test_serve_still_fails_loud_for_a_real_misconfiguration(
         # construction (inside the worker's load()), before ever binding
         # matters for the assertion.
         env["AMPLIFIER_CONTEXT_INTELLIGENCE_SERVER_SERVER_PORT"] = "18321"
-        result = _run_cli(["serve"], cwd=PROJECT_ROOT, env=env)
+        result = _run_cli(
+            ["serve"], cwd=PROJECT_ROOT, env=env, timeout=_SERVE_CLI_TIMEOUT_S
+        )
         assert result.returncode != 0
         assert "neo4j_require_explicit_clients" in result.stdout + result.stderr
 
diff --git a/tests/test_main.py b/tests/test_main.py
index abc2f0e..1cb77fb 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 54ef4af..4e26592 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 294098b..3ae26ae 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