From e6c3f209c42bd5bfcb3774e1a9c15aedb95860fd Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 6 Nov 2026 22:31:30 +0800 Subject: [PATCH 1/2] fix(chat): continue settled work behind later original requests Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../capabilities/native_chat/codex_adapter.py | 46 ++++++++++++++++--- .../native_chat/external_conversations.py | 18 +++++++- .../collaboration/conversation_binding.ts | 6 ++- .../conversation_binding.test.ts | 6 ++- tests/test_chat_codex_history.py | 36 +++++++++++---- tests/test_lark_private_conversations.py | 43 ++++++++++++----- 6 files changed, 124 insertions(+), 31 deletions(-) diff --git a/loopx/capabilities/native_chat/codex_adapter.py b/loopx/capabilities/native_chat/codex_adapter.py index 2b82ed21aa..822ffe75cc 100644 --- a/loopx/capabilities/native_chat/codex_adapter.py +++ b/loopx/capabilities/native_chat/codex_adapter.py @@ -6,6 +6,7 @@ from __future__ import annotations from dataclasses import dataclass +import hashlib import json import logging import os @@ -269,8 +270,9 @@ def terminal() -> dict[str, Any] | None: return None -def completed_work_readback(session: CodexChatAgentSession, turn_id: str) -> dict[str, Any] | None: - """Read a latest, stable completed inventory with an explicit unfinished report. +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. This proves settled native items, not the safety of repeating their effects. Core owns the original source, grant, bounded continuation and stop checks. @@ -286,12 +288,16 @@ def turns(): inventory = turns() ids = [item.get("id") for item in inventory] - if (not ids or ids[0] != turn_id or inventory[0].get("status") != "completed" + target = next((item for item in inventory if item.get("id") == turn_id), None) + if (target is None or target.get("status") != "completed" 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 inventory) or session._background_terminals(deadline)): return None + later = ids[:ids.index(turn_id)] + if any(value not in (known_turn_ids or set()) for value in later): + return None # A foreign native input may cancel or supersede the old request. items = list(session._control_items("thread/items/list", {"turnId": turn_id}, deadline)) seen: set[str] = set() users = 0 @@ -322,8 +328,35 @@ def turns(): unfinished = parse_agent_response(finals[0], protected_paths=[session.work_dir]).get("unfinished_work") if unfinished is None: return None + later_items = {} + for value in later: + rows = list(session._control_items("thread/items/list", {"turnId": value}, deadline)) + if any(not isinstance(row, dict) or not isinstance(row.get("item"), dict) for row in rows): + return None + item_ids = [row.get("item", {}).get("id") for row in rows] + if (not rows or any(row.get("turnId") != value for row in rows) + or any(not isinstance(item_id, str) or not item_id for item_id in item_ids) + or len(set(item_ids)) != len(item_ids) + or sum(row["item"].get("type") == "userMessage" for row in rows) != 1): + return None + for row in rows: + item = row["item"] + if (item.get("type") not in {"userMessage", "agentMessage", "commandExecution", "fileChange", + "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"): + return None + later_items[value] = rows + if (any(list(session._control_items("thread/items/list", {"turnId": value}, deadline)) != rows + for value, rows in later_items.items()) + or turns() != inventory or session._background_terminals(deadline)): + return None + digest = hashlib.sha256(json.dumps([inventory, items, later_items], + sort_keys=True, separators=(",", ":")).encode()).hexdigest() return {"thread_id": session.thread_id, "turn_id": turn_id, - "completed_work": True, "all_effects_settled": True, "unfinished_work": unfinished} + "completed_work": True, "all_effects_settled": True, "unfinished_work": unfinished, + "native_inventory_sha256": digest} except (CodexChatAgentError, OSError, TypeError, ValueError): return None @@ -384,7 +417,8 @@ class CodexAppServerAdapter: session: CodexChatAgentSession goal_driver: CodexGoalDriver | None = None - def completed_work_observation(self, *, thread_ids: list[str], turn_id: str) -> dict[str, Any] | None: + def completed_work_observation(self, *, thread_ids: list[str], turn_id: str, + known_turn_ids: set[str] | None = None) -> dict[str, Any] | None: """Read live effects through the adapter that owns the original thread. A cold app-server can read persisted items, but cannot inventory another @@ -393,7 +427,7 @@ def completed_work_observation(self, *, thread_ids: list[str], turn_id: str) -> """ if self.upstream_thread_id not in thread_ids or not self.healthcheck(): return None - return completed_work_readback(self.session, turn_id) + return completed_work_readback(self.session, turn_id, known_turn_ids=known_turn_ids) @staticmethod def failed_request_observation(*, codex_bin: str, codex_home: Path, diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 897d8ef577..fae7278b88 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -142,6 +142,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *, later_equivalent = None image_failure = turn.get("error_code") == "runtime_error" and bool(row.get("attachments")) runtime_failure = turn.get("error_code") in {"model_auth_unavailable", "resume_failed"} + unfinished = turn.get("status") == "completed" and bool((turn.get("response") or {}).get("unfinished_work")) if turn.get("error_code") == "manager_authorization_unavailable" or image_failure or runtime_failure: # Collect canonical facts only. The typed planner owns the exact # pre-dispatch sequence, retry budget and authority decision. @@ -152,6 +153,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *, {"event_id": item["event_id"], "kind": item["kind"], **{key: item.get("payload", {})[key] for key in ("phase", "schema_version", "error_code", "delivery_mode", "message") if key in item.get("payload", {})}} for item in events]} + if preparation is not None or unfinished: created = datetime.fromisoformat(turn["created_at"]) later_equivalent = any(item.get("role") == "user" and item.get("turn_id") != turn["turn_id"] and item.get("text") == row.get("message") and item.get("origin") in {None, "lark", "web"} @@ -430,7 +432,8 @@ def source_requests() -> list[dict[str, Any]]: adapter = self.controller.adapters.get(row["session_id"]) if isinstance(adapter, CodexAppServerAdapter): proof = adapter.completed_work_observation( - thread_ids=plan["native_thread_ids"], turn_id=turn["upstream_turn_id"]) + thread_ids=plan["native_thread_ids"], turn_id=turn["upstream_turn_id"], + known_turn_ids=self._settled_native_turn_ids(session["session_id"])) else: proof = CodexAppServerAdapter.failed_request_observation( codex_bin=self.controller.codex_bin, codex_home=Path(session["codex_home"]), @@ -889,6 +892,16 @@ def validate_metadata_review_response(self, session: dict[str, Any], turn: dict[ # append each actual receipt/failure before the original worker replies. return {**response, "message": "原委派格式已复核，以下为实际提交回执；接收者结论仍待回传。"} + def _settled_native_turn_ids(self, session_id: str) -> set[str]: + """Correlate native history through canonical Core Turns, without adopting it.""" + store = self.controller.store + ids = {item["turn_id"] for item in store.messages(session_id, strict=True) + if item.get("role") == "user" and item.get("turn_id")} + turns = [store.load_turn(session_id, value) for value in ids] + return {item["upstream_turn_id"] for item in turns if item + and item.get("status") in {"completed", "failed"} + and isinstance(item.get("upstream_turn_id"), str) and item["upstream_turn_id"]} + def prepare_continuation_input(self, session: dict[str, Any], turn: dict[str, Any], native: Any, message: str) -> str: """Recheck reserved unfinished work at dispatch; keep the stored input intact.""" @@ -900,7 +913,8 @@ def prepare_continuation_input(self, session: dict[str, Any], turn: dict[str, An return message proof = continuation.get("native_observation") or {} if (self.controller.store.queued_turns(session["session_id"]) - or completed_work_readback(native, str(proof.get("turn_id") or "")) != proof): + or completed_work_readback(native, str(proof.get("turn_id") or ""), + known_turn_ids=self._settled_native_turn_ids(session["session_id"])) != proof): raise CodexChatAgentError( "The original unfinished-work inventory changed or newer input is waiting; continuation stopped.", error_code="host_gate", gate=None, diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index b240392e57..4aa290185e 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -1033,7 +1033,7 @@ function nativeRecoveryCapacity(recovery: JsonObject, row: JsonObject, current: /** A completed model Turn can explicitly report unfinished business. The * proposal alone grants nothing: only the original worker may reserve one - * continuation after stable latest-native/source/grant/stop checks. + * continuation after stable correlated-native/source/grant/stop checks. */ function unfinishedProjectWork(result: JsonObject): JsonObject | null { const raw = result.unfinished_work; @@ -1078,7 +1078,8 @@ function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: Json admitted_turn_id: sessionIdentity(admitted.turn_id, "admitted unfinished work"), native_thread_ids: [], not_before: previous.not_before}; } - if (current.active_turn_id != null || current.status !== "ready") return no; + if (current.active_turn_id != null || current.status !== "ready" + || recovery.queued_input_present !== false || recovery.later_equivalent_input_present !== false) return no; const now = Date.parse(String(recovery.observed_at)), ended = Date.parse(String(completed.completed_at)); if (!Number.isFinite(now) || !Number.isFinite(ended) || now < ended + 60_000) return no; const native = sessionIdentity(current.upstream_thread_id, "original native thread"); @@ -1088,6 +1089,7 @@ function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: Json return {...planned, recovery_eligible: recovery.source_verified === true && observed.thread_id === native && observed.turn_id === completed.upstream_turn_id && observed.completed_work === true && observed.all_effects_settled === true + && typeof observed.native_inventory_sha256 === "string" && BARE_SHA256_PATTERN.test(observed.native_inventory_sha256) && matchesWork(observed.unfinished_work)}; } diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index d6e2e3734c..3e56cdaeb4 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/tests/control_plane_ts/conversation_binding.test.ts @@ -799,7 +799,7 @@ for (const endpoint of ["codex", "antigravity", "kiro-cli"]) { }); } -test("unfinished completed work requires typed intent and latest settled native/source proof, once", () => { +test("unfinished completed work requires typed intent and pinned settled native/source proof, once", () => { const configured = planConversationBinding(request).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, @@ -812,18 +812,22 @@ test("unfinished completed work requires typed intent and latest settled native/ upstream_thread_id: "native", active_turn_id: null, project_context: context, channel_id: `project.external.${row.binding_id}.${source.source_ref}`}; const native = {thread_id: "native", turn_id: "native-partial", completed_work: true, + native_inventory_sha256: "a".repeat(64), all_effects_settled: true, unfinished_work: {effects_settled: true, dependency: work.dependency, remaining: work.remaining}}; const packet = {request: {request_ref: "f".repeat(24), command: null, status: "accepted", delivery_verified: true, message: "Original request", origin: "lark", binding_id: row.binding_id, session_id: "original", turn_id: "partial", source}, binding: row, current_session: session, context_recovery: {version: 21, failed_turn: completed, observed_at: "2026-01-01T00:10:00Z", steering_observed: false, stop_observed: false, attachments_present: false, + queued_input_present: false, later_equivalent_input_present: false, source_verified: true, native_observation: native}}; const plan = planBoundConversationRequest(packet); assert.equal(plan.recovery_eligible, true); assert.equal(plan.work_attempt, 1); assert.ok(String(plan.client_turn_id).endsWith("-work-1")); for (const change of [{source_verified: false}, {steering_observed: true}, {stop_observed: true}, + {queued_input_present: true}, {later_equivalent_input_present: true}, {later_equivalent_input_present: null}, {attachments_present: true}, {observed_at: "2026-01-01T00:00:59Z"}, + {native_observation: {...native, native_inventory_sha256: null}}, {native_observation: {...native, all_effects_settled: false}}, {native_observation: {...native, thread_id: "foreign"}}, {native_observation: {...native, unfinished_work: {...work, remaining: "Another task"}}}]) { diff --git a/tests/test_chat_codex_history.py b/tests/test_chat_codex_history.py index 263c4cdab8..f661eab270 100644 --- a/tests/test_chat_codex_history.py +++ b/tests/test_chat_codex_history.py @@ -11,7 +11,7 @@ from test_native_steward_private import steward # noqa: F401 -def test_source_revocation_during_history_compaction_preserves_original_audience(steward, monkeypatch): +def test_source_revocation_during_history_compaction_preserves_original_audience(steward, monkeypatch): # noqa: F811 import threading from concurrent.futures import ThreadPoolExecutor @@ -50,7 +50,7 @@ def native_start(**kwargs): @pytest.mark.parametrize("action", ["stop", "close"]) -def test_native_history_preparation_observes_user_stop_before_its_adapter_lock(conversation, monkeypatch, action): +def test_native_history_preparation_observes_user_stop_before_its_adapter_lock(conversation, monkeypatch, action): # noqa: F811 import threading from concurrent.futures import ThreadPoolExecutor @@ -102,7 +102,7 @@ def restore(): @pytest.mark.parametrize("change", ["close", "settings", "requested-model"]) -def test_history_guard_rechecks_after_native_start_before_publishing(conversation, monkeypatch, change): +def test_history_guard_rechecks_after_native_start_before_publishing(conversation, monkeypatch, change): # noqa: F811 from test_chat_manager_model_adoption import Adapter root, store, runtime, session, _ = conversation sid = session["session_id"] @@ -130,9 +130,10 @@ def native_start(**kwargs): runtime.close() -@pytest.mark.parametrize("fault", [None, "image-view", "later-user", "live", "failed", "background", "unknown", "live-tool", +@pytest.mark.parametrize("fault", [None, "image-view", "known-later", "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_latest_stable_explicit_settled_inventory(fault): +def test_completed_work_readback_requires_stable_correlated_settled_inventory(fault): from loopx.chat import CHAT_REVIEW_OPEN_TAG, CHAT_REVIEW_CLOSE_TAG from loopx.capabilities.native_chat.codex_adapter import completed_work_readback import json @@ -142,15 +143,29 @@ def test_completed_work_readback_requires_latest_stable_explicit_settled_invento "message": "", "unfinished_work": work}) + CHAT_REVIEW_CLOSE_TAG calls = [] counts = {"thread/turns/list": 0, "thread/items/list": 0} + later_reads = 0 def items(method, params, deadline): + nonlocal later_reads calls.append(method) counts[method] += 1 if method == "thread/turns/list": turns = [{"id": "original-turn", "status": "failed" if fault == "failed" else "completed"}] - if fault in {"later-user", "live"} or fault == "moving-turn" and counts[method] > 1: + if fault in {"known-later", "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"}] + if fault == "duplicate-turn": + turns.append(turns[0]) return iter(turns) + if params == {"turnId": "later"}: + later_reads += 1 + 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-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 == "unknown": @@ -172,11 +187,14 @@ 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") - if fault in {None, "image-view"}: + 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"}: + 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": 2, "thread/items/list": 2} + assert counts == {"thread/turns/list": 3, "thread/items/list": 4 if fault == "known-later" else 2} else: assert proof is None assert set(calls) <= {"thread/turns/list", "thread/items/list"} diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 3a75a32ef8..91740c42b9 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -1704,7 +1704,8 @@ def counted_revision(): @pytest.mark.parametrize("fault", [None, "changed-source", "revoked", "stop", "unknown-native", - "background", "later-user", "reservation-crash", "post-enqueue-crash", "queued-newer", "queued-input", "private-path", "still-incomplete", "cold-native", "dead-native", "foreign-native", "failed-effect"]) + "background", "later-user", "completed-later", "equivalent-later", "later-failed-effect", + "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.""" from datetime import datetime, timedelta, timezone @@ -1746,13 +1747,15 @@ def test_original_worker_continues_explicit_unfinished_project_work_once_after_r {{"turnId": active_turn, "item": {{"id": "user", "type": "userMessage"}}}}, {{"turnId": active_turn, "item": {{"id": "effect", "type": "fileChange", "status": "completed"}}}}, {{"turnId": active_turn, "item": {{"id": "view", "type": "imageView", "path": str(artifact)}}}}, - {{"turnId": active_turn, "item": {{"id": "answer", "type": "agentMessage", "phase": "final_answer", "text": response}}}}]}})) + {{"turnId": active_turn, "item": {{"id": "answer", "type": "agentMessage", "phase": "final_answer", "text": response}}}}, + *previous.get("items", [])]}})) ''' + source[end:] start = source.index(' elif method == "thread/turns/list":') end = source.index(' else:\n result = {}', start) source = source[:start] + f''' elif method in {{"thread/turns/list", "thread/items/list"}}: snapshot = json.load(open({str(inventory)!r})) - result = {{"data": snapshot["turns" if method == "thread/turns/list" else "items"]}} + result = {{"data": snapshot["turns"] if method == "thread/turns/list" else + [row for row in snapshot["items"] if row["turnId"] == request["params"]["turnId"]]}} elif method == "thread/backgroundTerminals/list": if not loaded: print(json.dumps({{"id": request["id"], "error": {{"code": -32600, "message": "thread not found"}}}}), flush=True) @@ -1776,6 +1779,20 @@ 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"}: + 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) + later_ref = identity_ref(binding["provider_ref"], later_event["message_id"]) + later = transport.core.read_request(later_ref) + 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": + 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" + inventory.write_text(json.dumps(snapshot)) if fault == "cold-native": runtime.adapters.pop(original["session_id"]).close_session() elif fault == "dead-native": @@ -1808,15 +1825,15 @@ def now(cls, tz=None): def worker(): return LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent, runner=provider, cli_bin="lark-cli") selected = worker() - if fault in {"reservation-crash", "post-enqueue-crash", "queued-newer", "queued-input"}: + if fault in {"reservation-crash", "post-enqueue-crash", "queued-newer", "queued-items", "queued-input"}: enqueue = runtime.enqueue_turn resume = runtime.resume_session_queue - if fault in {"queued-newer", "queued-input"}: + if fault in {"queued-newer", "queued-items", "queued-input"}: # Hold dispatch before enqueue, rather than relying on a slow # adapter startup to let the newer-input fixture win a race. monkeypatch.setattr(runtime, "resume_session_queue", lambda **kwargs: None) def crash(**kwargs): - if fault in {"post-enqueue-crash", "queued-newer", "queued-input"}: + if fault in {"post-enqueue-crash", "queued-newer", "queued-items", "queued-input"}: enqueue(**kwargs) raise OSError("synthetic continuation admission crash") monkeypatch.setattr(runtime, "enqueue_turn", crash) @@ -1827,21 +1844,25 @@ def crash(**kwargs): snapshot = json.loads(inventory.read_text()) snapshot["turns"].insert(0, {"id": "newer-after-reservation", "status": "completed"}) inventory.write_text(json.dumps(snapshot)) + if fault == "queued-items": + snapshot = json.loads(inventory.read_text()) + snapshot["items"][1]["item"]["output"] = "Changed after admission" + inventory.write_text(json.dumps(snapshot)) if fault == "queued-input": transport.admit("notes-app", provider.event("notes-app", "newer-input", "Newer follow-up: inspect this correction first.")) - if fault in {"queued-newer", "queued-input"}: + if fault in {"queued-newer", "queued-items", "queued-input"}: monkeypatch.setattr(runtime, "resume_session_queue", resume) 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"}: + 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"}: assert continued["turn_id"] == original["turn_id"] and not continued.get("context_recovery") - assert artifact.read_text() == "first\n" + assert artifact.read_text() == ("first\nremaining\n" if fault == "equivalent-later" else "first\n") else: assert continued["turn_id"] != original["turn_id"] assert continued["session_id"] == original["session_id"] assert continued["context_recovery"]["work_attempt"] == 1 - interrupted = fault in {"queued-newer", "queued-input"} + interrupted = fault in {"queued-newer", "queued-items", "queued-input"} assert runtime.wait_for_turn(session_id=continued["session_id"], turn_id=continued["turn_id"], timeout_sec=10)["status"] == ("failed" if interrupted else "completed") if fault == "queued-input": queued = [item for item in selected.core.pending() if item["request_ref"] != ref and item.get("command") is None] @@ -1861,7 +1882,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 == "queued-newer" else 2) + assert len(starts) == (1 if fault in {"queued-newer", "queued-items"} else 3 if fault == "completed-later" else 2) if not interrupted: assert "Do only the remaining authorized work" in starts[-1]["input"][0]["text"] if fault == "queued-input": From dbf8a97753d262e4f1085e235c6dcaf5407082f9 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 6 Nov 2026 22:31:31 +0800 Subject: [PATCH 2/2] docs(chat): define bounded historical unfinished-work recovery Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../rfcs/app-conversation-and-async-inbox-v0.md | 11 ++++++++--- .../rfcs/app-conversation-and-async-inbox-v0.zh-CN.md | 9 ++++++--- 2 files changed, 14 insertions(+), 6 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 a770e12aee..0701303259 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -1829,10 +1829,15 @@ steward, group, attached-host and attachment-bearing requests remain outside it. 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 latest Turn completed, its full item inventory is stable, +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 -answer explicitly contains that report. A later native Turn, unknown item, -failed/declined/live effect, edited source or revoked grant remains a hold. +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 +revoked grant remains a hold. Native loss remains a hold: a cold client cannot +certify another process's background effects by resuming the thread. The original Session queue and deterministic `work-1` reservation own execution; restart adopts an already committed Turn instead of dispatching it again. 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 6f0e710526..19ad89f5c4 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 @@ -1195,10 +1195,13 @@ provider 当前正文等于已知前文或目标正文。正文改变、效果 及携带附件的请求不进入该路径。 既有原消息回复 worker 在 60 秒后才可预留一次续办。Core 重新核对原本人/App/ -来源、未变化的消息、工作区 grant 和停止/纠正状态。独立原生读回必须证明精确最新 +来源、未变化的消息、工作区 grant 和停止/纠正状态。独立原生读回必须证明精确原始 Turn 已完成、完整 item 库存稳定、所有效果 item 均已完成、没有后台工作，且同一 -最终回答确实包含该报告。后续原生 Turn、未知 item、失败/拒绝/活动效果、来源 -被编辑或 grant 被撤销均保持等待。 +最终回答确实包含该报告。历史工作只有在全部后续原生 Turn 均关联已结束的规范 +Core Turn，且各自完整 item 库存独立稳定、没有未知或未结束效果时才可续办。 +证明绑定这些库存的摘要；后续同内容来源提交、外来原生输入、未知 item、失败/ +拒绝/活动效果、来源被编辑或 grant 被撤销均保持等待。原生 owner 丢失仍保持等待： +冷客户端不能通过恢复 thread 来证明另一个进程没有后台效果。 执行复用原 Session 队列和确定性的 `work-1` 预留；重启采用已提交 Turn，不能再次 派发。实际派发前再次读回原生库存，有新输入排队时让其优先。worker 先读原历史和