fix(chat): distinguish settled command failures during unfinished-work recovery

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md

@@ -1830,15 +1830,22 @@ The existing original-source reply worker may reserve one continuation after
 60 seconds. Core revalidates the original owner/App/source, unchanged message,
 workspace grant and stop/correction state. Native readback must independently
 show that the exact original Turn completed, its full item inventory is stable,
-every effect item completed, no background work remains, and the same final
+every effect item is settled, no background work remains, and the same final
 answer explicitly contains that report. Historical work may continue only when
 every later native Turn correlates to a settled canonical Core Turn and its full
 item inventory is independently stable with no unknown or unfinished effects.
 The proof pins a digest of these inventories; a later equivalent source submission,
-foreign native input, unknown item, failed/declined/live effect, edited source or
+foreign native input, unknown item, unsettled/declined/live effect, edited source or
 revoked grant remains a hold. Native loss remains a hold: a cold client cannot
 certify another process's background effects by resuming the thread.
 
+A failed native command can be settled without having succeeded: the provider
+must retain its ordinary nonzero process exit, ordered start/end timestamps and
+output. Missing exits, timeout/signal exits and failed file-change or MCP effects
+remain holds. This distinction does not certify a business outcome or authorize
+repeating the failed command; the worker still verifies existing artifacts and
+performs only the remaining work under the original settled report and grant.
+
 The original Session queue and deterministic `work-1` reservation own execution;
 restart adopts an already committed Turn instead of dispatching it again. The
 native inventory is checked again immediately before dispatch, and waiting new

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md

@@ -1196,13 +1196,18 @@ provider 当前正文等于已知前文或目标正文。正文改变、效果
 
 既有原消息回复 worker 在 60 秒后才可预留一次续办。Core 重新核对原本人/App/
 来源、未变化的消息、工作区 grant 和停止/纠正状态。独立原生读回必须证明精确原始
-Turn 已完成、完整 item 库存稳定、所有效果 item 均已完成、没有后台工作，且同一
+Turn 已完成、完整 item 库存稳定、所有效果 item 均已结束、没有后台工作，且同一
 最终回答确实包含该报告。历史工作只有在全部后续原生 Turn 均关联已结束的规范
 Core Turn，且各自完整 item 库存独立稳定、没有未知或未结束效果时才可续办。
-证明绑定这些库存的摘要；后续同内容来源提交、外来原生输入、未知 item、失败/
+证明绑定这些库存的摘要；后续同内容来源提交、外来原生输入、未知 item、未结束/
 拒绝/活动效果、来源被编辑或 grant 被撤销均保持等待。原生 owner 丢失仍保持等待：
 冷客户端不能通过恢复 thread 来证明另一个进程没有后台效果。
 
+原生命令失败不等于进程效果未结束：provider 必须保留明确的普通非零退出码、
+有序的开始/结束时间戳及输出。缺少退出码、超时/信号退出，以及失败的文件改动
+或 MCP 效果仍保持等待。这不证明业务成功，也不授权重放失败命令；worker 仍须
+核验现有产物，只在原已结束报告与 grant 下完成剩余工作。
+
 执行复用原 Session 队列和确定性的 `work-1` 预留；重启采用已提交 Turn，不能再次
 派发。实际派发前再次读回原生库存，有新输入排队时让其优先。worker 先读原历史和
 实际产物，只完成仍获授权的剩余工作。新结果使用独立的

loopx/capabilities/native_chat/codex_adapter.py

@@ -270,6 +270,22 @@ def terminal() -> dict[str, Any] | None:
         return None
 
 
+def _settled_work_effect(item: dict[str, Any]) -> bool:
+    """A recorded process exit can settle a command without making it succeed.
+
+    This is native DTO evidence only. The original worker's explicit settled
+    report, live background inventory and Core continuation checks still apply.
+    Missing exits, timeouts/signals and failed non-process effects stay unknown.
+    """
+    if item.get("status") == "completed":
+        return True
+    code, started, ended = item.get("exitCode"), item.get("startedAtMs"), item.get("completedAtMs")
+    return (item.get("type") == "commandExecution" and item.get("status") == "failed"
+            and type(code) is int and 0 < code < 128 and code not in {124, 125}
+            and type(started) is int and type(ended) is int and 0 < started <= ended
+            and isinstance(item.get("aggregatedOutput"), str))
+
+
 def completed_work_readback(session: CodexChatAgentSession, turn_id: str, *,
                             known_turn_ids: set[str] | None = None) -> dict[str, Any] | None:
     """Read stable settled work, including history correlated by its Core owner.
@@ -317,8 +333,8 @@ def turns():
                 elif item.get("phase") != "commentary":
                     return None
             elif kind in {"commandExecution", "fileChange", "mcpToolCall"}:
-                if item.get("status") != "completed":
-                    return None  # Declined/failed/live/unknown effects cannot request writable continuation.
+                if not _settled_work_effect(item):
+                    return None  # Declined/live/unobserved effects cannot request writable continuation.
             elif kind not in {"reasoning", "webSearch", "imageView", "contextCompaction"}:
                 return None
         if (users != 1 or len(finals) != 1 or turns() != inventory
@@ -345,7 +361,7 @@ def turns():
                         "mcpToolCall", "reasoning", "webSearch", "imageView", "contextCompaction"}
                         or item.get("type") == "agentMessage" and item.get("phase") not in {"commentary", "final_answer"}
                         or item.get("type") in {"commandExecution", "fileChange", "mcpToolCall"}
-                        and item.get("status") != "completed"):
+                        and not _settled_work_effect(item)):
                     return None
             later_items[value] = rows
         if (any(list(session._control_items("thread/items/list", {"turnId": value}, deadline)) != rows

tests/test_chat_codex_history.py

@@ -130,7 +130,10 @@ def native_start(**kwargs):
     runtime.close()
 
 
-@pytest.mark.parametrize("fault", [None, "image-view", "known-later", "later-user", "later-failed-effect", "later-moving-items",
+@pytest.mark.parametrize("fault", [None, "image-view", "known-later", "settled-command", "later-settled-command",
+    "command-no-exit", "command-open", "command-invalid-clock", "command-signal", "command-timeout", "command-bool-exit",
+    "command-no-output", "command-bool-clock", "command-zero-exit",
+    "later-user", "later-failed-effect", "later-moving-items",
     "later-unknown", "missing-original", "duplicate-turn", "live", "failed", "background", "unknown", "live-tool",
     "declined", "duplicate", "foreign", "second-user", "no-final", "legacy-prose", "moving-turn", "moving-items"])
 def test_completed_work_readback_requires_stable_correlated_settled_inventory(fault):
@@ -151,7 +154,7 @@ def items(method, params, deadline):
         counts[method] += 1
         if method == "thread/turns/list":
             turns = [{"id": "original-turn", "status": "failed" if fault == "failed" else "completed"}]
-            if fault in {"known-later", "later-user", "later-failed-effect", "later-moving-items", "later-unknown", "live"} or fault == "moving-turn" and counts[method] > 1:
+            if fault in {"known-later", "later-settled-command", "later-user", "later-failed-effect", "later-moving-items", "later-unknown", "live"} or fault == "moving-turn" and counts[method] > 1:
                 turns.insert(0, {"id": "later", "status": "inProgress" if fault == "live" else "completed"})
             if fault == "missing-original":
                 turns = [{"id": "later", "status": "completed"}]
@@ -163,11 +166,35 @@ def items(method, params, deadline):
             rows = [{"turnId": "later", "item": {"id": "later-user", "type": "userMessage"}},
                 {"turnId": "later", "item": {"id": "later-effect", "type": "fileChange",
                     "status": "failed" if fault == "later-failed-effect" else "completed"}}]
+            if fault == "later-settled-command":
+                rows[1]["item"].update(type="commandExecution", status="failed", exitCode=1,
+                    startedAtMs=1000, completedAtMs=1001, aggregatedOutput="Process exited with code 1.")
             if fault == "later-unknown" or fault == "later-moving-items" and later_reads > 1:
                 rows.append({"turnId": "later", "item": {"id": "unknown", "type": "futureEffect"}})
             return iter(rows)
         assert params == {"turnId": "original-turn"}
         effect = {"id": "effect", "type": "fileChange", "status": "completed"}
+        if fault == "settled-command" or str(fault).startswith("command-"):
+            effect.update(type="commandExecution", status="failed", exitCode=1,
+                startedAtMs=1000, completedAtMs=1001, aggregatedOutput="Process exited with code 1.")
+            if fault == "command-no-exit":
+                effect.pop("exitCode")
+            elif fault == "command-open":
+                effect["completedAtMs"] = None
+            elif fault == "command-invalid-clock":
+                effect["completedAtMs"] = 999
+            elif fault == "command-signal":
+                effect["exitCode"] = 137
+            elif fault == "command-timeout":
+                effect["exitCode"] = 124
+            elif fault == "command-bool-exit":
+                effect["exitCode"] = True
+            elif fault == "command-no-output":
+                effect.pop("aggregatedOutput")
+            elif fault == "command-bool-clock":
+                effect["startedAtMs"] = True
+            elif fault == "command-zero-exit":
+                effect["exitCode"] = 0
         if fault == "unknown":
             effect["type"] = "futureEffect"
         if fault == "image-view":
@@ -188,13 +215,13 @@ def items(method, params, deadline):
     native = SimpleNamespace(thread_id="native", work_dir=Path("/synthetic/notes"), response_timeout_sec=.1, _control_items=items,
         _background_terminals=lambda deadline: [{"processId": "live"}] if fault == "background" else [])
     proof = completed_work_readback(native, "original-turn", known_turn_ids={"later"}
-        if fault in {"known-later", "later-failed-effect", "later-moving-items", "later-unknown"} else None)
-    if fault in {None, "image-view", "known-later"}:
+        if fault in {"known-later", "later-settled-command", "later-failed-effect", "later-moving-items", "later-unknown"} else None)
+    if fault in {None, "image-view", "known-later", "settled-command", "later-settled-command"}:
         digest = proof.pop("native_inventory_sha256")
         assert len(digest) == 64 and set(digest) <= set("0123456789abcdef")
         assert proof == {"thread_id": "native", "turn_id": "original-turn", "completed_work": True,
                          "all_effects_settled": True, "unfinished_work": work}
-        assert counts == {"thread/turns/list": 3, "thread/items/list": 4 if fault == "known-later" else 2}
+        assert counts == {"thread/turns/list": 3, "thread/items/list": 4 if fault in {"known-later", "later-settled-command"} else 2}
     else:
         assert proof is None
     assert set(calls) <= {"thread/turns/list", "thread/items/list"}

tests/test_lark_private_conversations.py

@@ -1705,6 +1705,7 @@ def counted_revision():
 
 @pytest.mark.parametrize("fault", [None, "changed-source", "revoked", "stop", "unknown-native",
     "background", "later-user", "completed-later", "equivalent-later", "later-failed-effect",
+    "settled-command", "later-settled-command", "unsettled-command",
     "reservation-crash", "post-enqueue-crash", "queued-newer", "queued-items", "queued-input", "private-path", "still-incomplete", "cold-native", "dead-native", "foreign-native", "failed-effect"])
 def test_original_worker_continues_explicit_unfinished_project_work_once_after_restart(ordinary, monkeypatch, fault):  # noqa: F811
     """Reply-worker restart retains native ownership; native loss stays fenced."""
@@ -1779,7 +1780,7 @@ def test_original_worker_continues_explicit_unfinished_project_work_once_after_r
     original_turn_bytes = store._turn_path(original["session_id"], original["turn_id"]).read_bytes()
     session_before = store.load_session(original["session_id"])
     Path(session_before["codex_home"]).mkdir(parents=True, exist_ok=True)
-    if fault in {"completed-later", "equivalent-later", "later-failed-effect"}:
+    if fault in {"completed-later", "equivalent-later", "later-failed-effect", "later-settled-command"}:
         message = event["content"] if fault == "equivalent-later" else "Newer follow-up: a different completed request."
         later_event = provider.event("notes-app", "later-correlated", message)
         transport.admit("notes-app", later_event)
@@ -1788,11 +1789,22 @@ def test_original_worker_continues_explicit_unfinished_project_work_once_after_r
         later_turn = runtime.wait_for_turn(session_id=later["session_id"], turn_id=later["turn_id"], timeout_sec=10)
         assert later_turn["status"] == "completed"
         transport.reconcile_request(transport.root / (later_ref + ".json"))
-        if fault == "later-failed-effect":
+        if fault in {"later-failed-effect", "later-settled-command"}:
             snapshot = json.loads(inventory.read_text())
-            next(row["item"] for row in snapshot["items"] if row["turnId"] == later_turn["upstream_turn_id"]
-                and row["item"]["type"] == "fileChange")["status"] = "failed"
+            effect = next(row["item"] for row in snapshot["items"] if row["turnId"] == later_turn["upstream_turn_id"]
+                and row["item"]["type"] == "fileChange")
+            effect["status"] = "failed"
+            if fault == "later-settled-command":
+                effect.update(type="commandExecution", exitCode=1, startedAtMs=1000,
+                    completedAtMs=1001, aggregatedOutput="Process exited with code 1.")
             inventory.write_text(json.dumps(snapshot))
+    if fault in {"settled-command", "unsettled-command"}:
+        snapshot = json.loads(inventory.read_text())
+        snapshot["items"][1]["item"].update(type="commandExecution", status="failed", exitCode=1,
+            startedAtMs=1000, completedAtMs=1001, aggregatedOutput="Process exited with code 1.")
+        if fault == "unsettled-command":
+            snapshot["items"][1]["item"]["exitCode"] = None
+        inventory.write_text(json.dumps(snapshot))
     if fault == "cold-native":
         runtime.adapters.pop(original["session_id"]).close_session()
     elif fault == "dead-native":
@@ -1855,7 +1867,7 @@ def crash(**kwargs):
         selected = worker()
     selected.reconcile()
     continued = selected.core.read_request(ref)
-    if fault in {"changed-source", "revoked", "stop", "unknown-native", "background", "later-user", "cold-native", "dead-native", "foreign-native", "failed-effect", "equivalent-later", "later-failed-effect"}:
+    if fault in {"changed-source", "revoked", "stop", "unknown-native", "background", "later-user", "cold-native", "dead-native", "foreign-native", "failed-effect", "equivalent-later", "later-failed-effect", "unsettled-command"}:
         assert continued["turn_id"] == original["turn_id"] and not continued.get("context_recovery")
         assert artifact.read_text() == ("first\nremaining\n" if fault == "equivalent-later" else "first\n")
     else:
@@ -1882,7 +1894,7 @@ def crash(**kwargs):
         assert after["project_context"] == session_before["project_context"]
         assert after["upstream_thread_id"] == session_before["upstream_thread_id"]
         starts = [json.loads(line)["params"] for line in capture.read_text().splitlines() if json.loads(line).get("method") == "turn/start"]
-        assert len(starts) == (1 if fault in {"queued-newer", "queued-items"} else 3 if fault == "completed-later" else 2)
+        assert len(starts) == (1 if fault in {"queued-newer", "queued-items"} else 3 if fault in {"completed-later", "later-settled-command"} else 2)
         if not interrupted:
             assert "Do only the remaining authorized work" in starts[-1]["input"][0]["text"]
         if fault == "queued-input":