diff --git a/docs/reference/protocols/manager-evidence-and-continuity-v0.md b/docs/reference/protocols/manager-evidence-and-continuity-v0.md index 051287deeb..c9b5042a68 100644 --- a/docs/reference/protocols/manager-evidence-and-continuity-v0.md +++ b/docs/reference/protocols/manager-evidence-and-continuity-v0.md @@ -416,6 +416,26 @@ without reapplying. After a crash with an uncertain external effect, reconcile by provider idempotency/readback before retry; if the provider cannot prove the result, retain outcome-unknown. Do not promise universal exactly-once effects. +For private native Chat, a verified failure notice does not close its original +request. The existing reply worker may reserve a bounded, separately correlated +continuation after a recognized routing discovery failure or timeout. Admission +rechecks the original provider message, binding, grant and idle Session. Native +readback must enumerate every Turn and the failed Turn's complete item history: +all Turns are terminal, and an older failure contains exactly its original user +item. Later completed work alone does not establish execution of that request. +The native compaction exception applies only to the latest failed Turn. Any +answer, tool, correction, unknown item, local response delta or interrupted work +keeps the original failure pending; it cannot be replayed by this path. Existing +backoff, three-attempt limit, reservation recovery and per-attempt delivery +readback remain in force, preserving the failed Turn and its original notice. + +个人原生 Chat 的失败通知不代表原请求完成。既有回复 worker 只能在明确识别的路由 +发现失败或超时、原消息和绑定权限再次核验、会话空闲及原生完整读回后，有界接续原 +请求。历史失败必须只有原始用户输入；之后其他请求完成不能证明它已经执行。仅最新 +失败可包含原生压缩记录；任何回答、工具、纠正、未知项、本地回答增量或中断的工作 +仍保留待核对，不能走此路径重放。退避、三次上限、稳定接续身份、失败历史和逐次 +原受众送达读回均保持原约束。 + Default manager duties allow authorized reads and summaries; reminders use an existing authorized destination/subscription. Goal changes, steering, cross-Goal dispatch, pause/resume, publishing, and trading keep their existing diff --git a/loopx/capabilities/native_chat/codex_adapter.py b/loopx/capabilities/native_chat/codex_adapter.py index f9dd101a64..a2a5124d01 100644 --- a/loopx/capabilities/native_chat/codex_adapter.py +++ b/loopx/capabilities/native_chat/codex_adapter.py @@ -26,9 +26,10 @@ def failed_request_readback(session: CodexChatAgentSession, turn_id: str) -> dic """Observe a historical pre-execution failure without resuming it. Unlike a same-worker retry, a migration may inspect an older failure. - Every native Turn must be terminal. Routing recovery additionally - requires the latest Turn and permits only user input plus native - compaction bookkeeping: no answer, tool, correction or unknown item. + Every native Turn must be terminal. A historical routing failure must + contain only the original user input. Native compaction bookkeeping is + allowed only on the latest Turn: no answer, tool, correction or unknown + item qualifies. Core still rechecks the original source and grant. The Core still owns any new continuation admission. """ deadline = time.monotonic() + session.response_timeout_sec @@ -46,12 +47,11 @@ def terminal() -> dict[str, Any] | None: if not selected or selected.get("status") != "failed": return None if _routing_discovery_timeout(selected.get("error")): - # A later request may supersede this one. Compaction is native - # bookkeeping, but only the exact latest routing failure can - # use that exception to the historical user-only inventory. - if ids[0] != turn_id: - return None - return {"codex_error_info": "other", "pre_execution_routing_timeout": True} + # Later work must not strand an untouched accepted request. Keep + # the historical user-only rule; only the exact latest failure + # can use the compaction exception to that inventory. + return {"codex_error_info": "other", "pre_execution_routing_timeout": True, + "latest_routing_failure": ids[0] == turn_id} info = _retry_error_details(selected.get("error")).get("retry", {}).get("codex_error_info") # `other` is evidence of a failed native request, not permission to # retry it. The typed admission owner separately decides whether @@ -73,7 +73,8 @@ def terminal() -> dict[str, Any] | None: return None items = list(session._control_items("thread/items/list", {"turnId": turn_id}, deadline)) routing = proof.get("pre_execution_routing_timeout") is True - allowed = {"userMessage", "contextCompaction"} if routing else {"userMessage"} + allowed = ({"userMessage", "contextCompaction"} + if routing and proof["latest_routing_failure"] else {"userMessage"}) ids: set[str] = set() users = 0 for entry in items: @@ -86,6 +87,7 @@ def terminal() -> dict[str, Any] | None: users += item["type"] == "userMessage" if users != 1 or (not routing and len(items) != 1) or terminal() != proof: return None + proof.pop("latest_routing_failure", None) return {"thread_id": session.thread_id, "turn_id": turn_id, "only_original_user_item": len(items) == 1, **({"only_pre_execution_items": True} if routing else {}), **proof} diff --git a/loopx/chat_agent.py b/loopx/chat_agent.py index cb29c72465..0668d349e3 100644 --- a/loopx/chat_agent.py +++ b/loopx/chat_agent.py @@ -208,6 +208,8 @@ def _routing_discovery_timeout(error: Any) -> bool: and error.get("message") in ( "workspace routing discovery timed out", "Error running remote compact task: workspace routing discovery timed out", + "workspace routing discovery failed", + "Error running remote compact task: workspace routing discovery failed", )) diff --git a/tests/test_chat_agent.py b/tests/test_chat_agent.py index 8458f7f2c2..47645d9265 100644 --- a/tests/test_chat_agent.py +++ b/tests/test_chat_agent.py @@ -511,8 +511,9 @@ def write(packet): @pytest.mark.parametrize("compact", [False, True]) @pytest.mark.parametrize("reverse", [False, True]) +@pytest.mark.parametrize("diagnostic", ["timed out", "failed"]) @pytest.mark.parametrize("fault", [None, "tool", "answer", "unknown", "correction", "older", "changed", "duplicate"]) -def test_routing_failure_readback_allows_only_native_compaction_before_execution(tmp_path, monkeypatch, compact, reverse, fault): +def test_routing_failure_readback_allows_only_native_compaction_before_execution(tmp_path, monkeypatch, compact, reverse, diagnostic, fault): session = chat_agent.CodexChatAgentSession(process=SimpleNamespace(poll=lambda: None), messages=queue.Queue(), thread_id="routing-thread", work_dir=tmp_path, response_timeout_sec=.2) reads = 0 @@ -521,7 +522,7 @@ def write(packet): nonlocal reads if packet["method"] == "thread/turns/list": reads += 1 - message = ("Error running remote compact task: " if compact else "") + "workspace routing discovery timed out" + message = ("Error running remote compact task: " if compact else "") + f"workspace routing discovery {diagnostic}" if fault == "changed" and reads > 1: message = "different private error" rows = [{"id": "failed", "status": "failed", "error": {"codexErrorInfo": "other", "message": message}}] @@ -546,18 +547,19 @@ def write(packet): monkeypatch.setattr(session, "_write", write) proof = failed_request_readback(session, "failed") - if fault is None: + if fault is None or (fault == "older" and not compact): assert proof == {"thread_id": "routing-thread", "turn_id": "failed", "only_original_user_item": not compact, "only_pre_execution_items": True, "codex_error_info": "other", "pre_execution_routing_timeout": True} error = chat_agent._terminal_turn_error({"codexErrorInfo": "other", "message": - ("Error running remote compact task: " if compact else "") + "workspace routing discovery timed out"}, "fallback") + ("Error running remote compact task: " if compact else "") + f"workspace routing discovery {diagnostic}"}, "fallback") assert error.error_code == "upstream_transport_failure" and error.native_routing_timeout else: assert proof is None -@pytest.mark.parametrize("message", [None, {}, [], "workspace routing discovery timed out: extra", "unknown"]) +@pytest.mark.parametrize("message", [None, {}, [], "workspace routing discovery timed out: extra", + "workspace routing discovery failed: extra", "Please retry: workspace routing discovery failed", "unknown"]) def test_routing_diagnostic_is_an_exact_provider_diagnostic(message): error = {"codexErrorInfo": "other", "message": message} assert not chat_agent._routing_discovery_timeout(error) diff --git a/tests/test_private_chat_routing_recovery.py b/tests/test_private_chat_routing_recovery.py index b29a7ef8b1..c034b1c6dd 100644 --- a/tests/test_private_chat_routing_recovery.py +++ b/tests/test_private_chat_routing_recovery.py @@ -113,3 +113,102 @@ def crash(**kwargs): starts = [json.loads(line) for line in capture.read_text().splitlines() if json.loads(line).get("method") == "turn/start"] assert len(starts) == 3 and len(store.list_sessions()) == 1 + + +@pytest.mark.parametrize("profile", ["notes-app", "steward-app"]) +@pytest.mark.parametrize("fault", [None, "tool", "compaction", "active", "changed-source", "delta", "policy"]) +def test_original_worker_recovers_historical_user_only_routing_failure(steward, monkeypatch, tmp_path, profile, fault): # noqa: F811 + """Read the persisted native inventory over stdio, then use the original queue. + + A later successful request cannot prove an earlier untouched request ran. + Tools, compaction on that older Turn, changed sources and unknown local + effects still prevent recovery. No test substitutes a recovery proof. + """ + store, runtime, provider, transport, _, capture, _ = steward + fake = Path(runtime.codex_bin) + source = fake.read_text().replace('"codexErrorInfo": "cyberPolicy"', '"codexErrorInfo": "other"') + source = source.replace('"private-fixture-upstream-detail"', + '"Error running remote compact task: workspace routing discovery failed"') + source = source.replace(' print(json.dumps({"method": "item/agentMessage/delta", "params": {"threadId": "durable-thread", "turnId": active_turn, "delta": "Partial answer."}}), flush=True)\n', '') + fake.write_text(source) + binding = next(row for row in transport.bindings.read()["bindings"] if row["transport_ref"] == profile) + event = provider.event(profile, "unstarted-request", "typed terminal failure") + transport.admit(profile, event) + ref = identity_ref(binding["provider_ref"], event["message_id"]) + original = transport.core.read_request(ref) + failed = finish(runtime, original) + assert failed["status"] == "failed" and failed["delta_count"] == 0 + assert failed["error_code"] == "upstream_transport_failure" + transport.reconcile() + failed_path = store._turn_path(original["session_id"], original["turn_id"]) + if fault == "delta": + store.update_turn(original["session_id"], original["turn_id"], delta_count=1) + original_bytes = failed_path.read_bytes() + transport_path = transport.root / f"{ref}.json" + old_notice = json.loads(transport_path.read_text())["deliveries"]["terminal"] + later_event = provider.event(profile, "separate-request", "Complete this independent request.") + transport.admit(profile, later_event) + later = transport.core.read_request(identity_ref(binding["provider_ref"], later_event["message_id"])) + assert later["session_id"] == original["session_id"] and finish(runtime, later)["status"] == "completed" + transport.reconcile() + runtime.adapters.pop(original["session_id"]).close_session() + Path(store.load_session(original["session_id"])["codex_home"]).mkdir(parents=True, exist_ok=True) + + # A fresh native stdio process reads a durable provider snapshot. The model + # fixture now succeeds; only the existing transport worker may admit it. + failed_native = failed["upstream_turn_id"] + later_native = store.load_turn(later["session_id"], later["turn_id"])["upstream_turn_id"] + inventory = tmp_path / "native-inventory.json" + error = {"codexErrorInfo": "cyberPolicy" if fault == "policy" else "other", + "message": "Error running remote compact task: workspace routing discovery failed"} + entries = [{"turnId": failed_native, "item": {"type": "userMessage", "id": "original-user"}}] + if fault in {"tool", "compaction"}: + entries.append({"turnId": failed_native, "item": {"type": + "commandExecution" if fault == "tool" else "contextCompaction", "id": "additional-item"}}) + inventory.write_text(json.dumps({"turns": [{"id": later_native, + "status": "inProgress" if fault == "active" else "completed"}, + {"id": failed_native, "status": "failed", "error": error}], "items": entries})) + begin = source.index(' elif method == "thread/turns/list":') + end = source.index(' else:\n result = {}', begin) + reader = f''' elif method in {{"thread/turns/list", "thread/items/list"}}: + with open({str(inventory)!r}) as observed: + snapshot = json.load(observed) + result = {{"data": snapshot["turns" if method == "thread/turns/list" else "items"]}} + elif method == "thread/backgroundTerminals/list": + result = {{"data": []}} +''' + source = source[:begin] + reader + source[end:] + fake.write_text(source.replace('if "typed terminal failure" in json.dumps(request.get("params") or {}):', 'if False:')) + if fault == "changed-source": + provider.messages[event["message_id"]]["content"] = "Changed requirement" + clock = datetime.now(timezone.utc) + timedelta(seconds=61) + class AfterBackoff(datetime): + @classmethod + def now(cls, tz=None): + return clock + monkeypatch.setattr(external_conversations, "datetime", AfterBackoff) + worker = LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent, + runner=provider, cli_bin="lark-cli") + worker.reconcile() + recovered = worker.core.read_request(ref) + if fault is not None: + assert recovered["turn_id"] == original["turn_id"] and not recovered.get("context_recovery") + assert failed_path.read_bytes() == original_bytes + return + assert recovered["turn_id"] != original["turn_id"] + assert recovered["session_id"] == original["session_id"] + assert recovered["context_recovery"]["routing_attempt"] == 1 + assert finish(runtime, recovered)["status"] == "completed" + worker.reconcile() + delivered = json.loads(transport_path.read_text()) + assert delivered["deliveries"]["terminal"] == old_notice + assert delivered["deliveries"][presentation_phase(recovered, "terminal")]["verified"] + assert failed_path.read_bytes() == original_bytes + before = list(provider.writes) + LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent, + runner=provider, cli_bin="lark-cli").reconcile() + assert provider.writes == before + requests = [json.loads(line) for line in capture.read_text().splitlines()] + assert sum(item.get("method") == "turn/start" for item in requests) == 3 + assert any(item.get("method") == "thread/items/list" for item in requests) + assert len(store.list_sessions()) == 1