fix(native-chat): avoid redundant cancel and progress resets on Antigravity step errors

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

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。

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"

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(

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