fix(chat): amortize historical recovery discovery reads

examples/capability-extension-registry-smoke.py

@@ -84,6 +84,7 @@ def run_cli(runtime_root: Path, *args: str) -> dict[str, object]:
         "reliability-diagnostics",
         "progress-review-sentinel",
         "goal-capability-organization",
+        "anti-entropy-governance",
     ]
     assert all(item["provider_id"] == "loopx-core" for item in builtin_capabilities)
     value_summary = next(

loopx/capabilities/native_chat/external_conversations.py

@@ -38,6 +38,37 @@ def _commission_goal_identity(goal: dict[str, Any]) -> dict[str, str]:
     return identity
 
 
+class _RecoveryDiscoveryReads:
+    """Lazy facts for one queue-discovery pass, never recovery admission.
+
+    Historical requests share Session history. Retain queue/control/history
+    hints, never provider identity or a recovery decision. A new
+    pass builds new observations; normal admission never receives this object.
+    """
+
+    def __init__(self, core: ChatExternalConversations) -> None:
+        self.core = core
+        self.queues: dict[str, bool] = {}
+        self.histories: dict[str, list[dict[str, Any]]] = {}
+        self.controls: list[dict[str, Any]] | None = None
+
+    def queued(self, session_id: str) -> bool:
+        if session_id not in self.queues:
+            self.queues[session_id] = bool(self.core.controller.store.queued_turns(session_id))
+        return self.queues[session_id]
+
+    def user_messages(self, session_id: str) -> list[dict[str, Any]]:
+        if session_id not in self.histories:
+            self.histories[session_id] = [message for message in self.core.controller.store.messages(session_id)
+                if message.get("role") == "user"]
+        return self.histories[session_id]
+
+    def source_controls(self) -> list[dict[str, Any]]:
+        if self.controls is None:
+            self.controls = self.core.pending()
+        return self.controls
+
+
 class ChatExternalConversations:
     def __init__(self, controller: Any) -> None:
         self.controller = controller
@@ -52,10 +83,13 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                                selected_binding: dict[str, Any] | None = None,
                                discovery_only: bool = False,
                                preparation_scope_available: bool = False,
-                               metadata_capacity: dict[str, Any] | None = None) -> dict[str, Any]:
+                               metadata_capacity: dict[str, Any] | None = None,
+                               discovery_reads: _RecoveryDiscoveryReads | None = None) -> dict[str, Any]:
         from ...chat_manager import MANAGER_CONTEXT_VERSION
         from ...control_plane.effect_runtime import effect_runtime_result
 
+        if discovery_reads is not None and not discovery_only:
+            raise ValueError("discovery observations cannot authorize recovery")
         store = self.controller.store
         session = store.load_session(str(row.get("session_id") or ""))
         turn = store.load_turn(str(row.get("session_id") or ""), str(row.get("turn_id") or ""))
@@ -99,18 +133,21 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
             later_equivalent = any(item.get("role") == "user" and item.get("turn_id") != turn["turn_id"]
                 and item.get("text") == row.get("message") and item.get("origin") in {None, "lark", "web"}
                 and datetime.fromisoformat(item["created_at"]) > created
-                for item in store.messages(session["session_id"]))
+                for item in (discovery_reads.user_messages(session["session_id"])
+                             if discovery_reads is not None else store.messages(session["session_id"])))
         from ...chat_manager import steward_machine_defaults
         return effect_runtime_result("collaboration.conversation.request", {
             "request": {key: value for key, value in row.items() if key != "attachments"},
             "current_session": session, "binding": selected_binding,
             "context_recovery": {"version": MANAGER_CONTEXT_VERSION, "failed_turn": turn,
                 "observed_at": datetime.now(timezone.utc).isoformat(),
                 "admitted_turn": admitted,
-                "queued_input_present": bool(store.queued_turns(session["session_id"])),
+                "queued_input_present": (discovery_reads.queued(session["session_id"])
+                    if discovery_reads is not None else bool(store.queued_turns(session["session_id"]))),
                 "attachments_present": bool(row.get("attachments")),
                 "steering_observed": any(event.get("kind") == "turn.steered" for event in events),
-                "stop_observed": self._source_stop_observed(row, turn, events),
+                "stop_observed": self._source_stop_observed(row, turn, events,
+                    source_controls=discovery_reads.source_controls if discovery_reads is not None else None),
                 "preparation_observation": preparation,
                 "later_equivalent_input_present": later_equivalent,
                 "preparation_scope_available": preparation_scope_available,
@@ -187,7 +224,13 @@ def quota_backlog_pending(self, row: dict[str, Any], *,
             return effect_runtime_result("collaboration.conversation.request", params).get("pending") is True
         return True
 
-    def context_recovery_candidate(self, request_ref: str) -> bool:
+    def context_recovery_observer(self) -> Callable[[str], bool]:
+        """Reuse scan-local hints; reread each request, Turn and binding."""
+        reads = _RecoveryDiscoveryReads(self)
+        return lambda ref: self.context_recovery_candidate(ref, discovery_reads=reads)
+
+    def context_recovery_candidate(self, request_ref: str, *,
+                                   discovery_reads: _RecoveryDiscoveryReads | None = None) -> bool:
         """Discovery only; every authority and native fact is re-read at admission."""
         try:
             row = self.read_request(request_ref)
@@ -209,7 +252,8 @@ def context_recovery_candidate(self, request_ref: str) -> bool:
                 invalid_metadata = ((turn.get("response") or {}).get("gate") or {}).get("kind") == "response_metadata_invalid"
                 if not open_literal and not invalid_metadata:
                     return False  # A lexical probe hint, never permission to adopt or send.
-            return self._context_recovery_plan(row, discovery_only=True).get("recovery_candidate") is True
+            return self._context_recovery_plan(row, discovery_only=True,
+                discovery_reads=discovery_reads).get("recovery_candidate") is True
         except (KeyError, ValueError, RuntimeError, OSError):
             return False
 

loopx/extensions/lark/private_conversations.py

@@ -1290,9 +1290,11 @@ def inbox() -> Path:
             except (KeyError, ValueError, OSError):
                 return  # Unknown authority or provider effects stay pending.
 
+    @effect_runtime_request_scope()
     def pending_delivery_paths(self) -> list[Path]:
         """The durable transport store is the queue; no in-memory admission."""
         paths = []
+        recovery_candidate = self.core.context_recovery_observer()
         for path in sorted(self.root.glob("*.json")):
             row = _read_json(path)
             if row["status"] != "delivered":
@@ -1303,7 +1305,7 @@ def pending_delivery_paths(self) -> list[Path]:
                 # The canonical request, provider effect inventory and source
                 # are independently re-read by recover_failed_request.
                 self._context_recovery_probe_at[path] = time.monotonic() + 60
-                if legacy_recovery_repairs(row) or self.core.context_recovery_candidate(row["request_ref"]):
+                if legacy_recovery_repairs(row) or recovery_candidate(row["request_ref"]):
                     paths.append(path)
         return [*paths, *self._web_projection_paths()]
 

tests/test_context_recovery_discovery.py

@@ -0,0 +1,164 @@
+"""Discovery amortizes history IO; execution still needs current authority."""
+from collections import Counter
+from datetime import datetime, timedelta
+from pathlib import Path
+
+import pytest
+
+from test_native_steward_private import steward, finish  # noqa: F401
+from loopx.capabilities.native_chat import external_conversations
+from loopx.extensions.lark.conversation_identity import identity_ref
+
+
+@pytest.fixture
+def failed_source(steward):  # noqa: F811
+    store, runtime, provider, transport, binding, capture, _ = steward
+    fake = Path(runtime.codex_bin)
+    fake.write_text(fake.read_text().replace('"codexErrorInfo": "cyberPolicy"',
+        '"codexErrorInfo": "contextWindowExceeded"').replace(
+        '            print(json.dumps({"method": "item/agentMessage/delta", "params": {"threadId": "durable-thread", "turnId": active_turn, "delta": "Partial answer."}}), flush=True)\n', ''))
+    event = provider.event("steward-app", "discovery-history", "typed terminal failure")
+    transport.admit("steward-app", event)
+    ref = identity_ref(binding["provider_ref"], event["message_id"])
+    row = transport.core.read_request(ref)
+    assert finish(runtime, row)["status"] == "failed"
+    store.update_session(row["session_id"], manager_context_version=20)
+    assert transport.reconcile() == 1
+    return store, runtime, provider, transport, binding, capture, ref, row
+
+
+def test_one_discovery_pass_reuses_queue_history_and_controls(failed_source, monkeypatch):
+    store, _, provider, transport, _, capture, ref, row = failed_source
+    # Model a retained pre-dispatch failure after its existing backoff. Discovery
+    # is only a hint: the original native event sequence still owns admission.
+    store.update_turn(row["session_id"], row["turn_id"], upstream_turn_id=None,
+        error_code="manager_authorization_unavailable", delta_count=0, response=None)
+    real_datetime = datetime
+    class AfterBackoff(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return real_datetime.now(tz) + timedelta(minutes=2)
+    monkeypatch.setattr(external_conversations, "datetime", AfterBackoff)
+    counts = Counter()
+    for owner, name in [(store, "queued_turns"), (store, "messages"), (transport.core, "pending")]:
+        original = getattr(owner, name)
+        def counted(*args, _name=name, _read=original, **kwargs):
+            counts[_name] += 1
+            return _read(*args, **kwargs)
+        monkeypatch.setattr(owner, name, counted)
+    before_calls, before_requests = len(provider.calls), capture.read_bytes()
+    observe = transport.core.context_recovery_observer()
+    assert [observe(ref) for _ in range(4)] == [True] * 4
+    assert counts == {"queued_turns": 1, "messages": 1, "pending": 1}
+    # The next pass rereads rather than retaining a negative/positive cache.
+    assert transport.core.context_recovery_observer()(ref)
+    assert counts == {"queued_turns": 2, "messages": 2, "pending": 2}
+    assert len(provider.calls) == before_calls and capture.read_bytes() == before_requests
+
+
+@pytest.mark.parametrize("change", ["queued-input", "revoked", "owner-unavailable"])
+def test_discovery_hint_never_authorizes_recovery(failed_source, monkeypatch, change):
+    store, _, provider, transport, binding, capture, ref, row = failed_source
+    observe = transport.core.context_recovery_observer()
+    assert observe(ref)
+    original = store._turn_path(row["session_id"], row["turn_id"]).read_bytes()
+    before = capture.read_bytes()
+    if change == "queued-input":
+        store.create_queued_turn(row["session_id"], client_turn_id="new-original-input",
+            message="Resolve the newer input first", origin="web")
+        queued = store.queued_turns
+        observed = []
+        deny_after_read = False
+        def current_queue(session_id):
+            result = queued(session_id)
+            observed.append(bool(result))
+            if deny_after_read:
+                provider.owner_ready = False
+            return result
+        monkeypatch.setattr(store, "queued_turns", current_queue)
+        transport.core.context_recovery_observer()(ref)
+        assert observed == [True], "The next observation reads the new queue"
+        # A candidate is not an eligibility verdict. Normal admission must read
+        # the new queue itself, then reject authority lost during that read.
+        deny_after_read = True
+        with pytest.raises(ValueError):
+            transport.core.recover_failed_request(ref,
+                source_observation=lambda _selected: pytest.fail("source IO after denied authority"))
+        assert observed == [True, True]
+    else:
+        if change == "revoked":
+            transport.bindings.disconnect(binding["binding_id"],
+                expected_revision=transport.bindings.read()["revision"])
+            assert not observe(ref), "Even the same discovery pass rereads the binding"
+        else:
+            provider.owner_ready = False
+        with pytest.raises(ValueError):
+            transport.core.recover_failed_request(ref,
+                source_observation=lambda _selected: pytest.fail("source IO after denied authority"))
+    assert capture.read_bytes() == before
+    assert store._turn_path(row["session_id"], row["turn_id"]).read_bytes() == original
+
+
+def test_unavailable_queue_is_not_cached_as_empty(failed_source, monkeypatch):
+    store, _, _, transport, _, _, ref, _ = failed_source
+    original = store.queued_turns
+    calls = 0
+    def flaky(session_id):
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            raise OSError("synthetic unavailable queue")
+        return original(session_id)
+    monkeypatch.setattr(store, "queued_turns", flaky)
+    observe = transport.core.context_recovery_observer()
+    assert not observe(ref)
+    assert observe(ref)
+    assert calls == 2
+
+
+def test_admission_rejects_discovery_observations(failed_source):
+    _, _, _, transport, _, _, _, row = failed_source
+    reads = external_conversations._RecoveryDiscoveryReads(transport.core)
+    with pytest.raises(ValueError, match="cannot authorize recovery"):
+        transport.core._context_recovery_plan(row, discovery_reads=reads)
+
+
+def test_discovery_facts_do_not_cross_sessions(failed_source):
+    store, _, _, transport, _, _, _, row = failed_source
+    second = store.create_session(goal_id="separate-synthetic-goal", agent_id="codex", channel_id="separate-source",
+        adapter_kind="codex_app_server", upstream_thread_id="separate-synthetic-thread")
+    queued, created = store.create_queued_turn(second["session_id"], client_turn_id="other-source-input",
+        message="Only the second source has queued input", origin="web")
+    assert created
+    reads = external_conversations._RecoveryDiscoveryReads(transport.core)
+    assert not reads.queued(row["session_id"])
+    assert reads.queued(second["session_id"])
+    assert {message["turn_id"] for message in reads.user_messages(second["session_id"])} == {queued["turn_id"]}
+    assert all(message["turn_id"] != queued["turn_id"] for message in reads.user_messages(row["session_id"]))
+
+
+def test_lark_scanner_shares_only_one_pass_and_pins_one_code_revision(failed_source, monkeypatch):
+    from loopx.control_plane import effect_runtime
+    store, runtime, provider, transport, binding, _, ref, row = failed_source
+    event = provider.event("steward-app", "second-discovery-history", "typed terminal failure")
+    transport.admit("steward-app", event)
+    second = identity_ref(binding["provider_ref"], event["message_id"])
+    next_row = transport.core.read_request(second)
+    assert finish(runtime, next_row)["status"] == "failed"
+    assert transport.reconcile_request(transport.root / f"{second}.json") == 1
+    store.update_session(row["session_id"], manager_context_version=20)
+    calls = Counter()
+    for owner, name in [(store, "queued_turns"), (effect_runtime, "_runtime_fingerprint")]:
+        original = getattr(owner, name)
+        def counted(*args, _name=name, _read=original, **kwargs):
+            calls[_name] += 1
+            return _read(*args, **kwargs)
+        monkeypatch.setattr(owner, name, counted)
+    provider_calls = len(provider.calls)
+    expected = {transport.root / f"{request}.json" for request in (ref, second)}
+    for number in (1, 2):
+        # Advance the existing discovery clock, not a new retry mechanism.
+        transport._context_recovery_probe_at.clear()
+        assert set(transport.pending_delivery_paths()) == expected
+        assert calls == {"queued_turns": number, "_runtime_fingerprint": number}
+    assert len(provider.calls) == provider_calls