diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 2e32283b30..9031f23788 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -131,23 +131,41 @@ 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"): - 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) + # 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) 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 ad95c75ad0..0bed68905a 100644 --- a/tests/test_chat_conversation_bindings.py +++ b/tests/test_chat_conversation_bindings.py @@ -17,6 +17,82 @@ def _history_page_brief(): "inputs": [], "acceptance": ["Return the original conclusion"], "return_requirement": "Return here"} +@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 + 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)) + + 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=source_observation) + 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 == "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: + release.set() + if change == "revoked": + with pytest.raises(ValueError): + recovery.result(timeout=10) + 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"] + 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 in {"unchanged", "unrelated-source"}) + assert store._turn_path(row["session_id"], row["turn_id"]).read_bytes() == original + if change not in {"unchanged", "unrelated-source"}: + 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