fix(chat): avoid holding source routes during recovery provider reads

loopx/capabilities/native_chat/external_conversations.py

@@ -21,7 +21,7 @@
     automation_cadence_readback,
     unavailable_automation_cadence_readback,
 )
-from ...file_lock import exclusive_file_lock
+from ...file_lock import exclusive_file_lock, try_exclusive_file_lock
 from .codex_context import checked_session_home
 
 # Request journals are named by their 24-hex request_ref; lock metadata and
@@ -49,6 +49,7 @@ def __init__(self, controller: Any) -> None:
     def _context_recovery_plan(self, row: dict[str, Any], *,
                                observation: dict[str, Any] | None = None,
                                source_verified: bool = False,
+                               selected_binding: dict[str, Any] | None = None,
                                discovery_only: bool = False) -> dict[str, Any]:
         from ...chat_manager import MANAGER_CONTEXT_VERSION
         from ...control_plane.effect_runtime import effect_runtime_result
@@ -58,15 +59,15 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
         turn = store.load_turn(str(row.get("session_id") or ""), str(row.get("turn_id") or ""))
         if not session or not turn:
             return {}
-        if discovery_only:
+        if selected_binding is None and discovery_only:
             # A queue hint must not serialize provider identity IO across all
             # historical failures. Reuse the same typed planner with the local
             # enabled binding; admission independently resolves fresh authority.
             selected_binding = next((item for item in self.bindings.read()["bindings"]
                 if item["binding_id"] == row["binding_id"] and item.get("enabled") is True), None)
             if selected_binding is None:
                 return {}
-        else:
+        elif selected_binding is None:
             selected_binding = self.bindings.resolve(binding_id=row["binding_id"], **row["source"])["binding"]
         reservation = row.get("context_recovery") or {}
         admitted = (store.turn_for_client(session["session_id"], reservation["client_turn_id"])
@@ -156,17 +157,31 @@ def source_requests() -> list[dict[str, Any]]:
                 **({"read_only_review": True} if plan.get("read_only_review") else {}),
                 **({"completed_work": True, "work_dir": Path(context["project"])} if plan.get("work_attempt") else {}),
                 thread_ids=plan["native_thread_ids"], turn_id=turn["upstream_turn_id"])
-            with exclusive_file_lock(self.root / "source-fences" / f"{row['binding_id']}.{row['source']['source_ref']}.json", operation="route_external_chat_request"):
-                selected = self.bindings.resolve(binding_id=row["binding_id"], **row["source"], include_observation=True)
-                source = source_observation(selected)
+            selected = self.bindings.resolve(binding_id=row["binding_id"], **row["source"], include_observation=True)
+            source = source_observation(selected)
+            # Identity and original-message reads are fallible provider IO too.
+            # Never hold up newer source input while they wait. If another
+            # route already owns the fence, leave this request unchanged for
+            # the existing worker's next fresh visit, rather than admitting
+            # from an observation that waited behind that route.
+            with try_exclusive_file_lock(self.root / "source-fences" / f"{row['binding_id']}.{row['source']['source_ref']}.json", operation="route_external_chat_request") as acquired:
+                if acquired is None:
+                    return row
+                current = self.bindings.read()
+                if next((item for item in current.get("bindings", [])
+                         if item.get("binding_id") == row["binding_id"]), None) != selected["binding"]:
+                    raise ValueError("conversation binding changed during recovery source verification")
+                selected = self.bindings._resolve_observed(current, selected["binding"],
+                    selected["observation"], row["source"])
                 # Adopting an already admitted Turn only repairs correlation:
                 # its execution and newer input may legitimately have advanced.
                 # The typed planner below must still prove that exact receipt.
                 if not plan.get("admitted_turn_id") and (self.controller.store.load_session(row["session_id"]) != session
                         or self.controller.store.load_turn(row["session_id"], row["turn_id"]) != turn
                         or source_requests() != requests):
                     return row
-                plan = self._context_recovery_plan(row, observation=proof, source_verified=source is not None)
+                plan = self._context_recovery_plan(row, observation=proof, source_verified=source is not None,
+                    selected_binding=selected["binding"])
                 if not plan.get("recovery_eligible"):
                     return row
                 recovery = row.get("context_recovery")

tests/test_native_steward_private.py

@@ -74,6 +74,82 @@ def finish(runtime, row):
     return runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=15)
 
 
+@pytest.mark.parametrize("change", ["status", "correction", "revoked", "source-changed", "unknown", "unchanged", "route-busy"])
+def test_recovery_source_read_does_not_block_new_input_or_reuse_waited_observations(steward, monkeypatch, change):
+    import threading
+    from concurrent.futures import ThreadPoolExecutor, TimeoutError
+    from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter
+    from loopx.file_lock import exclusive_file_lock
+
+    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", "source-read-recovery", "typed terminal failure")
+    transport.admit("steward-app", event)
+    ref = identity_ref(binding["provider_ref"], event["message_id"])
+    row = transport.core.read_request(ref)
+    failed = finish(runtime, row)
+    store.update_session(row["session_id"], manager_context_version=20)
+    assert transport.reconcile() == 1
+    original = store._turn_path(row["session_id"], row["turn_id"]).read_bytes()
+    before = capture.read_bytes()
+    record = json.loads((transport.root / f"{ref}.json").read_text())
+    monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(lambda **_: None
+        if change == "unknown" else {"thread_id": "durable-thread", "turn_id": failed["upstream_turn_id"],
+            "only_original_user_item": True, "codex_error_info": "contextWindowExceeded"}))
+    started, release = threading.Event(), threading.Event()
+
+    def source_read(selected):
+        started.set()
+        assert release.wait(10)
+        return transport._recovery_source_observation(record, selected)
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        recovery = pool.submit(transport.core.recover_failed_request, ref, source_observation=source_read)
+        responsive = True
+        try:
+            assert started.wait(5)
+            if change == "status":
+                peer = pool.submit(transport.admit, "steward-app", provider.event("steward-app", "fresh-status", "/status"))
+                try:
+                    assert peer.result(timeout=1)["status"] == "command_recorded"
+                except TimeoutError:
+                    responsive = False
+            elif change == "correction":
+                store.create_queued_turn(row["session_id"], client_turn_id="source-read-correction",
+                    message="Inspect this new correction first", origin="web")
+            elif change == "revoked":
+                transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"])
+            elif change == "source-changed":
+                provider.messages[event["message_id"]]["content"] = "Revised original request"
+            elif change == "route-busy":
+                fence = transport.core.root / "source-fences" / f"{row['binding_id']}.{row['source']['source_ref']}.json"
+                with exclusive_file_lock(fence, operation="other_original_input"):
+                    release.set()
+                    assert recovery.result(timeout=1)["turn_id"] == row["turn_id"]
+        finally:
+            release.set()
+        if change == "revoked":
+            with pytest.raises(ValueError, match="binding|authorized"):
+                recovery.result(timeout=10)
+        else:
+            continued = recovery.result(timeout=10)
+            assert (continued["turn_id"] != row["turn_id"]) is (change == "unchanged")
+        assert responsive, "A provider source read must leave new original input admissible"
+    assert store._turn_path(row["session_id"], row["turn_id"]).read_bytes() == original
+    if change != "unchanged":
+        assert capture.read_bytes() == before, "Changed, unknown or busy state cannot start model work"
+    if change == "route-busy":
+        # The existing worker's next visit gets fresh source proof. No new
+        # scheduler or retry queue is needed, and no old observation is reused.
+        continued = transport.core.recover_failed_request(ref,
+            source_observation=lambda selected: transport._recovery_source_observation(record, selected))
+        assert continued["turn_id"] != row["turn_id"]
+        finish(runtime, continued)
+
+
 @pytest.mark.parametrize("case", ["enabled", "disabled", "missing", "authority-unavailable"])
 def test_recovery_discovery_never_observes_provider_but_admission_still_does(steward, monkeypatch, case):
     store, runtime, provider, transport, binding, capture, _ = steward