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 1ab737b14a..903e837f71 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -1995,6 +1995,13 @@ card cannot substitute for final-message readback. ### Recovering a native result after Core failure +A native terminal error remains a failure. When the complete current invocation +is idle, every observed step is settled and its final error is done, the adapter +retires that invocation's cancellation target before returning the structured +error. Active or unknown steps retain existing bounded cancellation and permission +handling. Avoiding a second cancellation neither proves effect-free execution +nor grants replay, recovery, result adoption or business-completion authority. + A quota-failure notice remains pending work. For Antigravity, the original worker may reserve one new quota-continuation Turn only after two independent reads of the same idle native trajectory prove its latest invocation contains exactly the 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 a84dadcc21..b4ba02b332 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 @@ -1307,6 +1307,11 @@ reaction 清理未完成时，也只恢复原结果 proof，不能再次发送 ### Core 失败后的原生结果恢复 +原生终止错误仍是失败。只有完整的当前调用已空闲、所有已观察步骤均结束，且 +最后错误步骤为 done，适配器才先撤下该调用的取消目标，再返回原结构化错误。 +仍在活动或状态未知的步骤保留既有有界取消和权限处理。避免第二次取消不证明 +没有执行效果，也不授予重放、恢复、结果采用或业务完成资格。 + 额度耗尽的失败提示仍是待办。对于 Antigravity，只在原 worker 两次独立读回同一 空闲轨迹，证明最新一次调用恰好是原始输入和结构化 `QUOTA_EXHAUSTED` 错误、 没有任何模型输出或工具步骤时，现有队列才可预约一次新的 quota 续办 Turn。 diff --git a/loopx/capabilities/native_chat/antigravity_adapter.py b/loopx/capabilities/native_chat/antigravity_adapter.py index f8fdf7670f..7940337cd2 100644 --- a/loopx/capabilities/native_chat/antigravity_adapter.py +++ b/loopx/capabilities/native_chat/antigravity_adapter.py @@ -1186,7 +1186,9 @@ def _run_turn( or (step.get("plannerResponse") or {}).get("response"), (step.get("plannerResponse") or {}).get("thinking"), (step.get("runCommand") or {}).get("combinedOutput"), - (step.get("runCommand") or {}).get("exitCode")) for index, step in current_steps] + (step.get("runCommand") or {}).get("exitCode")) + for index, step in current_steps + if step.get("type") != "CORTEX_STEP_TYPE_ERROR_MESSAGE"] now = time.monotonic() if progress and progress != previous_progress: previous_progress = progress @@ -1204,6 +1206,11 @@ def _run_turn( error = CodexChatAgentError(summary or "Antigravity cascade step error", gate=None, error_code="antigravity_step_error") error.native_error_info = detail + if (final.get("status") == "CORTEX_STEP_STATUS_DONE" + and all(step.get("status") in _SETTLED_STEP_STATUSES for _, step in current_steps)): + # Retire only this settled cancellation target. The + # original failure and possible effects remain unknown. + self._active_turn_id = None raise error planner = final.get("plannerResponse") or {} if (final.get("status") == "CORTEX_STEP_STATUS_CANCELED" diff --git a/tests/test_antigravity_steward_adapter.py b/tests/test_antigravity_steward_adapter.py index 0835b00924..ffb9c81a8e 100644 --- a/tests/test_antigravity_steward_adapter.py +++ b/tests/test_antigravity_steward_adapter.py @@ -1427,12 +1427,40 @@ def test_native_terminal_error_preserves_structured_cause(tmp_path, language_ser detail = {"shortError": "stream reading error: unexpected EOF", "userErrorMessage": "Network interrupted", "errorCode": 2} frame["steps"][-1] = {"type": "CORTEX_STEP_TYPE_ERROR_MESSAGE", "status": "CORTEX_STEP_STATUS_DONE", "metadata": {"sourceTrajectoryStepInfo": {"stepIndex": 1}}, "errorMessage": {"error": detail}} + language_server["cancel_failure"] = True _native_clock(monkeypatch, language_server, [frame]) with pytest.raises(CodexChatAgentError) as raised: adapter_for(tmp_path).start_turn("original task") assert raised.value.error_code == "antigravity_step_error" assert str(raised.value) == "Network interrupted" assert raised.value.native_error_info == detail + assert not language_server["cancelled"].is_set() + assert not any(method == "CancelCascadeInvocation" for method, _ in language_server["calls"]) + + +def test_repeated_native_error_messages_do_not_reset_no_progress_deadline(tmp_path, language_server, monkeypatch): + from loopx.chat_agent import CodexChatAgentError + user = {"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_DONE", + "metadata": {"sourceTrajectoryStepInfo": {"stepIndex": 0}}} + frames = [] + steps = [user] + for idx, at in enumerate((1.0, 120.0, 240.0, 305.0), start=1): + steps = [ + *steps, + { + "type": "CORTEX_STEP_TYPE_ERROR_MESSAGE", + "status": "CORTEX_STEP_STATUS_DONE", + "metadata": {"sourceTrajectoryStepInfo": {"stepIndex": idx}}, + "errorMessage": {"error": {"shortError": f"API error (attempt {idx}): EOF"}}, + }, + ] + frames.append({"at": at, "steps": steps, "status": "CASCADE_RUN_STATUS_RUNNING"}) + _native_clock(monkeypatch, language_server, frames) + with pytest.raises(CodexChatAgentError) as raised: + adapter_for(tmp_path).start_turn("original task") + assert raised.value.error_code == "response_timeout" + assert language_server["now"] == 305.0 + assert language_server["cancelled"].is_set() def test_settled_intermediate_tool_error_or_cancellation_does_not_abort_completed_turn( diff --git a/tests/test_steward_interrupted_answer.py b/tests/test_steward_interrupted_answer.py index 708a7874d6..38d5a47ed9 100644 --- a/tests/test_steward_interrupted_answer.py +++ b/tests/test_steward_interrupted_answer.py @@ -16,6 +16,90 @@ from loopx.extensions.lark.conversation_identity import identity_ref +@pytest.mark.parametrize("prior_status, error_status, cancellation_needed", [ + ("CORTEX_STEP_STATUS_DONE", "CORTEX_STEP_STATUS_DONE", False), + ("CORTEX_STEP_STATUS_ERROR", "CORTEX_STEP_STATUS_DONE", False), + ("CORTEX_STEP_STATUS_CANCELED", "CORTEX_STEP_STATUS_DONE", False), + ("CORTEX_STEP_STATUS_CLEARED", "CORTEX_STEP_STATUS_DONE", False), + ("CORTEX_STEP_STATUS_RUNNING", "CORTEX_STEP_STATUS_DONE", True), + ("CORTEX_STEP_STATUS_GENERATING", "CORTEX_STEP_STATUS_DONE", True), + ("CORTEX_STEP_STATUS_WAITING", "CORTEX_STEP_STATUS_DONE", True), + (None, "CORTEX_STEP_STATUS_DONE", True), + ("CORTEX_STEP_STATUS_DONE", "CORTEX_STEP_STATUS_WAITING", True), + ("CORTEX_STEP_STATUS_DONE", None, True), +]) +def test_idle_native_error_keeps_only_unsettled_work_cancellable( + tmp_path, language_server, monkeypatch, prior_status, error_status, cancellation_needed, +): + from test_antigravity_steward_adapter import _native_clock, _native_frames, adapter_for + from loopx.chat_agent import CodexChatAgentError + + frame = _native_frames()[-1] + detail = {"shortError": "stream interrupted", "userErrorMessage": "Network interrupted", "errorCode": 2} + frame["steps"] = [frame["steps"][0], + {"type": "CORTEX_STEP_TYPE_RUN_COMMAND", "status": prior_status, + "metadata": {"sourceTrajectoryStepInfo": {"stepIndex": 1}}, + "runCommand": {"commandLine": "synthetic inspection", "exitCode": 0}}, + {"type": "CORTEX_STEP_TYPE_ERROR_MESSAGE", "status": error_status, + "metadata": {"sourceTrajectoryStepInfo": {"stepIndex": 2}}, + "errorMessage": {"error": detail}}, + ] + _native_clock(monkeypatch, language_server, [frame]) + language_server["cancel_failure"] = True + adapter = adapter_for(tmp_path) + events = [] + with pytest.raises(CodexChatAgentError) as raised: + adapter.start_turn("Inspect existing state", lambda kind, data: events.append((kind, data))) + + error = raised.value + waiting_for_permission = "CORTEX_STEP_STATUS_WAITING" in {prior_status, error_status} + if waiting_for_permission: + assert error.error_code == "host_tool_gate" and error.gate is not None + else: + assert error.error_code == "antigravity_step_error" and error.native_error_info == detail + assert str(error).startswith("Network interrupted") + methods = [method for method, _ in language_server["calls"]] + assert methods.count("SendUserCascadeMessage") == 1 + assert methods.count("CancelCascadeInvocation") == int(cancellation_needed) + assert ("cancellation remains unconfirmed" in str(error)) is cancellation_needed + assert not any(kind in {"turn.native_result", "answer.final"} for kind, _ in events) + + +def test_settled_native_error_releases_core_for_independent_input(tmp_path, language_server): + from loopx.chat_runtime import ChatRuntimeController + from loopx.chat_store import ChatSessionStore + + store = ChatSessionStore(tmp_path / "runtime") + controller = ChatRuntimeController(store=store, codex_bin="unused-codex") + session, _ = controller.open_session(goal_id="synthetic-goal", agent_id="antigravity", + work_dir=tmp_path, objective="Synthetic conversation", mode="new", channel_id="synthetic-argon-chat") + sid = session["session_id"] + language_server.update(quota_error_indices=[0], cancel_failure=True) + try: + original, _ = controller.submit_turn(session_id=sid, client_turn_id="original-failure", + message="Original request", work_dir=tmp_path, objective="Synthetic conversation") + failed = controller.wait_for_turn(session_id=sid, turn_id=original["turn_id"], timeout_sec=8) + assert failed["status"] == "failed" and failed["error_code"] == "antigravity_step_error" + assert "cancellation remains unconfirmed" not in failed["error"] + assert store.load_session(sid)["active_turn_id"] is None + retained = store._turn_path(sid, original["turn_id"]).read_bytes() + + language_server["quota_error_indices"] = [] + later, _ = controller.submit_turn(session_id=sid, client_turn_id="later-independent", + message="A separate question", work_dir=tmp_path, objective="Synthetic conversation") + completed = controller.wait_for_turn(session_id=sid, turn_id=later["turn_id"], timeout_sec=8) + assert completed["status"] == "completed" and completed["response"]["message"] == "Argon answer 2" + assert store._turn_path(sid, original["turn_id"]).read_bytes() == retained + assert [m["text"] for m in store.messages(sid) if m["role"] == "user"] == [ + "Original request", "A separate question"] + methods = [method for method, _ in language_server["calls"]] + assert methods.count("SendUserCascadeMessage") == 2 + assert "CancelCascadeInvocation" not in methods + assert not any(e["kind"] == "answer.final" for e in store.events_after(sid, original["turn_id"], None)) + finally: + controller.close_session(sid) + + @pytest.mark.parametrize("fault", [None, "stop", "source-changed", "unknown-native", "no-capacity", "native-changed", "review-action", "post-enqueue-crash", "image-view", "checkpoint-live", "checkpoint-cold", "checkpoint-before-review", "checkpoint-before-review-account-change",