From ebba631c42568a5f81f7713faac9c096534b5352 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 20 Oct 2026 01:13:21 +0800 Subject: [PATCH] fix(lark): isolate recovery drafts and repair verified legacy notices Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 11 ++ ...p-conversation-and-async-inbox-v0.zh-CN.md | 8 + loopx/extensions/lark/inbox_reply.py | 3 + .../extensions/lark/private_conversations.py | 108 ++++++++--- loopx/extensions/lark/private_progress.py | 58 ++++++ tests/test_lark_private_recovery_progress.py | 176 ++++++++++++++++++ 6 files changed, 336 insertions(+), 28 deletions(-) create mode 100644 tests/test_lark_private_recovery_progress.py 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 a3ab0fafbf..1ea36de886 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -1324,6 +1324,17 @@ original-source result. A restart resumes the deterministic reservation rather than dispatching it twice. Existing provider replies accompany recovery as data so completed work can be retained without repeating its effects. +Continuation admission, progress, cursor and final delivery use a version-scoped +presentation journal. They never reopen the acknowledged original Inbox source +or edit its failure notice as a new draft. The final replaces only the new draft. +For already-returned legacy continuations that reused the old notice, the same +worker may restore that notice and retire subsequent stale drafts using their +exact confirmed edit proofs. This is edit-only presentation repair: current +source/grant, canonical Turn and unchanged prior-or-desired provider body must +verify. Changed bodies, uncertain effects and edit-limit rejections stay pending; +repair neither enqueues work nor appends another result. Acceptance independently +checks actual provider bodies, including rollover and lost acknowledgements. + Qualification covers the actual background scanner, source/grant counterexamples, native inventory reads and crashes before admission, after enqueue and before transport synchronization. Synthetic model/provider journeys and read-only native 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 cf4987cefa..a72264cca6 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 @@ -840,6 +840,14 @@ Turn，确认完整条目清单仅包含原用户输入后，Core 才能为本 确定性预留，不重复派发。已有 provider 回复作为数据随恢复提供，供保留已完成工作、 避免重复效果。 +续办的接收提示、进展、游标与最终回传按版本独立记账，不重新打开原 Inbox 已确认的 +消息，也不把原失败通知改成新草稿；最终结果只替换本次新草稿。对于已经回传、但旧版 +误用原失败通知的续办，原 worker 可凭确切已确认的编辑凭证恢复旧通知，并结束随后 +残留的旧草稿。这仅是消息编辑修复：必须核验当前来源与 grant、原生规范 Turn，以及 +provider 当前正文等于已知前文或目标正文。正文改变、效果不明与编辑次数耗尽继续 +保留待办；不重新排队执行，也不另发结果。验收独立检查 provider 实际正文，覆盖草稿 +续段与回执丢失。 + 资格验证覆盖真实后台扫描器、来源与授权反例、原生条目读取，以及受理前、排队后、 传输同步前的中断。合成模型/provider 通路及只读原生观察不证明真实模型完成， 也不代表更广泛的会话 SLA 已通过。 diff --git a/loopx/extensions/lark/inbox_reply.py b/loopx/extensions/lark/inbox_reply.py index 3f9d74082b..230b2fcb29 100644 --- a/loopx/extensions/lark/inbox_reply.py +++ b/loopx/extensions/lark/inbox_reply.py @@ -991,6 +991,7 @@ def update_lark_inbox_reply( before_send: Callable[[str], Any], delivery_attempt_recorder: Callable[[Any], None], finalize_reactions: bool = False, + require_previous_text_match: bool = False, ) -> dict[str, Any]: """Replace one known Bot reply, verifying source and exact provider readback. @@ -1052,6 +1053,8 @@ def update_lark_inbox_reply( "intent_digest": _intent_digest(profile, chat_id, receipt)} edit_limit_reached = False if not lark_markdown_readback_matches(text=text, message=message): + if require_previous_text_match and not lark_markdown_readback_matches(text=previous_text, message=message): + return {"ok": False, "reply_verified": False, "blocker": "provider_delivery_mismatch"} edit_args = [*base, "im", "+messages-edit", "--message-id", ref, "--msg-type", "post", "--content", content, "--as", "bot", "--format", "json"] preview = _call(runner, [*edit_args, "--dry-run"]) diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 60966d2d1b..1cabbfa61f 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -31,7 +31,8 @@ from .inbox_reactions import mark_lark_event_inbox_processing, mark_lark_event_inbox_received from .outbound import present_lark_conversation_text as _presentation_text from .private_images import private_message_caption, private_message_images -from .private_progress import ACTIVITY_UPDATE_INTERVAL_SEC, UPDATE_INTERVAL_SEC, project_progress +from .private_progress import (ACTIVITY_UPDATE_INTERVAL_SEC, UPDATE_INTERVAL_SEC, legacy_recovery_repairs, + presentation_phase, presentation_stream, project_progress) # Conservative transport budget, not a claim about the provider's maximum. # Reserve edits for a final/closing notice; continue long Turns on a new draft. @@ -745,7 +746,10 @@ def _feedback(self, path: Path, record: dict[str, Any], *, inbox: Callable[[], P Presentation failures never reject a persisted Turn. Existing reaction journals prevent uncertain provider writes from being repeated on replay. """ - if not self.reaction_feedback: + if not self.reaction_feedback or record.get("context_recovery"): + # The original failure already acknowledged this Inbox source. + # Preserve its reaction journal; a continuation uses its own draft + # rather than trying to reopen the completed processing reaction. return phase = "processing" if processing else "received" feedback = record.setdefault("feedback", {}) @@ -761,9 +765,12 @@ def _feedback(self, path: Path, record: dict[str, Any], *, inbox: Callable[[], P feedback[phase] = result _atomic_write_json(path, record) - def _deliver(self, path: Path, record: dict[str, Any], phase: str, text: str, *, inbox: Callable[[], Path]) -> bool: + def _deliver(self, path: Path, record: dict[str, Any], phase: str, text: str, *, inbox: Callable[[], Path], + edit_only: bool = False) -> bool: if not text: return False + kind = phase.split(".", 1)[0] + phase = presentation_phase(record, phase) phase_state = record["deliveries"].get(phase, {}) if phase_state.get("verified"): return True @@ -777,17 +784,19 @@ def _deliver(self, path: Path, record: dict[str, Any], phase: str, text: str, *, config = inbox() kwargs = dict(project=self.runtime_root, config_path=config, message_id=record["event"]["message_id"], text=text, runner=self._reply_runner, - finalize_reactions=phase not in {"admission", "progress"} and not (phase == "terminal" and record.get("commission_resources")), + finalize_reactions=not edit_only and kind not in {"admission", "progress"} and not (kind == "terminal" and record.get("commission_resources")), source_authority_verifier=lambda profile, chat, message: self._source_authority(record, profile, chat, message)) - progress = record["deliveries"].get("progress") or {} - stream = record.get("stream") or {} + progress = record["deliveries"].get(presentation_phase(record, "progress")) or {} + stream = presentation_stream(record) edit = phase_state.get("edit_from") - if phase in {"terminal", "commission_result"} and progress.get("attempt") and not stream.get("closed") and not stream.get("edit_limit_reached") and not phase_state.get("started"): + if not edit_only and kind in {"terminal", "commission_result"} and progress.get("attempt") and not stream.get("closed") and not stream.get("edit_limit_reached") and not phase_state.get("started"): # Freeze the known message proof before the first final edit. Later # recovery can replace the same ID, never append another answer. edit = {"text": _presentation_text(stream.get("confirmed_text") or progress["text"]), "attempt": stream.get("confirmed_attempt") or progress["attempt"]} phase_state["edit_from"] = edit + if edit_only and not edit: + return False if edit: def before_edit(_intent: str) -> dict[str, bool]: phase_state.update(text=text, started=True) @@ -801,14 +810,15 @@ def edit_attempt(value: Any) -> None: _atomic_write_json(path, record) result = update_lark_inbox_reply(**kwargs, previous_text=edit["text"], attempt=edit["attempt"], - before_send=before_edit, delivery_attempt_recorder=edit_attempt) - if result.get("blocker") == "provider_update_edit_limit": + before_send=before_edit, delivery_attempt_recorder=edit_attempt, + require_previous_text_match=edit_only) + if not edit_only and result.get("blocker") == "provider_update_edit_limit": # A confirmed edit rejection plus fresh non-matching readback # proves this final did not reach the draft. Preserve its proof # and use the existing source-bound, idempotent final sender. # Recovery of older, already-started final edits takes this # same path; an ambiguous failure must never take it. - record.setdefault("stream", {})["edit_limit_reached"] = True + stream["edit_limit_reached"] = True phase_state["rejected_update"] = {"blocker": result["blocker"], "edit_from": edit, "attempt": phase_state.get("attempt")} for key in ("edit_from", "attempt", "started"): @@ -816,10 +826,10 @@ def edit_attempt(value: Any) -> None: record["deliveries"][phase] = phase_state _atomic_write_json(path, record) return self._deliver(path, record, phase, text, inbox=inbox) - if result.get("blocker") in {"provider_update_too_large", "provider_update_format_unsupported"}: + if not edit_only and result.get("blocker") in {"provider_update_too_large", "provider_update_format_unsupported"}: # Do not truncate long/mention-bearing final answers to fit a # draft. Close it, then use the established full final sender. - if not self._close_progress(path, record, inbox=inbox) and not record.get("stream", {}).get("edit_limit_reached"): + if not self._close_progress(path, record, inbox=inbox) and not stream.get("edit_limit_reached"): return False phase_state.pop("edit_from", None) record["deliveries"][phase] = phase_state @@ -850,15 +860,17 @@ def attempt(value: Any) -> None: phase_state["verified"] = result.get("reply_verified") is True and result.get("ok") is True phase_state["blocker"] = result.get("blocker") record["deliveries"][phase] = phase_state + if not edit_only and record.get("context_recovery") and kind == "terminal" and phase_state["verified"]: + stream["closed"] = True _atomic_write_json(path, record) return bool(phase_state["verified"]) def _edit_progress(self, path: Path, record: dict[str, Any], text: str, *, inbox: Callable[[], Path], finalizing: bool = False) -> bool: text = _presentation_text(text) - progress = record["deliveries"].get("progress") or {} + progress = record["deliveries"].get(presentation_phase(record, "progress")) or {} if not progress.get("verified"): return self._deliver(path, record, "progress", text, inbox=inbox) - stream = record.setdefault("stream", {}) + stream = presentation_stream(record) if stream.get("edit_limit_reached") or (not finalizing and int(stream.get("edit_attempts") or 0) >= PROGRESS_EDIT_BUDGET): return False confirmed: dict[str, Any] = {} @@ -889,7 +901,7 @@ def _close_progress(self, path: Path, record: dict[str, Any], *, inbox: Callable text = "回答已生成，完整结果将单独发送。" if not self._edit_progress(path, record, text, inbox=inbox, finalizing=True): return False - record["stream"]["closed"] = True + presentation_stream(record)["closed"] = True _atomic_write_json(path, record) return True @@ -900,23 +912,24 @@ def _continue_progress(self, path: Path, record: dict[str, Any], *, inbox: Calla permits rollover. An uncertain close/create keeps its frozen intent on replay; continuation never submits or restarts the Core Turn. """ - progress = record["deliveries"].get("progress") or {} + progress_key = presentation_phase(record, "progress") + progress = record["deliveries"].get(progress_key) or {} if not progress.get("verified"): return False - stream = record["stream"] + stream = presentation_stream(record) if not stream.get("edit_limit_reached"): closed = self._edit_progress(path, record, "仍在处理，后续进展见下一条消息。", inbox=inbox, finalizing=True) if not closed and not stream.get("edit_limit_reached"): return False segment = int(stream.get("segment") or 0) - record["deliveries"][f"progress.{segment}"] = { + record["deliveries"][f"{progress_key}.{segment}"] = { **progress, "text": stream.get("confirmed_text") or progress["text"], "attempt": stream.get("confirmed_attempt") or progress["attempt"], "edit_attempts": int(stream.get("edit_attempts") or 0), "edit_limit_reached": bool(stream.get("edit_limit_reached")), } - record["deliveries"].pop("progress") + record["deliveries"].pop(progress_key) for key in ("confirmed_text", "confirmed_attempt", "edit_attempts", "edit_limit_reached", "blocker", "closed"): stream.pop(key, None) stream["segment"] = segment + 1 @@ -925,7 +938,7 @@ def _continue_progress(self, path: Path, record: dict[str, Any], *, inbox: Calla def _progress(self, path: Path, record: dict[str, Any], *, inbox: Callable[[], Path], session_id: str | None = None, turn_id: str | None = None) -> None: - stream = record.setdefault("stream", {}) + stream = presentation_stream(record) events = self.core.controller.store.events_after(session_id or record["session_id"], turn_id or record["turn_id"], stream.get("cursor")) text = project_progress(stream, events) @@ -949,7 +962,8 @@ def _progress(self, path: Path, record: dict[str, Any], *, inbox: Callable[[], P _atomic_write_json(path, record) # A failed creation must resume its frozen intent rather than trying a # new body on an uncertain message. New chunks remain in the view. - progress = record["deliveries"].get("progress") or {} + progress_key = presentation_phase(record, "progress") + progress = record["deliveries"].get(progress_key) or {} if progress.get("verified") and (stream.get("edit_limit_reached") or int(stream.get("edit_attempts") or 0) >= PROGRESS_EDIT_BUDGET): if not self._continue_progress(path, record, inbox=inbox): @@ -959,7 +973,7 @@ def _progress(self, path: Path, record: dict[str, Any], *, inbox: Callable[[], P if not progress.get("verified"): initial = str(progress.get("text") or text) if self._deliver(path, record, "progress", initial, inbox=inbox): - stream.update(confirmed_text=initial, confirmed_attempt=progress.get("attempt") or record["deliveries"]["progress"]["attempt"]) + stream.update(confirmed_text=initial, confirmed_attempt=progress.get("attempt") or record["deliveries"][progress_key]["attempt"]) _atomic_write_json(path, record) else: self._edit_progress(path, record, text, inbox=inbox) @@ -967,6 +981,41 @@ def _progress(self, path: Path, record: dict[str, Any], *, inbox: Callable[[], P stream["confirmed_answer_digest"] = answer_digest _atomic_write_json(path, record) + def _repair_recovery_presentation(self, path: Path) -> None: + """Edit proven legacy drafts only; preserve execution and final receipts.""" + with exclusive_file_lock(path, operation="deliver_private_chat_request"): + record = _read_json(path) + repairs = legacy_recovery_repairs(record) + if record["status"] != "delivered" or not repairs: + return + try: + native = self.core.read_request(record["request_ref"]) + turn = self.core.controller.store.load_turn(record["session_id"], record["turn_id"]) + if (native.get("delivery_verified") is not True or not turn + or turn["status"] not in TERMINAL_TURN_STATES + or any(native.get(key) != record.get(key) for key in + ("request_ref", "binding_id", "source", "session_id", "turn_id", "context_recovery"))): + return + selected = self.bindings.resolve(binding_id=record["binding_id"], + include_observation=True, **record["source"]) + if self._recovery_source_observation(record, selected) is None: + return + config: Path | None = None + + def inbox() -> Path: + nonlocal config + if config is None: + config = self._inbox(record) + return config + + for repair in repairs: + state = record["deliveries"].setdefault(repair["phase"], {}) + state.setdefault("edit_from", repair["edit_from"]) + _atomic_write_json(path, record) + self._deliver(path, record, repair["phase"], repair["text"], inbox=inbox, edit_only=True) + except (KeyError, ValueError, OSError): + return # Unknown authority or provider effects stay pending. + def pending_delivery_paths(self) -> list[Path]: """The durable transport store is the queue; no in-memory admission.""" paths = [] @@ -980,7 +1029,7 @@ def pending_delivery_paths(self) -> list[Path]: # The canonical request, provider effect inventory and source # are independently re-read by recover_failed_request. self._context_recovery_probe_at[path] = time.monotonic() + 60 - if self.core.context_recovery_candidate(row["request_ref"]): + if legacy_recovery_repairs(row) or self.core.context_recovery_candidate(row["request_ref"]): paths.append(path) return paths @@ -1041,6 +1090,11 @@ def reconcile_request(self, path: Path) -> int: # Repair only this persisted source. Core still owns admission, recovery # and the shared Session queue/fence; transport workers never run models. row = _read_json(path) + if row["status"] == "delivered" and row.get("context_recovery"): + # Repair presentation through the original worker, without asking + # Core to recover/enqueue any execution or creating another reply. + self._repair_recovery_presentation(path) + return 0 if row["status"] == "delivered" and row.get("session_id") and row.get("turn_id"): # Legacy terminal delivery may be only a verified failure notice. # Core decides whether a migration continuation is safe; preserve @@ -1083,7 +1137,7 @@ def reconcile_request(self, path: Path) -> int: row = _read_json(path) native_path = self.core.root / f"{row['request_ref']}.json" if native_path.exists(): - if (self.reaction_feedback and row["status"] == "accepted" + if (self.reaction_feedback and not row.get("context_recovery") and row["status"] == "accepted" and not row.get("feedback", {}).get("received", {}).get("ok")): # One bounded presentation job per existing reply worker. A # slow host cannot delay Get, nor can slow provider feedback @@ -1153,7 +1207,7 @@ def inbox() -> Path: # Get already acknowledges admission. Reserve a text # notification for a real wait or unavailable feedback; # resume any old attempt with its original exact text. - admission = (record["deliveries"].get("admission") or {}).get("text") + admission = (record["deliveries"].get(presentation_phase(record, "admission")) or {}).get("text") if not admission: if native.get("agent_target"): admission = "已收到，等待原 Agent 宿主处理。/status 查看状态，/project 返回项目对话。" @@ -1197,9 +1251,7 @@ def inbox() -> Path: or _command_text(str(record.get("response_code") or "")) or str(record.get("response") or "")) if record.get("status_snapshot"): response = _status_text(record["status_snapshot"], help_requested=native.get("command") == "help") - terminal_phase = ("terminal.recovery." + str(record["context_recovery"]["version"]) - if record.get("context_recovery") else "terminal") - if self._deliver(path, record, terminal_phase, response, inbox=inbox): + if self._deliver(path, record, "terminal", response, inbox=inbox): if control_receipt_first: # The verified result is the useful control receipt. # Preserve the usual idempotent reaction, after it. diff --git a/loopx/extensions/lark/private_progress.py b/loopx/extensions/lark/private_progress.py index 052e5860af..7482e834d8 100644 --- a/loopx/extensions/lark/private_progress.py +++ b/loopx/extensions/lark/private_progress.py @@ -6,6 +6,7 @@ from __future__ import annotations from collections.abc import Mapping +import re from typing import Any ANSWER_WINDOW = 6000 @@ -13,6 +14,63 @@ ACTIVITY_UPDATE_INTERVAL_SEC = 10.0 +def presentation_phase(record: Mapping[str, Any], phase: str) -> str: + """A continuation has its own presentation, alongside the old failure.""" + recovery = record.get("context_recovery") or {} + if recovery and phase in {"admission", "progress", "terminal", "commission_result"}: + return f"{phase}.recovery.{recovery['version']}" + return phase + + +def presentation_stream(record: dict[str, Any]) -> dict[str, Any]: + recovery = record.get("context_recovery") or {} + if recovery: + return record.setdefault("recovery_streams", {}).setdefault(str(recovery["version"]), {}) + return record.setdefault("stream", {}) + + +def legacy_recovery_repairs(record: Mapping[str, Any]) -> list[dict[str, Any]]: + """Discover only known drafts reused across an already-returned recovery. + + This is presentation debt, never authority to enqueue/replay work. The + sender still verifies the exact prior/current provider body and source. + Unknown attempts stay unmodified; no message is created by this plan. + """ + if not record.get("context_recovery"): + return [] + deliveries = record.get("deliveries") or {} + final = deliveries.get(presentation_phase(record, "terminal")) or {} + original = deliveries.get("terminal") or {} + old_ref = (original.get("attempt") or {}).get("message_ref") + final_ref = (final.get("attempt") or {}).get("message_ref") + if not (original.get("verified") and final.get("verified") and old_ref and final_ref and final_ref != old_ref): + return [] + segments = sorted((key for key in deliveries if re.fullmatch(r"progress\.[0-9]+", key)), + key=lambda key: int(key.split(".")[-1])) + candidates = [(key, deliveries[key]) for key in segments] + progress, stream = deliveries.get("progress") or {}, record.get("stream") or {} + if progress.get("verified"): + candidates.append(("progress", {**progress, + "text": stream.get("confirmed_text") or progress.get("text"), + "attempt": stream.get("confirmed_attempt") or progress.get("attempt")})) + reused = next((index for index, (_, proof) in enumerate(candidates) + if (proof.get("attempt") or {}).get("message_ref") == old_ref + and proof.get("text") != original.get("text")), None) + if reused is None: + return [] + repairs = [] + for key, proof in candidates[reused:]: + ref = (proof.get("attempt") or {}).get("message_ref") + if not proof.get("verified") or not proof.get("text") or not ref or ref == final_ref: + continue + phase = presentation_phase(record, "terminal") + ".presentation." + key + if not deliveries.get(phase, {}).get("verified"): + repairs.append({"phase": phase, "text": original["text"] if ref == old_ref else + "本轮回复已回传，后续事项以完整回复为准。", "edit_from": { + "text": proof["text"], "attempt": proof["attempt"]}}) + return repairs + + def project_progress(state: dict[str, Any], events: list[dict[str, Any]]) -> str | None: """Fold each exact Turn event once into a bounded, replayable view.""" for event in events: diff --git a/tests/test_lark_private_recovery_progress.py b/tests/test_lark_private_recovery_progress.py new file mode 100644 index 0000000000..a4749ef945 --- /dev/null +++ b/tests/test_lark_private_recovery_progress.py @@ -0,0 +1,176 @@ +"""Provider bodies, not unchanged local receipts, prove recovery presentation.""" +import copy +import json +import time +from pathlib import Path + +import pytest +from test_native_steward_private import steward, finish # noqa: F401 + +from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter +from loopx.chat_store import _atomic_write_json, _read_json +from loopx.extensions.lark.conversation_identity import identity_ref +from loopx.extensions.lark.private_conversations import LarkPrivateConversations +from loopx.extensions.lark.private_progress import legacy_recovery_repairs + + +def body(provider, ref): + return json.loads(provider.messages[ref]["body"]["content"])["zh_cn"]["content"][0][0]["text"] + + +def legacy_failure(fixture, monkeypatch): + store, runtime, provider, transport, binding, capture, _ = fixture + fake = Path(runtime.codex_bin) + source = fake.read_text().replace('"codexErrorInfo": "cyberPolicy"', '"codexErrorInfo": "contextWindowExceeded"') + source = source.replace(' print(json.dumps({"method": "item/agentMessage/delta", "params": {"threadId": "durable-thread", "turnId": active_turn, "delta": "Partial answer."}}), flush=True)\n', '') + fake.write_text(source) + event = provider.event("steward-app", "legacy-presentation", "typed terminal failure") + transport.admit("steward-app", event) + ref = identity_ref(binding["provider_ref"], event["message_id"]) + row = transport.core.read_request(ref) + failed = finish(runtime, row) + assert failed["status"] == "failed" and failed["delta_count"] == 0 + path = transport.root / f"{ref}.json" + record = _read_json(path) + # Seed the real pre-failure draft through the production sender. The old + # terminal notice then replaces that actual provider message on reconcile. + transport._edit_progress(path, record, "⏳ **正在处理**", inbox=lambda: transport._inbox(record)) + assert transport.reconcile() == 1 + original = _read_json(path) + notice = original["deliveries"]["terminal"] + assert notice["attempt"]["message_ref"] == original["deliveries"]["progress"]["attempt"]["message_ref"] + assert body(provider, notice["attempt"]["message_ref"]) == notice["text"] + store.update_session(row["session_id"], manager_context_version=20) + runtime.adapters.pop(row["session_id"]).close_session() + source = source.replace('"durable-thread"', '"recovered-thread"').replace( + 'if "typed terminal failure" in json.dumps(request.get("params") or {}):', 'if False:').replace( + 'if "wait for interrupt" in json.dumps(request.get("params") or {}) or "wait for steer" in json.dumps(request.get("params") or {}):', + 'if True:\n print(json.dumps({"method": "item/agentMessage/delta", "params": {"threadId": "recovered-thread", "turnId": active_turn, "delta": "Continuation draft."}}), flush=True)') + fake.write_text(source) + monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(lambda **_: { + "thread_id": "durable-thread", "turn_id": failed["upstream_turn_id"], + "only_original_user_item": True, "codex_error_info": "contextWindowExceeded"})) + failed_path = store._turn_path(row["session_id"], row["turn_id"]) + failed_bytes = failed_path.read_bytes() + transport.reconcile() # Original scanner discovers the old delivered error. + deadline = time.monotonic() + 10 + while True: + current = transport.core.read_request(ref) + turn = store.load_turn(current["session_id"], current["turn_id"]) + transport.reconcile() + if (turn["status"] == "running" and turn.get("upstream_turn_id") + and current["turn_id"] != row["turn_id"] + and _read_json(path)["deliveries"].get(f"progress.recovery.{current['context_recovery']['version']}", {}).get("verified")): + break + assert time.monotonic() < deadline + time.sleep(.01) + return store, runtime, provider, transport, path, original, current, capture, failed_path, failed_bytes + + +def complete_recovery(runtime, transport, current): + turn = runtime.store.load_turn(current["session_id"], current["turn_id"]) + runtime.adapters[current["session_id"]].steer_turn("finish", turn["upstream_turn_id"]) + assert finish(runtime, current)["status"] == "completed" + assert transport.reconcile() == 1 + + +def test_scanner_continuation_uses_fresh_draft_and_updates_it_to_final(steward, monkeypatch): # noqa: F811 + store, runtime, provider, transport, path, old, current, capture, failed_path, failed_bytes = legacy_failure(steward, monkeypatch) + live = _read_json(path) + version = current["context_recovery"]["version"] + key = f"progress.recovery.{version}" + notice = old["deliveries"]["terminal"] + old_ref = notice["attempt"]["message_ref"] + assert body(provider, old_ref) == notice["text"] + new_ref = live["deliveries"][key]["attempt"]["message_ref"] + assert new_ref != old_ref and body(provider, old_ref) == notice["text"] + assert live["stream"] == old["stream"] # Original cursor/edit proof is historical. + complete_recovery(runtime, transport, current) + final = _read_json(path) + assert final["deliveries"][f"terminal.recovery.{version}"]["attempt"]["message_ref"] == new_ref + assert body(provider, new_ref) == "Steered response." + assert body(provider, old_ref) == notice["text"] + assert final["deliveries"]["terminal"] == notice + assert final["recovery_streams"][str(version)]["closed"] + assert failed_path.read_bytes() == failed_bytes and len(store.list_sessions()) == 1 + calls, captured = list(provider.calls), capture.read_bytes() + restored = LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent, runner=provider, cli_bin="lark-cli") + assert restored.reconcile() == 0 and restored.reconcile_request(path) == 0 + assert capture.read_bytes() == captured + assert not any("+messages-edit" in call or "+messages-send" in call for call in provider.calls[len(calls):]) + + +@pytest.mark.parametrize("case", ["same-draft", "rollover", "changed-body", "changed-source", "revoked", "foreign-turn", "unknown-proof", "edit-limit", "lost-ack", "lost-readback"]) +def test_existing_worker_repairs_legacy_reuse_without_work_or_new_messages(steward, monkeypatch, case): # noqa: F811 + store, runtime, provider, transport, path, old, current, capture, failed_path, failed_bytes = legacy_failure(steward, monkeypatch) + complete_recovery(runtime, transport, current) + record = _read_json(path) + recovery = record.pop("context_recovery") + # Reproduce the prior release's actual edits and durable proofs, including + # a rollover which archived the old notice and left a newer draft behind. + transport._edit_progress(path, record, "⏳ **正在生成回答**", inbox=lambda: transport._inbox(record)) + if case == "rollover": + assert transport._continue_progress(path, record, inbox=lambda: transport._inbox(record)) + transport._edit_progress(path, record, "⏳ **后续进展**", inbox=lambda: transport._inbox(record)) + record["context_recovery"] = recovery + _atomic_write_json(path, record) + notice = old["deliveries"]["terminal"] + old_ref = notice["attempt"]["message_ref"] + assert body(provider, old_ref) != notice["text"] and legacy_recovery_repairs(record) + if case == "changed-body": + provider.messages[old_ref]["body"]["content"] = json.dumps({"zh_cn": {"content": [[{"tag": "text", "text": "External edit; do not overwrite."}]]}}) + elif case == "changed-source": + provider.messages[record["event"]["message_id"]]["content"] = "Edited original request" + elif case == "revoked": + transport.bindings.disconnect(record["binding_id"], expected_revision=transport.bindings.read()["revision"]) + elif case == "foreign-turn": + altered = {**record, "turn_id": old["turn_id"]} + _atomic_write_json(path, altered) + elif case == "unknown-proof": + record["stream"]["confirmed_attempt"]["intent_digest"] = "sha256:" + "f" * 64 + _atomic_write_json(path, record) + original_runner, applied, hidden = provider, [], [] + def runner(args, cwd=None, timeout=None): + if (hidden and "+messages-mget" in args + and args[args.index("--message-ids") + 1] == old_ref): + hidden.pop() + return {"returncode": 1, "stderr": "unavailable readback after applied edit"} + if "+messages-edit" in args and "--dry-run" not in args: + if case == "edit-limit": + return {"returncode": 1, "stderr": json.dumps({"error": {"type": "api", "code": 230072}})} + result = original_runner(args, cwd=cwd, timeout=timeout) + applied.append(args) + if case == "lost-readback": + hidden.append(True) + if case in {"lost-ack", "lost-readback"}: + return {"returncode": 1, "stderr": "lost acknowledgement after applied edit"} + return result + return original_runner(args, cwd=cwd, timeout=timeout) + restored = LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent, runner=runner, cli_bin="lark-cli") + before_requests, before_writes = capture.read_bytes(), copy.deepcopy(provider.writes) + before_final = record["deliveries"][f"terminal.recovery.{recovery['version']}"] + # A delivered source enters the existing scanner solely for presentation + # debt. Core recovery/enqueue must not even be consulted on this route. + monkeypatch.setattr(restored.core, "recover_failed_request", lambda *_a, **_k: pytest.fail("business replay")) + monkeypatch.setattr(restored.core, "recover", lambda *_a, **_k: pytest.fail("business replay")) + assert restored.reconcile() == 0 + after = _read_json(path) + if case == "lost-readback": + assert legacy_recovery_repairs(after) and len(applied) == 1 + assert body(provider, old_ref) == notice["text"] + restarted = LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent, runner=runner, cli_bin="lark-cli") + restarted.reconcile() + after = _read_json(path) + held = case in {"changed-body", "changed-source", "revoked", "foreign-turn", "unknown-proof", "edit-limit"} + if held: + assert not applied and legacy_recovery_repairs(after) + else: + assert body(provider, old_ref) == notice["text"] and not legacy_recovery_repairs(after) + if case == "rollover": + assert body(provider, record["deliveries"]["progress"]["attempt"]["message_ref"]) == "本轮回复已回传，后续事项以完整回复为准。" + restored.reconcile_request(path) + assert len(applied) == (2 if case == "rollover" else 1) + assert after["deliveries"]["terminal"] == notice + assert after["deliveries"][f"terminal.recovery.{recovery['version']}"] == before_final + assert provider.writes == before_writes and capture.read_bytes() == before_requests + assert failed_path.read_bytes() == failed_bytes