From d198aa59ed7893d3c43bb0376076567328409d1e Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 00:34:19 +0800 Subject: [PATCH 1/2] fix(chat): recover interrupted Codex steward answers through the original queue Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../capabilities/native_chat/codex_adapter.py | 7 +- .../native_chat/external_conversations.py | 14 +- .../collaboration/conversation_binding.ts | 27 ++-- loopx/extensions/lark/manager_context.py | 3 + .../conversation_binding.test.ts | 59 +++++++++ tests/test_chat_agent.py | 12 +- tests/test_steward_interrupted_answer.py | 121 ++++++++++++++++++ 7 files changed, 229 insertions(+), 14 deletions(-) diff --git a/loopx/capabilities/native_chat/codex_adapter.py b/loopx/capabilities/native_chat/codex_adapter.py index a9ed165aba..5969f5978d 100644 --- a/loopx/capabilities/native_chat/codex_adapter.py +++ b/loopx/capabilities/native_chat/codex_adapter.py @@ -413,7 +413,7 @@ def turns() -> list[dict[str, Any]]: elif kind == "agentMessage": if item.get("phase") != "commentary": return None - elif kind in {"commandExecution", "fileChange", "mcpToolCall"}: + elif kind in {"commandExecution", "fileChange", "mcpToolCall", "dynamicToolCall"}: if item.get("status") not in {"completed", "failed", "declined"}: return None effects += 1 @@ -423,7 +423,10 @@ def turns() -> list[dict[str, Any]]: or list(session._control_items("thread/items/list", {"turnId": turn_id}, deadline)) != items): return None return {"thread_id": session.thread_id, "turn_id": turn_id, "read_only_review": True, - "no_final_answer": True, "effects_observed": True} + "no_final_answer": True, "effects_observed": True, + "native_inventory_sha256": hashlib.sha256(json.dumps([ + next(row for row in inventory if row["id"] == turn_id), items], + sort_keys=True, separators=(",", ":")).encode()).hexdigest()} except (CodexChatAgentError, OSError, TypeError, ValueError): return None diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index fae7278b88..3c9d796f71 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -379,6 +379,7 @@ def recover_failed_request(self, request_ref: str, *, return self._recover_native_result(path, row, session, turn, source_observation) metadata_review = plan.get("metadata_review") is True answer_review = plan.get("answer_review") is True + codex_answer = answer_review and session.get("agent_id") == "codex" quota_answer = answer_review and session.get("agent_id") == "codex" and turn.get("error_code") == "usage_limit_exceeded" preparation_retry = plan.get("preparation_attempt") is not None model_auth_home = (None if metadata_review or answer_review or preparation_retry @@ -397,7 +398,7 @@ def source_requests() -> list[dict[str, Any]]: # checks and the typed eligibility decision remain below that fence. proof = None capacity = None - if not plan.get("admitted_turn_id") and (metadata_review or answer_review and not quota_answer or turn.get("error_code") == "antigravity_step_error"): + if not plan.get("admitted_turn_id") and (metadata_review or answer_review and not codex_answer or turn.get("error_code") == "antigravity_step_error"): from .antigravity_adapter import AntigravityAdapter, ANTIGRAVITY_DEFAULT_MODEL events = self.controller.store.events_after(row["session_id"], row["turn_id"], None) anchors = [item["payload"]["native_input"] for item in events @@ -452,6 +453,11 @@ def source_requests() -> list[dict[str, Any]]: capacity = {"alternate_executor": route} else: proof.update(quota_available=True, quota_observed_at=route["observed_at"]) + if codex_answer and not quota_answer and proof is not None: + from ...chat_manager import manager_session_quota_route + route = manager_session_quota_route(self.controller, session) + if route and route["allocation"]["executor_endpoint"] == "codex": + capacity = {"alternate_executor": route} facts = proof.pop("retained_result", None) if proof is not None and metadata_review else None selected = self.bindings.resolve(binding_id=row["binding_id"], **row["source"], include_observation=True) source = source_observation(selected) @@ -801,6 +807,12 @@ def _checked_answer_review(self, session: dict[str, Any], turn: dict[str, Any], if observed is not None: observed = {key: observed.get(key) for key in ("thread_id", "turn_id", "only_original_user_item", "codex_error_info")} expected = {key: proof.get(key) for key in ("thread_id", "turn_id", "only_original_user_item", "codex_error_info")} + elif proof.get("read_only_review") is True and proof.get("turn_id"): + from .codex_adapter import CodexAppServerAdapter + observed = CodexAppServerAdapter.failed_request_observation(codex_bin=self.controller.codex_bin, + codex_home=Path(session["codex_home"]), thread_ids=[proof["thread_id"]], + turn_id=proof["turn_id"], read_only_review=True) + expected = proof else: observed = AntigravityAdapter.interrupted_answer_observation(cascade_id=proof["thread_id"], request_ref=row["request_ref"], message=original["message"], started_at=original["started_at"], diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index 4aa290185e..51d692daf0 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -743,15 +743,15 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO if ((previous?.read_only_review === true && previous.failed_turn_id !== row.turn_id) || (Array.isArray(row.context_recovery_history) && row.context_recovery_history.some(value => requireJsonObject(value, "prior recovery").read_only_review === true))) return no; + if (failed.error_code === "server_restarted") { + return planResultReview(recovery, row, current, failed, previous, version); + } const exhaustedAnswer = steward && current.agent_id === "codex" ? planExhaustedStewardAnswer(recovery, row, current, failed, previous, version) : null; if (exhaustedAnswer !== null) return exhaustedAnswer; if (failed.status === "completed" && !steward) { return planUnfinishedWork(recovery, row, current, failed, previous, version); } - if (!steward && failed.error_code === "server_restarted") { - return planResultReview(recovery, row, current, failed, previous, version); - } if (failed.session_id !== current.session_id || failed.turn_id !== row.turn_id || failed.status !== "failed" || failed.delta_count !== 0 || failed.response !== null || recovery.steering_observed !== false @@ -1223,16 +1223,20 @@ function planInterruptedStewardAnswer(recovery: JsonObject, row: JsonObject, cur return {...capacity, recovery_eligible: capacity.recovery_eligible === true && capacity.answer_executor === "codex"}; } -/** Original project result reviews retain their workspace-only read policy. */ +/** Reuse the original queue for a restricted result review, never business replay. */ function planResultReview(recovery: JsonObject, row: JsonObject, current: JsonObject, failed: JsonObject, previous: JsonObject | null, version: number): JsonObject { const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false}; + const steward = current.steward_context != null; + const review = steward ? {answer_review: true, answer_executor: "codex"} : {}; if (failed.session_id !== current.session_id || failed.turn_id !== row.turn_id || failed.status !== "failed" || failed.response !== null || !Number.isSafeInteger(failed.delta_count) || Number(failed.delta_count) <= 0 || recovery.steering_observed !== false || recovery.stop_observed !== false || recovery.attachments_present !== false - || typeof failed.upstream_turn_id !== "string" || !failed.upstream_turn_id) return no; + || typeof failed.upstream_turn_id !== "string" || !failed.upstream_turn_id + || steward && (failed.client_turn_id !== `external-${row.request_ref}` + || failed.message !== row.message || failed.origin !== row.origin)) return no; const client = `external-${ref(row.request_ref, "original request")}-recovery-${version}-review-1`; if (previous?.read_only_review === true && (previous.failed_turn_id !== failed.turn_id || previous.version !== version || previous.client_turn_id !== client)) return no; @@ -1242,7 +1246,7 @@ function planResultReview(recovery: JsonObject, row: JsonObject, current: JsonOb || admitted.turn_id === failed.turn_id || admitted.message !== row.message || admitted.origin !== row.origin || (admitted.attachments != null && (!Array.isArray(admitted.attachments) || admitted.attachments.length))) return no; - return {...no, recovery_candidate: true, recovery_eligible: recovery.source_verified === true, + return {...no, ...review, recovery_candidate: true, recovery_eligible: recovery.source_verified === true, read_only_review: true, client_turn_id: client, session_id: current.session_id, admitted_turn_id: sessionIdentity(admitted.turn_id, "admitted result review"), native_thread_ids: []}; } @@ -1250,15 +1254,20 @@ function planResultReview(recovery: JsonObject, row: JsonObject, current: JsonOb const now = Date.parse(String(recovery.observed_at)), completed = Date.parse(String(failed.completed_at)); if (!Number.isFinite(now) || !Number.isFinite(completed) || now < completed + 60_000) return no; const native = sessionIdentity(current.upstream_thread_id, "original native thread"); - const candidate = {...no, recovery_candidate: true, read_only_review: true, + const candidate = {...no, ...review, recovery_candidate: true, read_only_review: true, client_turn_id: client, session_id: current.session_id, failed_turn_id: failed.turn_id, native_thread_ids: [native]}; if (recovery.native_observation == null) return candidate; const observed = requireJsonObject(recovery.native_observation, "interrupted result inventory"); - return {...candidate, recovery_eligible: recovery.source_verified === true + const eligible = recovery.source_verified === true && observed.thread_id === native && observed.turn_id === failed.upstream_turn_id && observed.read_only_review === true && observed.no_final_answer === true - && observed.effects_observed === true}; + && observed.effects_observed === true && (!steward + || typeof observed.native_inventory_sha256 === "string" + && BARE_SHA256_PATTERN.test(observed.native_inventory_sha256)); + if (!eligible || !steward) return {...candidate, recovery_eligible: eligible}; + const capacity = nativeRecoveryCapacity(recovery, row, current, candidate); + return {...capacity, recovery_eligible: capacity.recovery_eligible === true && capacity.answer_executor === "codex"}; } /** A local manager-context failure precedes native dispatch. Only the Core's diff --git a/loopx/extensions/lark/manager_context.py b/loopx/extensions/lark/manager_context.py index 6d6ee14859..71067fe63c 100644 --- a/loopx/extensions/lark/manager_context.py +++ b/loopx/extensions/lark/manager_context.py @@ -80,6 +80,9 @@ def manager_failure_reply( if code in {"model_auth_unavailable", "resume_failed"} and preparation_recovery: return code, ("已收到你的消息，但本轮准备执行器连接时失败。请求已保留；只有核验为模型调用前失败、" "连接恢复且没有后续重交时，原队列才会按剩余次数接续。可用 /status 查看状态。") + if code == "server_restarted": + return code, ("处理被管家服务重启中断，原请求和会话已保留。系统会先核对原执行记录，" + "符合条件的由原队列恢复结果回复；执行效果不明的操作不会直接重做。") # Old hosts can terminate with `other` after typed disconnect retries. # Describe that observed history without changing the terminal category, # inferring an authorization failure, or granting permission to replay. diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index 3e56cdaeb4..dd9d66e42a 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/tests/control_plane_ts/conversation_binding.test.ts @@ -19,6 +19,65 @@ const current = {schema_version: "loopx_chat_conversation_bindings_v0", revision const request = {current, expected_revision: 0, operation: "configure", binding: row, observation, available_projects: [project]}; +test("a restarted Codex steward retains one restricted answer instead of replaying business", () => { + const selected = {...row, context_kind: "steward", grant: "portfolio_read", goal_ids: []}; + 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 ref = "f".repeat(24), message = "Explain the existing results"; + const session = {session_id: "original", goal_id: "loopx-manager", agent_id: "codex", upstream_mode: "chat", + status: "ready", active_turn_id: null, upstream_thread_id: "native", steward_context: context, + channel_id: `manager.external.native.${row.binding_id}.${source.source_ref}`}; + const failed = {session_id: "original", turn_id: "failed", client_turn_id: `external-${ref}`, + message, origin: "lark", status: "failed", error_code: "server_restarted", response: null, + delta_count: 1, upstream_turn_id: "interrupted", completed_at: "2026-01-01T00:00:00Z"}; + const proof = {thread_id: "native", turn_id: "interrupted", read_only_review: true, + no_final_answer: true, effects_observed: true, native_inventory_sha256: "1".repeat(64)}; + const allocation = {schema_version: "manager_executor_allocation_v0", selection_policy: "flexible", + configured_endpoint: "antigravity", executor_endpoint: "codex", allocation_reason: "flexible_availability_fallback", + configuration_revision: "current-pool", available: true, eligible_endpoints: ["antigravity", "codex"]}; + const configuration = {status: "ready", selection_policy: "flexible", executor_endpoint: "antigravity", + eligible_endpoints: ["antigravity", "codex"], configuration_revision: "current-pool"}; + const packet = {request: {request_ref: ref, command: null, status: "accepted", delivery_verified: true, + message, origin: "lark", binding_id: row.binding_id, session_id: "original", turn_id: "failed", source}, + binding: selected, current_session: session, context_recovery: {version: 22, failed_turn: failed, + observed_at: "2026-01-01T00:10:00Z", source_verified: true, steering_observed: false, + stop_observed: false, attachments_present: false, queued_input_present: false, native_observation: proof, + metadata_capacity: {alternate_executor: {allocation, observed_at: Date.parse("2026-01-01T00:10:00Z") / 1000}}, + executor_configuration: configuration}}; + const plan = planBoundConversationRequest(packet); + assert.equal(plan.recovery_eligible, true); + assert.equal(plan.answer_review, true); + assert.equal(plan.read_only_review, true); + assert.equal(plan.answer_executor, "codex"); + assert.equal(plan.work_attempt, undefined); + assert.equal(plan.quota_attempt, undefined); + assert.equal(plan.client_turn_id, `external-${ref}-recovery-22-review-1`); + for (const change of [{source_verified: false}, {stop_observed: true}, {steering_observed: true}, + {attachments_present: true}, {observed_at: "2026-01-01T00:00:59Z"}, {metadata_capacity: null}, + ...[{thread_id: "foreign"}, {turn_id: "other"}, {no_final_answer: false}, + {effects_observed: false}, {read_only_review: false}, {native_inventory_sha256: undefined}] + .map(value => ({native_observation: {...proof, ...value}})), + ...[{delta_count: 0}, {client_turn_id: "foreign"}, {message: "changed"}, {origin: "web"}, + {response: {}}, {error_code: "runtime_error"}].map(value => ({failed_turn: {...failed, ...value}}))]) { + assert.notEqual(planBoundConversationRequest({...packet, + context_recovery: {...packet.context_recovery, ...change}}).recovery_eligible, true); + } + assert.notEqual(planBoundConversationRequest({...packet, + current_session: {...session, status: "busy", active_turn_id: "later"}}).recovery_eligible, true); + const previous = {version: 22, failed_turn_id: "failed", client_turn_id: plan.client_turn_id, + answer_review: true, read_only_review: true}; + const admitted = {session_id: "original", turn_id: "review", client_turn_id: plan.client_turn_id, + message, origin: "lark"}; + const reserved = {...packet, request: {...packet.request, context_recovery: previous}, + context_recovery: {...packet.context_recovery, admitted_turn: admitted}}; + assert.equal(planBoundConversationRequest(reserved).admitted_turn_id, "review"); + assert.equal(planBoundConversationRequest({...reserved, request: {...reserved.request, turn_id: "review"}, + context_recovery: {...reserved.context_recovery, admitted_turn: null, + failed_turn: {...failed, turn_id: "review", client_turn_id: plan.client_turn_id}}}).recovery_candidate, false); +}); + test("an original project account-preparation failure shares its existing recovery budget", () => { const configured = planConversationBinding(request).state; const source = {source_ref: "e".repeat(24), sender_ref: row.operator_ref, private_human_message: true}; diff --git a/tests/test_chat_agent.py b/tests/test_chat_agent.py index d887a9afdb..2f3781aafb 100644 --- a/tests/test_chat_agent.py +++ b/tests/test_chat_agent.py @@ -202,7 +202,7 @@ def write(packet): @pytest.mark.parametrize("fault", [None, "active", "completed", "final", "missing-phase", "correction", - "unknown", "foreign", "duplicate", "live-tool", "late-items", "late-turn"]) + "unknown", "foreign", "duplicate", "live-tool", "late-items", "late-turn", "dynamic-tool", "live-dynamic-tool", "later-completed"]) def test_interrupted_result_review_requires_stable_full_inventory_without_final_or_live_effects(tmp_path, monkeypatch, fault): session = chat_agent.CodexChatAgentSession(process=SimpleNamespace(poll=lambda: None), messages=queue.Queue(), thread_id="original", work_dir=tmp_path, response_timeout_sec=.2) @@ -220,6 +220,8 @@ def write(packet): {"id": "interrupted", "status": "completed" if fault == "completed" else "interrupted"}] if fault == "late-turn" and turns_read > 1: rows.append({"id": "new", "status": "inProgress"}) + if fault == "later-completed" and turns_read > 2: + rows.append({"id": "other-completed", "status": "completed"}) result = {"data": rows} else: assert method == "thread/items/list" @@ -232,6 +234,8 @@ def write(packet): item.update({"type": "futureItem"} if fault == "unknown" else {"id": "user"} if fault == "duplicate" else {"status": "inProgress"} if fault == "live-tool" else {}) + if fault in {"dynamic-tool", "live-dynamic-tool"}: + item.update(type="dynamicToolCall", status="completed" if fault == "dynamic-tool" else "inProgress") rows = [{"turnId": "foreign" if fault == "foreign" else "interrupted", "item": item}, {"turnId": "interrupted", "item": {"id": "comment", "type": "agentMessage", "phase": "commentary"}}] if fault in {"final", "missing-phase", "correction"} or fault == "late-items" and items_read > 1: @@ -243,10 +247,14 @@ def write(packet): monkeypatch.setattr(session, "_write", write) proof = interrupted_result_readback(session, "interrupted") - if fault is None: + if fault in {None, "dynamic-tool", "later-completed"}: + digest = proof.pop("native_inventory_sha256") + assert len(digest) == 64 and set(digest) <= set("0123456789abcdef") assert proof == {"thread_id": "original", "turn_id": "interrupted", "read_only_review": True, "no_final_answer": True, "effects_observed": True} assert turns_read == items_read == 2 + if fault == "later-completed": + assert interrupted_result_readback(session, "interrupted")["native_inventory_sha256"] == digest else: assert proof is None assert set(calls) <= {"thread/turns/list", "thread/items/list"} diff --git a/tests/test_steward_interrupted_answer.py b/tests/test_steward_interrupted_answer.py index 755b468be3..8e76c03249 100644 --- a/tests/test_steward_interrupted_answer.py +++ b/tests/test_steward_interrupted_answer.py @@ -15,6 +15,127 @@ from loopx.extensions.lark.conversation_identity import identity_ref +@pytest.mark.parametrize("fault", [None, "stop", "source-changed", "unknown-native", "no-capacity", "native-changed", "review-action", "post-enqueue-crash"]) +def test_restarted_codex_steward_gets_one_original_read_only_answer(steward, monkeypatch, fault): + from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter + from loopx.chat_agent import CodexChatAgentError + from loopx.extensions.lark.private_conversations import LarkPrivateConversations + + store, runtime, provider, transport, binding, capture, workspace = steward + fake = Path(runtime.codex_bin) + code = fake.read_text().replace('result = {"thread": {"id": "durable-thread"}}', '''result = {"thread": {"id": "durable-thread"}, "model": "gpt-6.1-sol"} + permissions = request.get("params", {}).get("permissions") + if permissions: + result.update(activePermissionProfile={"id": permissions}, runtimeWorkspaceRoots=[request["params"]["cwd"]])''') + if fault == "review-action": + start = code.index(" response = 'Visible runtime response.") + end = code.index(' print(json.dumps({"method": "item/agentMessage/delta"', start) + reply = '' + json.dumps({"message": "Forbidden repeat", + "unfinished_work": {"remaining": "Repeat business", "dependency": "owner", "effects_settled": True}}) + '' + code = code[:start] + f" response = {reply!r}\n" + code[end:] + fake.write_text(code) + artifact = workspace / "original-output.txt" + start_turn = CodexAppServerAdapter.start_turn + + def restart_after_output(adapter, message, sink, **kwargs): + if adapter.session.answer_review: + return start_turn(adapter, message, sink, **kwargs) + def observed(kind, payload): + sink(kind, payload) + if kind == "answer.delta": + artifact.write_text("Original effect, do not repeat") + raise CodexChatAgentError("Synthetic service loss after output", error_code="server_restarted", gate=None) + return start_turn(adapter, message, observed, **kwargs) + + monkeypatch.setattr(CodexAppServerAdapter, "start_turn", restart_after_output) + event = provider.event("steward-app", "restart-original-question", "Explain the existing change") + transport.admit("steward-app", event) + ref = identity_ref(binding["provider_ref"], event["message_id"]) + original = transport.core.read_request(ref) + failed = runtime.wait_for_turn(session_id=original["session_id"], turn_id=original["turn_id"], timeout_sec=15) + assert failed["status"] == "failed" and failed["error_code"] == "server_restarted" and failed["delta_count"] > 0 + path = transport.root / f"{ref}.json" + transport.reconcile_request(path) + failure_bytes = store._turn_path(original["session_id"], original["turn_id"]).read_bytes() + notice = deepcopy(json.loads(path.read_text())["deliveries"]) + class Ahead(datetime): + @classmethod + def now(cls, tz=None): + return super().now(tz) + timedelta(seconds=600) + monkeypatch.setattr(external_conversations, "datetime", Ahead) + import loopx.chat_manager as manager_module + import loopx.chat_runtime as runtime_module + from test_chat_manager_model_adoption import configure + configure(runtime.coordination_runtime_root, model=None, effort=None, endpoint="antigravity", policy="flexible") + allocation = manager_module.manager_executor_allocation(runtime, + endpoint_capabilities=[{"agent_id": "codex", "available": True}, {"agent_id": "antigravity", "available": False}]) + def route(*_args, **_kwargs): + return None if fault == "no-capacity" else {"allocation": allocation, "blocked_primary": None, + "observed_at": time.time()} + monkeypatch.setattr(manager_module, "manager_session_quota_route", lambda *_args, **_kwargs: + None if fault == "no-capacity" else {"allocation": allocation, "blocked_primary": None, + "observed_at": Ahead.now().timestamp()}) + monkeypatch.setattr(runtime_module, "manager_session_quota_route", route) + probes = [] + def observe(**kwargs): + assert kwargs.get("read_only_review") is True + probes.append(kwargs) + if fault == "unknown-native": + return None + changed = fault == "native-changed" and bool((transport.core.read_request(ref).get("context_recovery") or {}).get("answer_review")) + return {"thread_id": "durable-thread", "turn_id": failed["upstream_turn_id"], + "read_only_review": True, "no_final_answer": True, "effects_observed": True, + "native_inventory_sha256": ("2" if changed else "1") * 64} + monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(observe)) + if fault == "stop": + store.append_event(original["session_id"], original["turn_id"], kind="turn.interrupt_requested", payload={}) + elif fault == "source-changed": + provider.messages[event["message_id"]]["content"] = "Revised original source" + restarted = LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent, + runner=provider, cli_bin="lark-cli") + if fault == "post-enqueue-crash": + enqueue = runtime.enqueue_turn + def crash(**kwargs): + enqueue(**kwargs) + raise OSError("synthetic unknown admission response") + monkeypatch.setattr(runtime, "enqueue_turn", crash) + with pytest.raises(OSError, match="unknown admission response"): + restarted.reconcile_request(path) + monkeypatch.setattr(runtime, "enqueue_turn", enqueue) + restarted.reconcile_request(path) + current = restarted.core.read_request(ref) + if fault in {"stop", "source-changed", "unknown-native", "no-capacity"}: + assert current["turn_id"] == original["turn_id"] and not current.get("context_recovery") + else: + recovery = current["context_recovery"] + assert recovery["answer_review"] is True and recovery["read_only_review"] is True + assert not any(recovery.get(key) for key in ["work_attempt", "quota_attempt", "routing_attempt"]) + done = runtime.wait_for_turn(session_id=current["session_id"], turn_id=current["turn_id"], timeout_sec=15) + assert done["status"] == ("failed" if fault in {"native-changed", "review-action"} else "completed"), done + restarted.reconcile_request(path) + delivered = json.loads(path.read_text()) + terminal = delivered["deliveries"]["terminal.recovery.22.review.1"] + assert terminal["verified"] + actual = provider.messages[terminal["attempt"]["message_ref"]] + assert actual["chat_id"] == event["chat_id"] and actual["sender"]["id"] == provider.profile_apps["steward-app"] + assert json.loads(actual["body"]["content"])["zh_cn"]["content"][0][0]["text"] == terminal["text"] + assert delivered["deliveries"]["terminal"] == notice["terminal"] + calls = [json.loads(line) for line in capture.read_text().splitlines()] + review_threads = [c for c in calls if c["method"] == "thread/start" and c["params"].get("permissions")] + assert len(review_threads) == 1 + opened = review_threads[0]["params"] + assert opened["config"]["features"]["apply_patch_freeform"] is False + assert opened["config"]["features"]["apps"] is False + assert opened["config"]["permissions"][opened["permissions"]]["network"]["enabled"] is False + assert [t["name"] for t in opened["dynamicTools"]] == ["loopx_manager_read"] + assert not any(c["method"] == "thread/resume" for c in calls), "unknown old business must not be resumed" + writes = list(provider.writes) + restarted.reconcile_request(path) + assert provider.writes == writes + assert store._turn_path(original["session_id"], original["turn_id"]).read_bytes() == failure_bytes + assert artifact.read_text() == "Original effect, do not repeat" + + @pytest.mark.parametrize("fault", [None, "stop", "source-changed", "revoked", "unknown-native", "native-changed", "no-capacity", "review-gate", "review-action", "post-enqueue-crash"]) def test_exhausted_quota_gets_one_restricted_original_answer(steward, monkeypatch, fault): from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter From f6766dde828b0d27312a302a2ad361b1ab6eca5b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 00:34:19 +0800 Subject: [PATCH 2/2] docs(chat): document qualified steward restart recovery Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 19 +++++++++++++++++++ ...p-conversation-and-async-inbox-v0.zh-CN.md | 13 +++++++++++++ 2 files changed, 32 insertions(+) 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 95a593a95c..3006adefde 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -1816,6 +1816,25 @@ native inventory reads and crashes before admission, after enqueue and before transport synchronization. Synthetic model/provider journeys and read-only native observations do not certify live model completion or the wider conversation SLA. +### A steward answer interrupted by service restart + +A Codex steward using the configured flexible allowance route can reserve one +restricted answer review after `server_restarted`, even when tools already ran. +The original reply worker must independently read a stable, complete interrupted +native inventory with no final answer or live items. Native dynamic-tool calls +are effect items too; a stable terminal inventory permits only this read-only +review, never business replay. Its digest is rechecked before review execution. + +The same source, owner, grant, stop checks and `review-1` queue reservation apply. +Fresh configured Codex capacity must verify before admission and again before +dispatch. The new restricted thread retains only authorized manager reads; writes, +network effects, handoffs and business continuation are disabled. It answers from +existing evidence or gives a concrete partial answer. The original failed Turn +and failure notice remain unchanged; restart or an uncertain enqueue response +adopts the reserved review instead of granting another attempt. Missing native +evidence or capacity remains pending. The failure notice describes this qualified +recovery instead of giving every restart the same blanket no-replay message. + ### Explicit unfinished work after a completed Turn A terminal model Turn is distinct from completed business work. A private 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 e783faf013..08ff985dfd 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 @@ -1185,6 +1185,19 @@ provider 当前正文等于已知前文或目标正文。正文改变、效果 传输同步前的中断。合成模型/provider 通路及只读原生观察不证明真实模型完成， 也不代表更广泛的会话 SLA 已通过。 +### 管家答复被服务重启中断 + +使用已配置灵活额度路由的 Codex 管家，遇到 `server_restarted` 且工具已经执行后， +仍可预约一次受限的答案复核。原回复 worker 必须独立读回稳定、完整、没有最终答案 +和活动 item 的原生中断清单。原生 dynamic-tool 调用同样属于效果条目；稳定的终态 +清单只允许这次只读复核，不授予业务重放权。实际复核前再次核验清单摘要。 + +复核沿用原来源、本人、grant、停止检查及同一 `review-1` 队列预留。受理及派发前 +都须重新核验已配置的 Codex 额度。新受限 thread 只保留原授权管家读取，禁用写入、 +网络效果、委派及业务续办；基于现有证据答复，或给出具体的部分答案。原失败 Turn +与失败通知不变；重启或入队回执不明时采用已预留复核，不产生新次数。缺少原生证据 +或额度仍保留待办。失败提示说明这些恢复条件，不再对所有重启一律提示不会自动重放。 + ### Turn 完成后的显式未完工作 模型 Turn 终态与业务完成分开。私有项目 worker 核验已结束工作的实际结果后，可在