fix(chat): avoid repeated source scans during recovery discovery

loopx/capabilities/native_chat/external_conversations.py

@@ -181,7 +181,8 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                 "later_source_input_present": any(item.get("binding_id") == row["binding_id"]
                     and item.get("source") == row["source"] and item.get("command") is None
                     and datetime.fromisoformat(item["created_at"]) > datetime.fromisoformat(row["created_at"])
-                    for item in self.pending()) if turn.get("error_code") == "antigravity_step_error" else None,
+                    for item in (discovery_reads.source_controls() if discovery_reads is not None
+                                 else self.pending())) if turn.get("error_code") == "antigravity_step_error" else None,
                 "native_input": anchors[0] if len(anchors) == 1 else None,
                 "response_literal_open": bool(open_literal),
                 "metadata_capacity": metadata_capacity,

tests/test_lark_private_conversations.py

@@ -118,7 +118,7 @@ def committed_then_disconnected(**kwargs):
 
 
 @pytest.mark.parametrize("fault", [None, "stop", "source-changed", "later-source", "no-capacity", "native-tool",
-    "stop-after-discovery", "source-changed-after-discovery"])
+    "stop-after-discovery", "source-changed-after-discovery", "later-source-after-discovery"])
 def test_original_worker_continues_proven_unstarted_argon_quota_on_codex(tmp_path, language_server, monkeypatch, fault):  # noqa: F811
     from copy import deepcopy
     from datetime import timedelta
@@ -180,20 +180,36 @@ def checked_update(session_id, **changes):
                 transitions.append(session_id)
             return update_session(session_id, **changes)
         monkeypatch.setattr(store, "update_session", checked_update)
+        if fault in {None, "later-source-after-discovery"}:
+            pending = transport.core.pending
+            scans = []
+            def counted_pending():
+                scans.append(True)
+                return pending()
+            monkeypatch.setattr(transport.core, "pending", counted_pending)
+            discovery = transport.core.context_recovery_observer()
+            assert discovery(ref) and discovery(ref)
+            assert len(scans) == 1  # One full source inventory per discovery pass.
         if fault in {"stop-after-discovery", "source-changed-after-discovery"}:
             assert path in transport.pending_delivery_paths()
         if fault in {"stop", "stop-after-discovery"}:
             store.append_event(original["session_id"], original["turn_id"], kind="turn.interrupt_requested", payload={})
             original_events = store.events_after(original["session_id"], original["turn_id"], None)
         elif fault in {"source-changed", "source-changed-after-discovery"}:
             provider.messages[event["message_id"]]["content"] = "Changed original input"
-        elif fault == "later-source":
+        elif fault in {"later-source", "later-source-after-discovery"}:
             transport.admit("steward-app", provider.event("steward-app", "source-correction", "/status"))
             # A later ordinary source is observed without creating model work.
             later = deepcopy(original)
             later.update(request_ref="e" * 24, created_at=(datetime.now().astimezone() + timedelta(seconds=1)).isoformat())
             from loopx.chat_store import _atomic_write_json
             _atomic_write_json(transport.core.root / f"{later['request_ref']}.json", later)
+            if fault == "later-source-after-discovery":
+                before_discovery = len(scans)
+                assert discovery(ref)  # Retained hint, never permission to dispatch.
+                assert len(scans) == before_discovery
+                transport.core.context_recovery_observer()(ref)
+                assert len(scans) == before_discovery + 1  # A new pass observes the new source input.
         elif fault == "native-tool":
             post = antigravity_adapter._rpc_post
             def with_tool(*args, **kwargs):