From 407a7d6c4916f3ce033727e287f03188c956ce1f Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 28 Oct 2026 00:24:44 +0800 Subject: [PATCH 1/4] fix(chat): release source route during native recovery reads Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../native_chat/external_conversations.py | 41 ++++++++---- tests/test_chat_conversation_bindings.py | 67 +++++++++++++++++++ 2 files changed, 95 insertions(+), 13 deletions(-) diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 2e32283b30..5f6aeb72a0 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -131,21 +131,36 @@ def recover_failed_request(self, request_ref: str, *, path = self.root / f"{request_ref}.json" with exclusive_file_lock(path, operation="admit_external_chat_request"): row = self.read_request(request_ref) + plan = self._context_recovery_plan(row) + if not plan.get("recovery_candidate"): + return row + session = self.controller.store.load_session(row["session_id"]) + turn = self.controller.store.load_turn(row["session_id"], row["turn_id"]) + assert session is not None and turn is not None + model_auth_home = checked_session_home(self.controller.codex_home, session) + context = self.controller.project_contexts.session_context(session) + + def source_requests() -> list[dict[str, Any]]: + return [item for item in self.pending() if item.get("binding_id") == row["binding_id"] + and item.get("source") == row["source"]] + + requests = source_requests() + # The read-only native client can consume its entire RPC deadline. + # Retain this request's fence, but let new source input/control in. + # Its observation is only usable if the original local state still + # matches after reacquiring the route fence; fresh authority/source + # checks and the typed eligibility decision remain below that fence. + proof = None if plan.get("admitted_turn_id") else CodexAppServerAdapter.failed_request_observation( + codex_bin=self.controller.codex_bin, codex_home=Path(session["codex_home"]), + model_auth_home=model_auth_home, + **({"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"): - plan = self._context_recovery_plan(row) - if not plan.get("recovery_candidate"): + if (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 - session = self.controller.store.load_session(row["session_id"]) - turn = self.controller.store.load_turn(row["session_id"], row["turn_id"]) - assert session is not None and turn is not None - model_auth_home = checked_session_home(self.controller.codex_home, session) - context = self.controller.project_contexts.session_context(session) - proof = None if plan.get("admitted_turn_id") else CodexAppServerAdapter.failed_request_observation( - codex_bin=self.controller.codex_bin, codex_home=Path(session["codex_home"]), - model_auth_home=model_auth_home, - **({"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"]) selected = self.bindings.resolve(binding_id=row["binding_id"], **row["source"], include_observation=True) source = source_observation(selected) plan = self._context_recovery_plan(row, observation=proof, source_verified=source is not None) diff --git a/tests/test_chat_conversation_bindings.py b/tests/test_chat_conversation_bindings.py index ad95c75ad0..65edf1e951 100644 --- a/tests/test_chat_conversation_bindings.py +++ b/tests/test_chat_conversation_bindings.py @@ -17,6 +17,73 @@ def _history_page_brief(): "inputs": [], "acceptance": ["Return the original conclusion"], "return_requirement": "Return here"} +@pytest.mark.parametrize("change", ["status", "new-input", "revoked", "unknown-native", "unchanged"]) +def test_slow_failed_request_probe_does_not_hold_the_source_route(steward, monkeypatch, change): # noqa: F811 + import threading + from concurrent.futures import ThreadPoolExecutor + from pathlib import Path + from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter + + 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", "slow-native-read", "typed terminal failure") + transport.admit("steward-app", event) + ref = identity_ref(binding["provider_ref"], event["message_id"]) + row = transport.core.read_request(ref) + failed = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=10) + assert failed["status"] == "failed" and failed["delta_count"] == 0 + 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() + started, release = threading.Event(), threading.Event() + + def slow_probe(**_): + started.set() + assert release.wait(10) + # Even a valid read cannot override input or revocation during IO. + return None if change == "unknown-native" else {"thread_id": "durable-thread", + "turn_id": failed["upstream_turn_id"], "only_original_user_item": True, + "codex_error_info": "contextWindowExceeded"} + + monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(slow_probe)) + with ThreadPoolExecutor(max_workers=1) as pool: + recovery = pool.submit(transport.core.recover_failed_request, ref, + source_observation=lambda _: {"verified": True}) + try: + assert started.wait(5) + if change in {"status", "unknown-native"}: + begin = time.monotonic() + admitted = transport.admit("steward-app", provider.event("steward-app", "during-native-read", "/status")) + assert admitted["status"] == "command_recorded" + assert time.monotonic() - begin < 1, "Native readback must not block fresh source admission" + elif change == "new-input": + store.create_queued_turn(row["session_id"], client_turn_id="later-correction", + message="A later original correction", origin="lark") + elif change == "revoked": + transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"]) + finally: + release.set() + if change == "revoked": + with pytest.raises(ValueError): + recovery.result(timeout=10) + elif change == "unchanged": + continued = recovery.result(timeout=10) + assert continued["turn_id"] != row["turn_id"] + assert continued["session_id"] == row["session_id"] + runtime.wait_for_turn(session_id=row["session_id"], turn_id=continued["turn_id"], timeout_sec=10) + assert transport.core.recover_failed_request(ref, source_observation=lambda _: pytest.fail("duplicate recovery"))["turn_id"] == continued["turn_id"] + else: + assert recovery.result(timeout=10)["turn_id"] == row["turn_id"] + assert bool(transport.core.read_request(ref).get("context_recovery")) is (change == "unchanged") + assert store._turn_path(row["session_id"], row["turn_id"]).read_bytes() == original + if change != "unchanged": + assert capture.read_bytes() == before, "Unknown or changed state cannot create or resume model work" + + def test_history_page_keeps_one_original_collaboration_carrier(flow, monkeypatch): # noqa: F811 from loopx.capabilities.manager_context.roundtrip import project_chat_session_snapshot From 41229956e61f83e479aafe9f9db0693636a26c95 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 28 Oct 2026 00:31:45 +0800 Subject: [PATCH 2/4] fix(chat): recheck recovery state after source verification Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../native_chat/external_conversations.py | 4 ++-- tests/test_chat_conversation_bindings.py | 11 +++++++++-- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 5f6aeb72a0..5a50c0ab63 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -157,12 +157,12 @@ def source_requests() -> list[dict[str, Any]]: **({"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) if (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 - selected = self.bindings.resolve(binding_id=row["binding_id"], **row["source"], include_observation=True) - source = source_observation(selected) plan = self._context_recovery_plan(row, observation=proof, source_verified=source is not None) if not plan.get("recovery_eligible"): return row diff --git a/tests/test_chat_conversation_bindings.py b/tests/test_chat_conversation_bindings.py index 65edf1e951..97300e5dde 100644 --- a/tests/test_chat_conversation_bindings.py +++ b/tests/test_chat_conversation_bindings.py @@ -17,7 +17,7 @@ def _history_page_brief(): "inputs": [], "acceptance": ["Return the original conclusion"], "return_requirement": "Return here"} -@pytest.mark.parametrize("change", ["status", "new-input", "revoked", "unknown-native", "unchanged"]) +@pytest.mark.parametrize("change", ["status", "new-input", "source-read-input", "revoked", "unknown-native", "unchanged"]) def test_slow_failed_request_probe_does_not_hold_the_source_route(steward, monkeypatch, change): # noqa: F811 import threading from concurrent.futures import ThreadPoolExecutor @@ -50,9 +50,16 @@ def slow_probe(**_): "codex_error_info": "contextWindowExceeded"} monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(slow_probe)) + + def source_observation(_): + if change == "source-read-input": + store.create_queued_turn(row["session_id"], client_turn_id="input-during-source-read", + message="A correction while reading the provider", origin="web") + return {"verified": True} + with ThreadPoolExecutor(max_workers=1) as pool: recovery = pool.submit(transport.core.recover_failed_request, ref, - source_observation=lambda _: {"verified": True}) + source_observation=source_observation) try: assert started.wait(5) if change in {"status", "unknown-native"}: From a89b8e19e82d5579b1f8b60eb8f769c56a1cdb82 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 28 Oct 2026 00:36:17 +0800 Subject: [PATCH 3/4] test(chat): keep independent sources out of recovery fences Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/test_chat_conversation_bindings.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/tests/test_chat_conversation_bindings.py b/tests/test_chat_conversation_bindings.py index 97300e5dde..0bed68905a 100644 --- a/tests/test_chat_conversation_bindings.py +++ b/tests/test_chat_conversation_bindings.py @@ -17,7 +17,7 @@ def _history_page_brief(): "inputs": [], "acceptance": ["Return the original conclusion"], "return_requirement": "Return here"} -@pytest.mark.parametrize("change", ["status", "new-input", "source-read-input", "revoked", "unknown-native", "unchanged"]) +@pytest.mark.parametrize("change", ["status", "new-input", "source-read-input", "unrelated-source", "revoked", "unknown-native", "unchanged"]) def test_slow_failed_request_probe_does_not_hold_the_source_route(steward, monkeypatch, change): # noqa: F811 import threading from concurrent.futures import ThreadPoolExecutor @@ -70,6 +70,8 @@ def source_observation(_): elif change == "new-input": store.create_queued_turn(row["session_id"], client_turn_id="later-correction", message="A later original correction", origin="lark") + elif change == "unrelated-source": + assert transport.admit("notes-app", provider.event("notes-app", "other-original-source", "/status"))["status"] == "command_recorded" elif change == "revoked": transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"]) finally: @@ -77,7 +79,7 @@ def source_observation(_): if change == "revoked": with pytest.raises(ValueError): recovery.result(timeout=10) - elif change == "unchanged": + elif change in {"unchanged", "unrelated-source"}: continued = recovery.result(timeout=10) assert continued["turn_id"] != row["turn_id"] assert continued["session_id"] == row["session_id"] @@ -85,9 +87,9 @@ def source_observation(_): assert transport.core.recover_failed_request(ref, source_observation=lambda _: pytest.fail("duplicate recovery"))["turn_id"] == continued["turn_id"] else: assert recovery.result(timeout=10)["turn_id"] == row["turn_id"] - assert bool(transport.core.read_request(ref).get("context_recovery")) is (change == "unchanged") + assert bool(transport.core.read_request(ref).get("context_recovery")) is (change in {"unchanged", "unrelated-source"}) assert store._turn_path(row["session_id"], row["turn_id"]).read_bytes() == original - if change != "unchanged": + if change not in {"unchanged", "unrelated-source"}: assert capture.read_bytes() == before, "Unknown or changed state cannot create or resume model work" From 2d8f2f58b6ddb766711e0bc53eb57016a3f30153 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 28 Oct 2026 00:39:35 +0800 Subject: [PATCH 4/4] fix(chat): retain exact admitted recovery correlation Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/capabilities/native_chat/external_conversations.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 5a50c0ab63..9031f23788 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -159,7 +159,10 @@ def source_requests() -> list[dict[str, Any]]: 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) - if (self.controller.store.load_session(row["session_id"]) != session + # 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