From cc70b352a4c2777c5535a6a503d8f97a12d4fcc8 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 28 Oct 2026 04:04:37 +0800 Subject: [PATCH] fix(chat): keep provider recovery reads outside source route fence Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../native_chat/external_conversations.py | 29 +++++-- tests/test_native_steward_private.py | 76 +++++++++++++++++++ 2 files changed, 98 insertions(+), 7 deletions(-) diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 9031f23788..804b328d1e 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/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,7 +59,7 @@ 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. @@ -66,7 +67,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *, 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,9 +157,22 @@ 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. @@ -166,7 +180,8 @@ def source_requests() -> list[dict[str, Any]]: 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") diff --git a/tests/test_native_steward_private.py b/tests/test_native_steward_private.py index 2bcbe6ba03..7ba6791a45 100644 --- a/tests/test_native_steward_private.py +++ b/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