fix(chat): recover untouched historical routing failures

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

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}

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",
             ))
 
 

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)

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