@@ -88,29 +88,53 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
         })
 
     def _source_stop_observed(self, row: dict[str, Any], turn: dict[str, Any],
-                              events: list[dict[str, Any]]) -> bool:
+                              events: list[dict[str, Any]], *,
+                              source_controls: Callable[[], list[dict[str, Any]]] | None = None) -> bool:
         """Read the same durable source controls for recovery and display."""
         return (any(event.get("kind") in {"turn.interrupt_requested", "turn.interrupted"} for event in events)
             or (turn.get("status") in {"failed", "completed"} and any(
                 item.get("binding_id") == row["binding_id"] and item.get("source") == row["source"]
                 and item.get("command") in {"stop", "new"}
                 and datetime.fromisoformat(item["created_at"]) >= datetime.fromisoformat(row["created_at"])
-                for item in self.pending())))
+                for item in (source_controls or self.pending)())))
 
-    def quota_backlog_pending(self, row: dict[str, Any]) -> bool:
+    def quota_backlog_observer(self) -> Callable[[dict[str, Any]], bool]:
+        """Share lazy journal IO within one read-only observation, never admission."""
+        controls: list[dict[str, Any]] | None = None
+
+        def read_controls() -> list[dict[str, Any]]:
+            nonlocal controls
+            if controls is None:
+                controls = self.pending()
+            return controls
+
+        return lambda row: self.quota_backlog_pending(row, source_controls=read_controls)
+
+    def quota_backlog_pending(self, row: dict[str, Any], *,
+                              source_controls: Callable[[], list[dict[str, Any]]] | None = None) -> bool:
         """Project existing work without resolving providers or admitting recovery."""
         from ...control_plane.effect_runtime import effect_runtime_result
 
         store = self.controller.store
         session = store.load_session(row["session_id"])
         turn = store.load_turn(row["session_id"], row["turn_id"])
-        events = store.events_after(row["session_id"], row["turn_id"], None) if turn else []
-        return effect_runtime_result("collaboration.conversation.request", {
+        observation = {"turn": turn, "stop_observed": False}
+        params = {
             "request": {key: value for key, value in row.items() if key != "attachments"},
             "current_session": session,
-            "pending_observation": {"turn": turn,
-                "stop_observed": self._source_stop_observed(row, turn, events) if turn else False},
-        }).get("pending") is True
+            "pending_observation": observation,
+        }
+        # The typed owner first excludes work that is not quota backlog even
+        # without a stop. Stop evidence can only remove a pending request;
+        # never expose a positive result before reading those native facts.
+        if effect_runtime_result("collaboration.conversation.request", params).get("pending") is not True:
+            return False
+        assert turn is not None  # Required by the typed pending projection.
+        events = store.events_after(row["session_id"], row["turn_id"], None)
+        if self._source_stop_observed(row, turn, events, source_controls=source_controls):
+            observation["stop_observed"] = True
+            return effect_runtime_result("collaboration.conversation.request", params).get("pending") is True
+        return True
 
     def context_recovery_candidate(self, request_ref: str) -> bool:
         """Discovery only; every authority and native fact is re-read at admission."""
 
@@ -18,6 +18,7 @@
 from ...capabilities.native_chat.external_conversations import ChatExternalConversations
 from ...chat_store import TERMINAL_TURN_STATES, _atomic_write_json, _read_json
 from ...control_plane.collaboration import conversation_reply_context
+from ...control_plane.effect_runtime import effect_runtime_request_scope
 from ...file_lock import exclusive_file_lock
 from ...presentation.markdown import markdown_scalar
 from ...presentation.renderers.conversation_status_markdown import render_conversation_status
@@ -219,8 +220,10 @@ def profiles(self) -> dict[str, dict[str, str]]:
                 "binding_id": row["binding_id"], "consumer_ref": hashlib.sha256(app_id.encode("utf-8")).hexdigest()[:32]}
         return profiles
 
+    @effect_runtime_request_scope()
     def health(self) -> dict[str, dict[str, int]]:
         health: dict[str, dict[str, int]] = {}
+        quota_pending = self.core.quota_backlog_observer()
         for path in [*self.root.glob("*.json"), *self.root.glob("web-projections/*.json")]:
             row = _read_json(path)
             entry = health.setdefault(row["binding_id"], {"pending_count": 0, "recovery_count": 0})
@@ -235,7 +238,7 @@ def health(self) -> dict[str, dict[str, int]]:
                 if (native.get("status") == "accepted" and native.get("command") is None
                         and all(native.get(key) == row.get(key) for key in
                             ("binding_id", "source", "session_id", "turn_id"))):
-                    pending = self.core.quota_backlog_pending(native)
+                    pending = quota_pending(native)
             if pending:
                 entry["pending_count"] += 1
                 entry["recovery_count"] += int(any(phase.get("started") and not phase.get("verified")
 
@@ -242,7 +242,7 @@ def connect_group(fixture):
 
 
 @pytest.mark.parametrize("info, pending", [("usageLimitExceeded", 1), ("cyberPolicy", 0), (None, 0)])
-def test_delivered_quota_failure_stays_pending_across_restart_without_replaying(ordinary, info, pending):  # noqa: F811
+def test_delivered_quota_failure_stays_pending_across_restart_without_replaying(ordinary, monkeypatch, info, pending):  # noqa: F811
     store, runtime, provider, transport = connect(ordinary)
     fake = ordinary[-2]
     fake.write_text(fake.read_text().replace('"codexErrorInfo": "cyberPolicy"',
@@ -259,6 +259,11 @@ def test_delivered_quota_failure_stays_pending_across_restart_without_replaying(
         transport.reconcile()
         saved = json.loads((transport.root / f"{row['request_ref']}.json").read_text())
         assert saved["status"] == "delivered" and saved["deliveries"]["terminal"]["verified"]
+        if not pending:
+            # The typed owner can exclude completed/non-quota work without
+            # scanning unrelated events or source-control journals.
+            monkeypatch.setattr(store, "events_after", lambda *_: pytest.fail("not quota backlog"))
+            monkeypatch.setattr(transport.core, "pending", lambda: pytest.fail("not quota backlog"))
         assert transport.health()[row["binding_id"]]["pending_count"] == pending
         before = ordinary[4].read_bytes(), list(provider.writes)
         restarted = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root,
@@ -270,6 +275,87 @@ def test_delivered_quota_failure_stays_pending_across_restart_without_replaying(
         runtime.close()
 
 
+@pytest.mark.parametrize("command", ["stop", "new"])
+def test_health_reads_source_controls_once_and_keeps_fresh_recovery(ordinary, monkeypatch, command):  # noqa: F811
+    from loopx.capabilities.native_chat import external_conversations
+    from loopx.control_plane import effect_runtime
+
+    store, runtime, provider, transport = connect(ordinary)
+    fake = ordinary[-2]
+    fake.write_text(fake.read_text().replace('"codexErrorInfo": "cyberPolicy"',
+        '"codexErrorInfo": "usageLimitExceeded"').replace(
+        '            print(json.dumps({"method": "item/agentMessage/delta", "params": {"threadId": "durable-thread", "turnId": active_turn, "delta": "Partial answer."}}), flush=True)\n', ''))
+    try:
+        binding = transport._binding("notes-app")
+        rows = []
+        for index in range(8):
+            event = provider.event("notes-app", f"health-quota-{index}", "typed terminal failure")
+            transport.admit("notes-app", event)
+            row = transport.core.read_request(identity_ref(binding["provider_ref"], event["message_id"]))
+            turn = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=10)
+            assert turn["status"] == "failed" and turn["error_code"] == "usage_limit_exceeded"
+            transport.reconcile()
+            rows.append(row)
+
+        def durable_state():
+            paths = {path for root in (transport.root, transport.core.root) for path in root.glob("*.json")}
+            paths.update(path for path in store.sessions_root.rglob("*") if path.suffix in {".json", ".jsonl"})
+            return {path: path.read_bytes() for path in paths}
+
+        before = durable_state()
+        effects = ordinary[4].read_bytes(), list(provider.writes)
+        reads = []
+        revisions = []
+        read_json = external_conversations._read_json
+        fingerprint = effect_runtime._runtime_fingerprint
+
+        def counted_read(path):
+            if path.parent == transport.core.root and re.fullmatch(r"[a-f0-9]{24}", path.stem):
+                reads.append(path)
+            return read_json(path)
+
+        monkeypatch.setattr(external_conversations, "_read_json", counted_read)
+        def counted_revision():
+            revisions.append(True)
+            return fingerprint()
+
+        monkeypatch.setattr(effect_runtime, "_runtime_fingerprint", counted_revision)
+        expected = {"pending_count": len(rows), "recovery_count": 0}
+        assert transport.health()[binding["binding_id"]] == expected
+        # Correlation reads plus one complete control observation, rather than
+        # a complete journal scan for each historical terminal request.
+        assert len(reads) <= 3 * len(rows)
+        assert len(revisions) <= 1  # One managed runtime revision for the health request.
+        assert durable_state() == before
+        assert (ordinary[4].read_bytes(), provider.writes) == effects
+
+        # Durable control observations must keep all three existing fences.
+        # These synthetic journal records are inputs, not new admissions.
+        for index, change in enumerate([
+            {"binding_id": "f" * 24},
+            {"source": dict(rows[0]["source"], source_ref="e" * 24)},
+            {"created_at": "2000-01-01T00:00:00+00:00"},
+        ]):
+            control_row = dict(rows[0], command=command, **change)
+            (transport.core.root / f"{index + 1:024x}.json").write_text(json.dumps(control_row))
+        # The journal reader already tolerates damaged JSON and ignores
+        # non-request metadata; observation reuse must retain that behavior.
+        (transport.core.root / ("d" * 24 + ".json")).write_text("{")
+        (transport.core.root / "metadata.json").write_text(json.dumps(dict(rows[0], command=command)))
+        assert transport.health()[binding["binding_id"]] == expected
+        assert all(transport.core.quota_backlog_pending(row) for row in rows)
+
+        # An actual later control must reach both a new health observation and
+        # the default recovery check, independently of the previous snapshot.
+        control = provider.event("notes-app", "health-control", f"/{command}")
+        transport.admit("notes-app", control)
+        assert transport.health()[binding["binding_id"]]["pending_count"] == 1  # undelivered control
+        assert all(not transport.core.quota_backlog_pending(row) for row in rows)
+        assert all(not transport.core.context_recovery_candidate(row["request_ref"]) for row in rows)
+    finally:
+        runtime.close()
+
+
 @pytest.mark.parametrize("fault", [None, "changed-source", "revoked", "stop", "unknown-native",
     "background", "later-user", "reservation-crash", "post-enqueue-crash", "queued-newer", "queued-input", "private-path", "still-incomplete"])
 def test_original_worker_continues_explicit_unfinished_project_work_once_after_restart(ordinary, monkeypatch, fault):  # noqa: F811