From 5f21f0fcedf615a72e8da254758741ed40c4107b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 21 Oct 2026 10:55:06 +0800 Subject: [PATCH 1/5] fix(chat): resume proven routing failures through original workers Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../capabilities/native_chat/codex_adapter.py | 78 ++++++++++++- .../native_chat/external_conversations.py | 16 ++- loopx/chat_agent.py | 68 +++-------- .../collaboration/conversation_binding.ts | 52 ++++++++- .../extensions/lark/private_conversations.py | 8 +- loopx/extensions/lark/private_progress.py | 4 +- .../conversation_binding.test.ts | 48 ++++++++ tests/test_chat_agent.py | 56 ++++++++- tests/test_private_chat_routing_recovery.py | 109 ++++++++++++++++++ 9 files changed, 371 insertions(+), 68 deletions(-) create mode 100644 tests/test_private_chat_routing_recovery.py diff --git a/loopx/capabilities/native_chat/codex_adapter.py b/loopx/capabilities/native_chat/codex_adapter.py index adf37db24f..7d589406a4 100644 --- a/loopx/capabilities/native_chat/codex_adapter.py +++ b/loopx/capabilities/native_chat/codex_adapter.py @@ -15,10 +15,84 @@ import time from typing import Any, Callable -from ...chat_agent import CodexChatAgentSession, CodexChatTimeoutError +from ...chat_agent import ( + CodexChatAgentError, CodexChatAgentSession, CodexChatTimeoutError, + _retry_error_details, _routing_discovery_timeout, +) from ...chat_codex_goal import CodexGoalDriver +def failed_request_readback(session: CodexChatAgentSession, turn_id: str) -> dict[str, Any] | None: + """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. + The Core still owns any new continuation admission. + """ + deadline = time.monotonic() + session.response_timeout_sec + + def terminal() -> dict[str, Any] | None: + turns = list(session._control_items("thread/turns/list", { + "sortDirection": "desc", "itemsView": "summary", + }, deadline)) + ids = [item.get("id") for item in turns] + if (not ids or any(not isinstance(value, str) or not value for value in ids) + or len(set(ids)) != len(ids) + or any(item.get("status") not in {"completed", "failed", "interrupted"} for item in turns)): + return None + selected = next((item for item in turns if item["id"] == turn_id), 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} + 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 + # this proven user-only failure is eligible for one continuation. + if info in ("contextWindowExceeded", "serverOverloaded", "internalServerError", "other"): + return {"codex_error_info": info} + if isinstance(info, dict) and len(info) == 1: + detail = next(iter(info.values())) + status = detail.get("httpStatusCode") if isinstance(detail, dict) else False + if isinstance(detail, dict) and "httpStatusCode" in detail and ( + status is None or type(status) is int and 500 <= status <= 599 + ): + return {"codex_error_info": info} + return None + + try: + proof = terminal() + if proof is 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"} + ids: set[str] = set() + users = 0 + for entry in items: + item = entry.get("item") + if (entry.get("turnId") != turn_id or not isinstance(item, dict) + or item.get("type") not in allowed or not item.get("id") + or not isinstance(item["id"], str) or item["id"] in ids): + return None + ids.add(item["id"]) + users += item["type"] == "userMessage" + if users != 1 or (not routing and len(items) != 1) or terminal() != proof: + return 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} + except (CodexChatAgentError, OSError, TypeError, ValueError): + return None + + @dataclass class CodexAppServerAdapter: session: CodexChatAgentSession @@ -50,7 +124,7 @@ def failed_request_observation(*, codex_bin: str, codex_home: Path, for thread_id in dict.fromkeys(thread_ids): native.thread_id = thread_id native.response_timeout_sec = max(.01, deadline - time.monotonic()) - proof = native.failed_request_readback(turn_id) + proof = failed_request_readback(native, turn_id) if proof is not None: return proof if time.monotonic() >= deadline: diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 23720e068c..ceed6010e4 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -70,6 +70,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *, "request": {key: value for key, value in row.items() if key != "attachments"}, "current_session": session, "binding": selected_binding, "context_recovery": {"version": MANAGER_CONTEXT_VERSION, "failed_turn": turn, + "observed_at": datetime.now(timezone.utc).isoformat(), "steering_observed": any(event.get("kind") == "turn.steered" for event in store.events_after(session["session_id"], turn["turn_id"], None)), "native_observation": observation, "source_verified": source_verified}, @@ -83,8 +84,9 @@ def context_recovery_candidate(self, request_ref: str) -> bool: if row.get("status") != "accepted": return False recovery = row.get("context_recovery") - if recovery and row.get("turn_id") != recovery["failed_turn_id"]: - return not row.get("delivery_verified") + if (recovery and row.get("turn_id") != recovery["failed_turn_id"] + and not row.get("delivery_verified")): + return True if not row.get("delivery_verified"): return False turn = self.controller.store.load_turn(row["session_id"], row["turn_id"]) @@ -127,11 +129,17 @@ def recover_failed_request(self, request_ref: str, *, if not plan.get("recovery_eligible"): return row recovery = row.get("context_recovery") - if recovery is None: + if recovery is None or recovery.get("client_turn_id") != plan["client_turn_id"]: + if recovery is not None: + row.setdefault("context_recovery_history", []).append(deepcopy(recovery)) recovery = {"version": MANAGER_CONTEXT_VERSION, "failed_turn_id": row["turn_id"], "native_observation": proof, "source_observation": source, "client_turn_id": plan["client_turn_id"], "failure_notice_delivery_verified": row["delivery_verified"]} + if plan.get("routing_attempt") is not None: + recovery.update(routing_attempt=plan["routing_attempt"], + presentation_key=f"{MANAGER_CONTEXT_VERSION}.routing.{plan['routing_attempt']}", + not_before=plan["not_before"]) row["context_recovery"] = recovery _atomic_write_json(path, row) # Freeze correlation before the existing queue can dispatch. @@ -147,7 +155,7 @@ def recover_failed_request(self, request_ref: str, *, return row def recovery_context(self, session: dict[str, Any], turn: dict[str, Any]) -> dict[str, Any] | None: - match = re.fullmatch(r"external-([a-f0-9]{24})-recovery-[1-9][0-9]*", str(turn.get("client_turn_id") or "")) + match = re.fullmatch(r"external-([a-f0-9]{24})-recovery-[1-9][0-9]*(?:-routing-[1-3])?", str(turn.get("client_turn_id") or "")) if match is None: return None row = self.read_request(match[1]) diff --git a/loopx/chat_agent.py b/loopx/chat_agent.py index ff5d45109f..454ae40e7e 100644 --- a/loopx/chat_agent.py +++ b/loopx/chat_agent.py @@ -205,12 +205,26 @@ def _retry_error_details(error: Any) -> dict[str, Any]: def _routing_discovery_timeout(error: Any) -> bool: # A fixed native provider diagnostic, never a substring/prose retry grant. return (isinstance(error, dict) and error.get("codexErrorInfo") == "other" - and error.get("message") == "workspace routing discovery timed out") + and error.get("message") in ( + "workspace routing discovery timed out", + "Error running remote compact task: workspace routing discovery timed out", + )) def _terminal_turn_error(error: Any, fallback: str) -> CodexChatAgentError: """Project only the app-server's typed error, never its arbitrary prose.""" info = error.get("codexErrorInfo") if isinstance(error, dict) else None + if _routing_discovery_timeout(error): + failure = CodexChatAgentError( + "Codex 上游工作区路由暂时不可用,本轮未完成。", + error_code="upstream_transport_failure", + gate=_host_tool_gate( + "Codex 上游工作区路由暂时不可用,本轮未完成。", + "保留原请求;仅在独立核验原生执行项、原消息和授权后允许有界接续。", + ), + ) + failure.native_routing_timeout = True + return failure # Some Responses providers return a structured error without an HTTP # status. The typed error still identifies a rejected request; arbitrary # nested prose may contain private request details and is never projected. @@ -1543,58 +1557,6 @@ def _control_items(self, method: str, params: dict[str, Any], deadline: float) - raise self._runtime_error("Codex returned an invalid control inventory cursor.") seen_cursors.add(cursor) - def failed_request_readback(self, turn_id: str) -> dict[str, Any] | None: - """Observe a historical pre-execution failure without resuming it. - - Unlike a same-worker transport retry, this may inspect an older failed - request. Every native Turn must be terminal, and the selected request - must contain exactly its user input: no answer, tool, correction or - unknown item. The Core still owns any new continuation admission. - """ - deadline = time.monotonic() + self.response_timeout_sec - - def terminal() -> dict[str, Any] | None: - turns = list(self._control_items("thread/turns/list", { - "sortDirection": "desc", "itemsView": "summary", - }, deadline)) - ids = [item.get("id") for item in turns] - if (not ids or any(not isinstance(value, str) or not value for value in ids) - or len(set(ids)) != len(ids) - or any(item.get("status") not in {"completed", "failed", "interrupted"} for item in turns)): - return None - selected = next((item for item in turns if item["id"] == turn_id), None) - if not selected or selected.get("status") != "failed": - return None - 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 - # this proven user-only failure is eligible for one continuation. - if info in ("contextWindowExceeded", "serverOverloaded", "internalServerError", "other"): - return {"codex_error_info": info} - if isinstance(info, dict) and len(info) == 1: - detail = next(iter(info.values())) - status = detail.get("httpStatusCode") if isinstance(detail, dict) else False - if isinstance(detail, dict) and "httpStatusCode" in detail and ( - status is None or type(status) is int and 500 <= status <= 599 - ): - return {"codex_error_info": info} - return None - - try: - proof = terminal() - if proof is None: - return None - items = list(self._control_items("thread/items/list", {"turnId": turn_id}, deadline)) - if (len(items) != 1 or items[0].get("turnId") != turn_id - or not isinstance(items[0].get("item"), dict) - or items[0]["item"].get("type") != "userMessage" - or not items[0]["item"].get("id") or terminal() != proof): - return None - return {"thread_id": self.thread_id, "turn_id": turn_id, - "only_original_user_item": True, **proof} - except (CodexChatAgentError, OSError, TypeError, ValueError): - return None - def transport_retry_readback(self, turn_id: str) -> dict[str, Any] | None: """Read native terminal/effect facts; notification history is insufficient. diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index 902e606fbc..69f6a02aab 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -477,14 +477,18 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO || current.upstream_mode !== "chat" || current.status === "closed" || current.session_mode === "attached_host") return no; const selected = binding(params.binding); - if (selected.context_kind !== "steward" || selected.audience === "group") return no; - const saved = normalizeStewardContext(current.steward_context); + if (selected.audience === "group") return no; + const steward = selected.context_kind === "steward"; + if (selected.executor_endpoint_id !== "codex" || current.goal_id !== (steward ? "loopx-manager" : null)) return no; + const saved = steward ? normalizeStewardContext(current.steward_context) + : normalizeProjectContext(current.project_context); const source = requireJsonObject(row.source, "original conversation source"); if (row.binding_id !== selected.binding_id || saved.binding_id !== selected.binding_id || saved.source_ref !== source.source_ref || saved.provider_ref !== selected.provider_ref || saved.operator_ref !== selected.operator_ref || source.sender_ref !== selected.operator_ref || source.private_human_message !== true - || current.channel_id !== `manager.external.native.${selected.binding_id}.${saved.source_ref}`) return no; + || saved.project_ref !== selected.project_ref || (!steward && saved.grant !== selected.grant) + || current.channel_id !== `${steward ? "manager.external.native" : "project.external"}.${selected.binding_id}.${saved.source_ref}`) return no; const failed = requireJsonObject(recovery.failed_turn, "original failed Turn"); if (failed.session_id !== current.session_id || failed.turn_id !== row.turn_id || failed.status !== "failed" || failed.delta_count !== 0 || failed.response !== null @@ -492,6 +496,9 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO || !["context_window_exceeded", "host_gate", "upstream_transport_failure", "server_overloaded"].includes(String(failed.error_code)) || typeof failed.upstream_turn_id !== "string" || !failed.upstream_turn_id) return no; const previous = row.context_recovery == null ? null : requireJsonObject(row.context_recovery, "reserved context recovery"); + const routing = planRoutingRecovery(recovery, row, current, failed, previous, version); + if (routing !== null) return routing; + if (!steward) return no; const client = `external-${ref(row.request_ref, "original request")}-recovery-${version}`; if (previous && (previous.version !== version || previous.failed_turn_id !== row.turn_id || previous.client_turn_id !== client)) return no; @@ -528,6 +535,45 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO return {...candidate, recovery_eligible: transient}; } +/** Routing outages outlive one worker or context migration. Reserve at most + * three separately correlated continuations, with a persisted failure clock. + * Provider IO must prove a latest, terminal, pre-execution request every time. + */ +function planRoutingRecovery(recovery: JsonObject, row: JsonObject, current: JsonObject, + failed: JsonObject, previous: JsonObject | null, version: number): JsonObject | null { + if (!["host_gate", "upstream_transport_failure"].includes(String(failed.error_code))) return null; + // An unfinished older reservation must finish under its original identity. + if (previous?.failed_turn_id === failed.turn_id && previous.routing_attempt === undefined) return null; + const observed = recovery.native_observation == null ? null + : requireJsonObject(recovery.native_observation, "native failure readback"); + if (observed && observed.pre_execution_routing_timeout !== true) return null; + const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false}; + const reserved = previous?.failed_turn_id === failed.turn_id; + const priorAttempt = previous?.routing_attempt ?? 0; + if (!Number.isSafeInteger(priorAttempt) || Number(priorAttempt) < 0) return no; + const attempt = Number(priorAttempt) + (reserved ? 0 : 1); + if (attempt < 1 || attempt > 3 || typeof current.upstream_thread_id !== "string" + || !current.upstream_thread_id || current.active_turn_id != null || current.status !== "ready") return no; + const now = typeof recovery.observed_at === "string" ? Date.parse(recovery.observed_at) : NaN; + const completed = typeof failed.completed_at === "string" ? Date.parse(failed.completed_at) : NaN; + const notBefore = completed + 60_000 * 2 ** (attempt - 1); + if (!Number.isFinite(now) || !Number.isFinite(completed) || now < notBefore) { + // Preserve the existing context-migration path when no routing diagnostic + // has been established. A proven routing timeout always obeys backoff. + return observed ? no : null; + } + const client = `external-${ref(row.request_ref, "original request")}-recovery-${version}-routing-${attempt}`; + if (reserved && (previous?.client_turn_id !== client || previous.version !== version)) return no; + const candidate = {...no, recovery_candidate: true, client_turn_id: client, + routing_attempt: attempt, not_before: new Date(notBefore).toISOString(), + session_id: current.session_id, failed_turn_id: failed.turn_id, + native_thread_ids: [sessionIdentity(current.upstream_thread_id, "current native thread")]}; + if (!observed) return candidate; + return {...candidate, recovery_eligible: recovery.source_verified === true + && observed.thread_id === current.upstream_thread_id && observed.turn_id === failed.upstream_turn_id + && observed.codex_error_info === "other" && observed.only_pre_execution_items === true}; +} + /** A labelled observation of the same authorized context and canonical queue. * It neither creates a Session nor certifies execution or result delivery. * The external request persists this snapshot so duplicate delivery cannot diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index fa19280665..03bc0a7d5c 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -1107,10 +1107,12 @@ def reconcile_request(self, path: Path) -> int: # and the shared Session queue/fence; transport workers never run models. row = _read_json(path) if row["status"] == "delivered" and row.get("context_recovery"): - # Repair presentation through the original worker, without asking - # Core to recover/enqueue any execution or creating another reply. + # Repair historical presentation first. A delivered failure of a + # continuation still needs Core's fresh recovery judgment; only a + # completed/ineligible request can stop at presentation repair. self._repair_recovery_presentation(path) - return 0 + if not self.core.context_recovery_candidate(row["request_ref"]): + return 0 if row["status"] == "delivered" and row.get("session_id") and row.get("turn_id"): # Legacy terminal delivery may be only a verified failure notice. # Core decides whether a migration continuation is safe; preserve diff --git a/loopx/extensions/lark/private_progress.py b/loopx/extensions/lark/private_progress.py index 7482e834d8..bd164a0d6e 100644 --- a/loopx/extensions/lark/private_progress.py +++ b/loopx/extensions/lark/private_progress.py @@ -18,14 +18,14 @@ def presentation_phase(record: Mapping[str, Any], phase: str) -> str: """A continuation has its own presentation, alongside the old failure.""" recovery = record.get("context_recovery") or {} if recovery and phase in {"admission", "progress", "terminal", "commission_result"}: - return f"{phase}.recovery.{recovery['version']}" + return f"{phase}.recovery.{recovery.get('presentation_key', recovery['version'])}" return phase def presentation_stream(record: dict[str, Any]) -> dict[str, Any]: recovery = record.get("context_recovery") or {} if recovery: - return record.setdefault("recovery_streams", {}).setdefault(str(recovery["version"]), {}) + return record.setdefault("recovery_streams", {}).setdefault(str(recovery.get("presentation_key", recovery["version"])), {}) return record.setdefault("stream", {}) diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index 12c40b3eb2..e9c68cd29b 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/tests/control_plane_ts/conversation_binding.test.ts @@ -17,6 +17,54 @@ const current = {schema_version: "loopx_chat_conversation_bindings_v0", revision const request = {current, expected_revision: 0, operation: "configure", binding: row, observation, available_projects: [project]}; +for (const kind of ["project", "steward"]) { + test(`${kind} routing recovery is bounded, source-bound and independent of context migrations`, () => { + const selected = kind === "steward" ? {...row, context_kind: kind, grant: "portfolio_read", goal_ids: []} : row; + const configured = planConversationBinding({...request, binding: selected}).state; + const source = {source_ref: "e".repeat(24), sender_ref: row.operator_ref, private_human_message: true}; + const context = resolveBoundConversation({current: configured, binding_id: row.binding_id, + observation, available_projects: [project], ...source}).context; + const session = {session_id: "original", goal_id: kind === "steward" ? "loopx-manager" : null, + agent_id: "codex", upstream_mode: "chat", status: "ready", + upstream_thread_id: "native", active_turn_id: null, manager_context_version: 21, + channel_id: `${kind === "steward" ? "manager.external.native" : "project.external"}.${row.binding_id}.${source.source_ref}`, + ...(kind === "steward" ? {steward_context: context} : {project_context: context})}; + const failed = {session_id: "original", turn_id: "failed", status: "failed", delta_count: 0, + response: null, error_code: "host_gate", upstream_turn_id: "native-failed", completed_at: "2026-01-01T00:00:00Z"}; + const observed = {thread_id: "native", turn_id: "native-failed", codex_error_info: "other", + pre_execution_routing_timeout: true, only_pre_execution_items: true, only_original_user_item: false}; + const packet = {request: {request_ref: "f".repeat(24), command: null, status: "accepted", delivery_verified: true, + binding_id: row.binding_id, session_id: "original", turn_id: "failed", source}, + binding: selected, current_session: session, context_recovery: {version: 21, failed_turn: failed, + observed_at: "2026-01-01T00:10:00Z", steering_observed: false, source_verified: true, native_observation: observed}}; + const plan = planBoundConversationRequest(packet); + assert.equal(plan.recovery_eligible, true); + assert.equal(plan.routing_attempt, 1); + assert.equal(plan.not_before, "2026-01-01T00:01:00.000Z"); + const reservation = {version: 21, failed_turn_id: "failed", routing_attempt: 1, client_turn_id: plan.client_turn_id}; + assert.deepEqual(planBoundConversationRequest({...packet, request: {...packet.request, context_recovery: reservation}}), plan); + const next = {...packet, request: {...packet.request, turn_id: "next", context_recovery: reservation}, + context_recovery: {...packet.context_recovery, failed_turn: {...failed, turn_id: "next"}}}; + assert.equal(planBoundConversationRequest(next).routing_attempt, 2); + assert.equal(planBoundConversationRequest(next).not_before, "2026-01-01T00:02:00.000Z"); + assert.equal(planBoundConversationRequest({...next, request: {...next.request, + context_recovery: {...reservation, routing_attempt: 3}}}).recovery_candidate, false); + for (const change of [{source_verified: false}, {steering_observed: true}, + {observed_at: "2026-01-01T00:00:59Z"}, {observed_at: "invalid"}, + {native_observation: {...observed, only_pre_execution_items: false}}, + {native_observation: {...observed, thread_id: "foreign"}}, + {native_observation: {...observed, codex_error_info: "cyberPolicy"}}]) { + assert.notEqual(planBoundConversationRequest({...packet, + context_recovery: {...packet.context_recovery, ...change}}).recovery_eligible, true); + } + for (const change of [{status: "closed"}, {status: "stale"}, {goal_id: "other"}, {active_turn_id: "other"}, + {session_mode: "attached_host"}, {channel_id: "foreign"}]) { + assert.notEqual(planBoundConversationRequest({...packet, + current_session: {...session, ...change}}).recovery_eligible, true); + } + }); +} + for (const kind of ["project", "steward"]) { const selected = kind === "steward" ? {...row, context_kind: kind, grant: "portfolio_read", goal_ids: []} : row; const configured = planConversationBinding({...request, binding: selected}).state; diff --git a/tests/test_chat_agent.py b/tests/test_chat_agent.py index 3189289d53..509068c618 100644 --- a/tests/test_chat_agent.py +++ b/tests/test_chat_agent.py @@ -1,5 +1,7 @@ from __future__ import annotations +from loopx.capabilities.native_chat.codex_adapter import failed_request_readback + import contextlib import io import json @@ -425,7 +427,7 @@ def write(packet): session.messages.put({"id": packet["id"], "result": result}) monkeypatch.setattr(session, "_write", write) - proof = session.failed_request_readback("failed") + proof = failed_request_readback(session, "failed") if fault in {None, "older", "host-other"}: assert proof == {"thread_id": "original-thread", "turn_id": "failed", "only_original_user_item": True, "codex_error_info": "other" if fault == "host-other" else "contextWindowExceeded"} @@ -434,6 +436,58 @@ def write(packet): assert set(calls) <= {"thread/turns/list", "thread/items/list"} +@pytest.mark.parametrize("compact", [False, True]) +@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, 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 + + 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" + if fault == "changed" and reads > 1: + message = "different private error" + rows = [{"id": "failed", "status": "failed", "error": {"codexErrorInfo": "other", "message": message}}] + if fault == "older": + rows.insert(0, {"id": "newer", "status": "completed"}) + result = {"data": rows} + else: + items = [{"turnId": "failed", "item": {"id": "user", "type": "userMessage"}}] + if compact: + items.append({"turnId": "failed", "item": {"id": "compact", "type": "contextCompaction"}}) + kind = {"tool": "commandExecution", "answer": "agentMessage", "unknown": "futureItem", + "correction": "userMessage", "duplicate": "contextCompaction"}.get(fault) + if kind: + items.append({"turnId": "failed", "item": {"id": "user" if fault == "duplicate" else "extra", "type": kind}}) + # Exercise the complete paginated inventory, including late tools. + cursor = packet["params"].get("cursor") + result = {"data": items[1:] if cursor else items[:1], + "nextCursor": "next" if len(items) > 1 and not cursor else None} + session.messages.put({"id": packet["id"], "result": result}) + + monkeypatch.setattr(session, "_write", write) + proof = failed_request_readback(session, "failed") + if fault is None: + 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") + 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"]) +def test_routing_diagnostic_is_an_exact_provider_diagnostic(message): + error = {"codexErrorInfo": "other", "message": message} + assert not chat_agent._routing_discovery_timeout(error) + assert chat_agent._terminal_turn_error(error, "fallback").error_code == "host_gate" + + def test_stop_cancels_completed_and_late_command_items_but_preserves_older_terminal(tmp_path, monkeypatch): session, state = _stop_server(tmp_path, monkeypatch) session.interrupt("current-turn") diff --git a/tests/test_private_chat_routing_recovery.py b/tests/test_private_chat_routing_recovery.py new file mode 100644 index 0000000000..0e3252f662 --- /dev/null +++ b/tests/test_private_chat_routing_recovery.py @@ -0,0 +1,109 @@ +"""The original delivery worker, not a manual replay, recovers routing outages.""" +from datetime import datetime, timedelta, timezone +import json +from pathlib import Path + +import pytest + +from test_native_steward_private import steward, finish # noqa: F401 +from loopx.capabilities.native_chat import external_conversations +from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter +from loopx.extensions.lark.conversation_identity import identity_ref +from loopx.extensions.lark.private_conversations import LarkPrivateConversations +from loopx.extensions.lark.private_progress import presentation_phase + + +@pytest.mark.parametrize("profile", ["notes-app", "steward-app"]) +@pytest.mark.parametrize("fault", [None, "changed-source", "revoked", "unknown-effects", + "reservation-crash", "post-enqueue-crash"]) +def test_original_scanner_recovers_across_another_failure_and_restart(steward, monkeypatch, profile, fault): # noqa: F811 + 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 timed out"') + 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, "routing-outage", "typed terminal failure") + transport.admit(profile, event) + ref = identity_ref(binding["provider_ref"], event["message_id"]) + original = transport.core.read_request(ref) + assert finish(runtime, original)["status"] == "failed" + failed_path = store._turn_path(original["session_id"], original["turn_id"]) + original_bytes = failed_path.read_bytes() + transport.reconcile() + path = transport.root / f"{ref}.json" + old_notice = json.loads(path.read_text())["deliveries"]["terminal"] + old_body = dict(provider.messages[old_notice["attempt"]["message_ref"]]) + + class AfterBackoff(datetime): + @classmethod + def now(cls, tz=None): + return datetime.now(timezone.utc) + timedelta(hours=1) + + monkeypatch.setattr(external_conversations, "datetime", AfterBackoff) + monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(lambda **kw: None + if fault == "unknown-effects" else {"thread_id": "durable-thread", "turn_id": kw["turn_id"], + "codex_error_info": "other", "pre_execution_routing_timeout": True, + "only_pre_execution_items": True, "only_original_user_item": False})) + if fault == "changed-source": + provider.messages[event["message_id"]]["content"] = "Changed requirement" + elif fault == "revoked": + transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"]) + + def restarted_worker(): + return LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent, + runner=provider, cli_bin="lark-cli") + + first = restarted_worker() + if fault in {"reservation-crash", "post-enqueue-crash"}: + enqueue = runtime.enqueue_turn + def crash(**kwargs): + if fault == "post-enqueue-crash": + enqueue(**kwargs) + raise OSError("simulated admission crash") + monkeypatch.setattr(runtime, "enqueue_turn", crash) + with pytest.raises(OSError, match="admission crash"): + first.reconcile() + monkeypatch.setattr(runtime, "enqueue_turn", enqueue) + first = restarted_worker() + first.reconcile() + continued = first.core.read_request(ref) + if fault in {"changed-source", "revoked", "unknown-effects"}: + assert continued["turn_id"] == original["turn_id"] + assert not continued.get("context_recovery") + assert failed_path.read_bytes() == original_bytes + return + assert continued["turn_id"] != original["turn_id"] + assert continued["context_recovery"]["routing_attempt"] == 1 + assert finish(runtime, continued)["status"] == "failed" + first.reconcile_request(path) + assert first.core.read_request(ref)["delivery_verified"] + assert first.core.context_recovery_candidate(ref), first.core._context_recovery_plan(first.core.read_request(ref), discovery_only=True) + first_notice = json.loads(path.read_text())["deliveries"][presentation_phase(continued, "terminal")] + second_failed = store._turn_path(continued["session_id"], continued["turn_id"]) + second_bytes = second_failed.read_bytes() + runtime.adapters.pop(original["session_id"]).close_session() + fake.write_text(source.replace('if "typed terminal failure" in json.dumps(request.get("params") or {}):', 'if False:')) + second = restarted_worker() + second.reconcile() + recovered = second.core.read_request(ref) + assert recovered["context_recovery"]["routing_attempt"] == 2, second.core._context_recovery_plan(recovered) + assert recovered["context_recovery_history"] == [continued["context_recovery"]] + assert recovered["session_id"] == original["session_id"] + assert finish(runtime, recovered)["status"] == "completed" + second.reconcile_request(path) + delivered = json.loads(path.read_text()) + assert delivered["deliveries"]["terminal"] == old_notice + assert delivered["deliveries"][presentation_phase(continued, "terminal")] == first_notice + assert delivered["deliveries"][presentation_phase(recovered, "terminal")]["verified"] + assert provider.messages[old_notice["attempt"]["message_ref"]] == old_body + assert failed_path.read_bytes() == original_bytes and second_failed.read_bytes() == second_bytes + before = list(provider.writes) + second.reconcile_request(path) + restarted_worker().reconcile() + assert provider.writes == before + 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 From f62a5460058ce3ae9dce6927dc76b8d00358e47b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 21 Oct 2026 11:00:26 +0800 Subject: [PATCH 2/5] fix(chat): adopt already admitted recovery after a crash Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../native_chat/external_conversations.py | 18 +++++++++++++----- .../collaboration/conversation_binding.ts | 17 +++++++++++++++++ .../conversation_binding.test.ts | 12 ++++++++++++ 3 files changed, 42 insertions(+), 5 deletions(-) diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index ceed6010e4..dad13047b9 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -66,11 +66,15 @@ def _context_recovery_plan(self, row: dict[str, Any], *, return {} else: 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"]) + if reservation.get("routing_attempt") and reservation.get("failed_turn_id") == row["turn_id"] else None) return effect_runtime_result("collaboration.conversation.request", { "request": {key: value for key, value in row.items() if key != "attachments"}, "current_session": session, "binding": selected_binding, "context_recovery": {"version": MANAGER_CONTEXT_VERSION, "failed_turn": turn, "observed_at": datetime.now(timezone.utc).isoformat(), + "admitted_turn": admitted, "steering_observed": any(event.get("kind") == "turn.steered" for event in store.events_after(session["session_id"], turn["turn_id"], None)), "native_observation": observation, "source_verified": source_verified}, @@ -120,7 +124,7 @@ def recover_failed_request(self, request_ref: str, *, assert session is not None and turn is not None self.controller._check_codex_home(session) context = self.controller.project_contexts.session_context(session) - proof = CodexAppServerAdapter.failed_request_observation( + 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"]), 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) @@ -144,10 +148,14 @@ def recover_failed_request(self, request_ref: str, *, _atomic_write_json(path, row) # Freeze correlation before the existing queue can dispatch. self._record_steward_ingress(row, selected, session, plan["client_turn_id"]) - continued, _ = self.controller.enqueue_turn(session_id=row["session_id"], - client_turn_id=plan["client_turn_id"], message=row["message"], - attachments=row.get("attachments"), origin=row.get("origin", "lark"), - work_dir=context["project"], objective=context["objective"]) + if plan.get("admitted_turn_id"): + continued = self.controller.store.load_turn(row["session_id"], plan["admitted_turn_id"]) + assert continued is not None + else: + continued, _ = self.controller.enqueue_turn(session_id=row["session_id"], + client_turn_id=plan["client_turn_id"], message=row["message"], + attachments=row.get("attachments"), origin=row.get("origin", "lark"), + work_dir=context["project"], objective=context["objective"]) row.update(turn_id=continued["turn_id"], delivery_verified=False) _atomic_write_json(path, row) self.controller.resume_session_queue(session_id=row["session_id"], diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index 69f6a02aab..466fb6c52d 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -496,6 +496,23 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO || !["context_window_exceeded", "host_gate", "upstream_transport_failure", "server_overloaded"].includes(String(failed.error_code)) || typeof failed.upstream_turn_id !== "string" || !failed.upstream_turn_id) return no; const previous = row.context_recovery == null ? null : requireJsonObject(row.context_recovery, "reserved context recovery"); + if (recovery.admitted_turn != null && previous?.failed_turn_id === failed.turn_id + && Number.isSafeInteger(previous.routing_attempt) && Number(previous.routing_attempt) >= 1 + && Number(previous.routing_attempt) <= 3 && previous.version === version) { + // A crash after queue admission must adopt that exact canonical Turn, + // even if it has already executed. This branch never creates model work. + const admitted = requireJsonObject(recovery.admitted_turn, "admitted recovery Turn"); + const client = `external-${ref(row.request_ref, "original request")}-recovery-${version}-routing-${previous.routing_attempt}`; + if (previous.client_turn_id !== client || admitted.client_turn_id !== client + || admitted.session_id !== current.session_id || admitted.turn_id === failed.turn_id + || typeof row.message !== "string" || typeof row.origin !== "string" + || admitted.message !== row.message || admitted.origin !== row.origin + || !Array.isArray(admitted.attachments ?? []) || (admitted.attachments as unknown[] | undefined)?.length) return no; + return {operation: "admit_turn", recovery_candidate: true, + recovery_eligible: recovery.source_verified === true, client_turn_id: client, + admitted_turn_id: sessionIdentity(admitted.turn_id, "admitted recovery Turn"), + session_id: current.session_id, native_thread_ids: []}; + } const routing = planRoutingRecovery(recovery, row, current, failed, previous, version); if (routing !== null) return routing; if (!steward) return no; diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index e9c68cd29b..99920617fc 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/tests/control_plane_ts/conversation_binding.test.ts @@ -34,6 +34,7 @@ for (const kind of ["project", "steward"]) { const observed = {thread_id: "native", turn_id: "native-failed", codex_error_info: "other", pre_execution_routing_timeout: true, only_pre_execution_items: true, only_original_user_item: false}; const packet = {request: {request_ref: "f".repeat(24), command: null, status: "accepted", delivery_verified: true, + message: "Original synthetic request", origin: "lark", binding_id: row.binding_id, session_id: "original", turn_id: "failed", source}, binding: selected, current_session: session, context_recovery: {version: 21, failed_turn: failed, observed_at: "2026-01-01T00:10:00Z", steering_observed: false, source_verified: true, native_observation: observed}}; @@ -43,6 +44,17 @@ for (const kind of ["project", "steward"]) { assert.equal(plan.not_before, "2026-01-01T00:01:00.000Z"); const reservation = {version: 21, failed_turn_id: "failed", routing_attempt: 1, client_turn_id: plan.client_turn_id}; assert.deepEqual(planBoundConversationRequest({...packet, request: {...packet.request, context_recovery: reservation}}), plan); + const admitted = {session_id: "original", turn_id: "already-admitted", client_turn_id: plan.client_turn_id, + message: packet.request.message, origin: "lark", status: "completed", attachments: []}; + const crash = {...packet, request: {...packet.request, context_recovery: reservation}, + current_session: {...session, status: "busy", active_turn_id: "already-admitted"}, + context_recovery: {...packet.context_recovery, native_observation: null, admitted_turn: admitted}}; + assert.equal(planBoundConversationRequest(crash).admitted_turn_id, "already-admitted"); + for (const change of [{session_id: "foreign"}, {client_turn_id: "foreign"}, + {message: "another request"}, {attachments: [{type: "image"}]}]) { + assert.equal(planBoundConversationRequest({...crash, + context_recovery: {...crash.context_recovery, admitted_turn: {...admitted, ...change}}}).recovery_eligible, false); + } const next = {...packet, request: {...packet.request, turn_id: "next", context_recovery: reservation}, context_recovery: {...packet.context_recovery, failed_turn: {...failed, turn_id: "next"}}}; assert.equal(planBoundConversationRequest(next).routing_attempt, 2); From fc3da3e0d3fd1bd51f713fcb6cb49fbdea463f2e Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 21 Oct 2026 11:06:07 +0800 Subject: [PATCH 3/5] test(chat): advance recovery backoff with a controlled clock Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/test_private_chat_routing_recovery.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tests/test_private_chat_routing_recovery.py b/tests/test_private_chat_routing_recovery.py index 0e3252f662..d86055a540 100644 --- a/tests/test_private_chat_routing_recovery.py +++ b/tests/test_private_chat_routing_recovery.py @@ -37,10 +37,11 @@ def test_original_scanner_recovers_across_another_failure_and_restart(steward, m old_notice = json.loads(path.read_text())["deliveries"]["terminal"] old_body = dict(provider.messages[old_notice["attempt"]["message_ref"]]) + clock = [datetime.now(timezone.utc) + timedelta(seconds=61)] class AfterBackoff(datetime): @classmethod def now(cls, tz=None): - return datetime.now(timezone.utc) + timedelta(hours=1) + return clock[0] monkeypatch.setattr(external_conversations, "datetime", AfterBackoff) monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(lambda **kw: None @@ -80,12 +81,13 @@ def crash(**kwargs): assert finish(runtime, continued)["status"] == "failed" first.reconcile_request(path) assert first.core.read_request(ref)["delivery_verified"] - assert first.core.context_recovery_candidate(ref), first.core._context_recovery_plan(first.core.read_request(ref), discovery_only=True) + assert not first.core.context_recovery_candidate(ref) # The next backoff survives a scanner restart. first_notice = json.loads(path.read_text())["deliveries"][presentation_phase(continued, "terminal")] second_failed = store._turn_path(continued["session_id"], continued["turn_id"]) second_bytes = second_failed.read_bytes() runtime.adapters.pop(original["session_id"]).close_session() fake.write_text(source.replace('if "typed terminal failure" in json.dumps(request.get("params") or {}):', 'if False:')) + clock[0] += timedelta(minutes=5) second = restarted_worker() second.reconcile() recovered = second.core.read_request(ref) From 0d5858663e55607896ae5992f4ef4ea2ceb8c18f Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 21 Oct 2026 11:10:37 +0800 Subject: [PATCH 4/5] docs(chat): disclose bounded routing recovery and retained failure history Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 21 +++++++++++++++++-- ...p-conversation-and-async-inbox-v0.zh-CN.md | 13 +++++++++++- 2 files changed, 31 insertions(+), 3 deletions(-) diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md index 659b812147..834b9da0ad 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -1352,8 +1352,8 @@ whose native error is `other`, only when the stable native inventory proves there was exactly the original user input and no execution item. It resumes the existing thread, without a history migration or error-message text matching. This proves absence of execution, not provider health or eventual success. A -terminal Core failure is returned and retained; it cannot enqueue another -continuation. +terminal Core failure on this unclassified-host/context-migration path is +returned and retained; it cannot enqueue another continuation on that path. The existing in-flight transport retry also recognizes the exact native `other` diagnostic `workspace routing discovery timed out`. This provider hint @@ -1365,6 +1365,23 @@ budget; it never reopens a terminal historical request or resets a reservation after restart. Native error prose is not copied into the retry journal. Existing App and Lark retry progress and final readback expose the same request's outcome. +After those in-flight retries end, the original private reply worker can also +recover a proven routing outage for ordinary project assistants and stewards. +The exact diagnostic above, or its native remote-compaction wrapper, must be +independently read back from the latest failed native Turn. Its complete stable +inventory may contain only one user input and native compaction bookkeeping; +answers, tools, corrections, unknown items and active native work remain holds. +Fresh source, App/owner and grant verification still precede each admission. +Core reserves at most three new, correlated continuations per original request, +with 60/120/240-second backoff from the preceding persisted failure. Restart +neither resets this budget nor reopens the failed Turn. A lost acknowledgement +after queue admission adopts the exact existing client identity, including an +already completed Turn, instead of enqueueing it again. Recovery attempts retain +their own presentation keys and earlier reservation/notice history. A delivered +failure of a continuation remains unfinished work; a completed answer does not +activate recovery. This does not establish provider availability or eventual +business completion, and it creates no additional event consumer. + Recovery discovery backoff is only a temporary hint. The existing reply worker retains its per-request fairness count while that canonical request still exists; an omitted hint does not retire work or let earlier unresolved requests repeatedly diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md index d03d70ca9b..754e0a043c 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md @@ -861,7 +861,8 @@ Turn,确认完整条目清单仅包含原用户输入后,Core 才能按每 该有界准入也覆盖当前格式中的 `host_gate` 失败:原生错误为 `other`,且稳定、完整 的原生清单证明只有原用户输入,没有任何执行条目时,才允许一次续办。它恢复既有 线程,不迁移历史,也不匹配错误消息文本。这证明尚未执行,不证明 provider 健康 -或最终成功;续办若进入 Core 失败终态,就回传并保留失败,不能再生成下一次续办。 +或最终成功;这一未分类宿主/上下文迁移路径的续办若进入 Core 失败终态,就回传并 +保留失败,不能沿这一条路径再生成下一次续办。 原有同轮传输重试也识别原生 `other` 类别下的确切诊断 `workspace routing discovery timed out`。这个 provider 提示本身不授予重试权: @@ -871,6 +872,16 @@ Turn、输入、Session 和至多三次的既有重试及退避预算,不重 也不因重启重置预留。重试 journal 不复制原生错误正文;App 与 Lark 沿既有进展和 最终读回路径展示同一请求的实际结果。 +同轮重试结束后,原私聊回复 worker 也可为普通项目助手与管家恢复已证实的路由故障。 +必须从最新失败的原生 Turn 独立读回上述确切诊断,或原生远端压缩任务附加的同一诊断。 +稳定、完整的条目清单只允许一条用户输入与原生压缩记账;回答、工具、追加纠正、未知 +条目和活动原生工作仍阻止恢复。每次受理前继续重新核验原消息、App/本人和 grant。 +Core 对每条原请求至多预留三次新的关联续办,分别从前次持久化失败起退避 60/120/240 +秒;重启不重置预算,也不重新打开失败 Turn。排队成功后丢失确认时,采用确切 client +身份对应的既有 Turn,包括已经完成的 Turn,不再次排队。每次恢复有独立的呈现 key, +并保留之前的预留与失败通知历史。续办的失败通知已送达仍不等于工作完成;已完成答复 +不会激活恢复。这不证明 provider 可用或业务最终完成,也不增加事件消费者。 + 恢复发现的退避只是临时提示。规范请求仍存在时,原回复 worker 保留该请求的公平 计数;某轮没有提示不代表工作已退役,也不能让前面的未结请求反复占满全部空位。 四 worker 和不增加执行积压的上限保持。 From b8c7cb90ccd1d6ab7afa1633899fc819561a4fd5 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 21 Oct 2026 11:15:01 +0800 Subject: [PATCH 5/5] test(chat): cover inventory order and excluded recovery scopes Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/control_plane_ts/conversation_binding.test.ts | 6 ++++++ tests/test_chat_agent.py | 5 ++++- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index 99920617fc..122aba3bbc 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/tests/control_plane_ts/conversation_binding.test.ts @@ -74,6 +74,12 @@ for (const kind of ["project", "steward"]) { assert.notEqual(planBoundConversationRequest({...packet, current_session: {...session, ...change}}).recovery_eligible, true); } + if (kind === "project") { + assert.equal(planBoundConversationRequest({...packet, + binding: {...selected, audience: "group", group_refs: ["1".repeat(24)]}}).recovery_eligible, false); + assert.equal(planBoundConversationRequest({...packet, + binding: {...selected, grant: "workspace_write"}}).recovery_eligible, false); + } }); } diff --git a/tests/test_chat_agent.py b/tests/test_chat_agent.py index 509068c618..6d84902b2d 100644 --- a/tests/test_chat_agent.py +++ b/tests/test_chat_agent.py @@ -437,8 +437,9 @@ def write(packet): @pytest.mark.parametrize("compact", [False, True]) +@pytest.mark.parametrize("reverse", [False, True]) @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, fault): +def test_routing_failure_readback_allows_only_native_compaction_before_execution(tmp_path, monkeypatch, compact, reverse, 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 @@ -462,6 +463,8 @@ def write(packet): "correction": "userMessage", "duplicate": "contextCompaction"}.get(fault) if kind: items.append({"turnId": "failed", "item": {"id": "user" if fault == "duplicate" else "extra", "type": kind}}) + if reverse: + items.reverse() # Exercise the complete paginated inventory, including late tools. cursor = packet["params"].get("cursor") result = {"data": items[1:] if cursor else items[:1],