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 927c48b270..85463c09a4 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -2066,6 +2066,16 @@ requires the original source, queue and native effect checks. The retained business obligation stays pending independently of the review's prose or failure; a review never starts a writable retry or another review. +A completed review does not consume the original one-shot `work-1` reservation. +If the original writable owner later becomes available through ordinary work, +the existing reply worker can reconsider the retained original declaration. +It must obtain fresh settled native-effect proof and recheck the exact original +source, grant, completed review and binding snapshot. A later equivalent source +submission still holds the old work. Review output and cold-reader inventory +cannot substitute for that proof. Admission and dispatch use the existing +continuation checks, and a crash adopts the same reserved Turn. A failed or +still-incomplete continuation remains pending and gets no additional attempt. + ### File-backed idle recovery observations Repeated recovery checks must not reparse every historical Turn and source body 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 47ea747e5e..187a9261e9 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 @@ -1356,6 +1356,13 @@ completed Turn 继续明确保留，本路径不提供通用动作或执行重 worker 先认回同一个 Turn。执行前仍重查原来源、队列和原生效果。原业务义务独立于复核正文或失败继续保留 为待办。复核不能触发可写重试或又一次复核。 +已完成的复核不消耗原请求仅一次的 `work-1` 预约。如果原可写运行进程通过普通工作 +恢复可用，现有回复 worker 可以重新检查保留的原未完声明。它必须取得新鲜的原生 +效果结清证明，并重新核验原消息、授权、已完成复核和 binding 快照。后续相同正文的 +独立来源请求仍会暂停旧工作。复核正文和冷读取清单不能替代该证明。准入及执行前 +复用已有续办检查，崩溃只认回同一预约 Turn。续办失败或仍不完整时继续保持待办， +不能增加执行次数。 + ### File 后端的空闲恢复观测 逐条检查旧请求时，不应反复解析全部历史 Turn 和来源正文。File Chat provider diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 8076eef6bb..040985aade 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -96,6 +96,18 @@ def __init__(self, controller: Any) -> None: self._source_control_discovery = FileChatDiscovery() self.actions: Any | None = None + def _retained_result_review(self, row: dict[str, Any]) -> dict[str, Any] | None: + """Read correlation only; the typed owner still checks continuation eligibility.""" + reservation = row.get("context_recovery") or {} + if reservation.get("work_attempt") == 1: + history = row.get("context_recovery_history") or [] + reservation = history[-1] if history else {} + if (reservation.get("read_only_review") is not True or not reservation.get("unfinished_work") + or reservation.get("answer_review") or reservation.get("history_review")): + return None + reviewed = self.controller.store.turn_for_client(row["session_id"], reservation["client_turn_id"]) + return {"reservation": reservation, "turn": reviewed} if reviewed else None + def _context_recovery_plan(self, row: dict[str, Any], *, observation: dict[str, Any] | None = None, source_verified: bool = False, @@ -114,6 +126,11 @@ def _context_recovery_plan(self, row: dict[str, Any], *, turn = store.load_turn(str(row.get("session_id") or ""), str(row.get("turn_id") or "")) if not session or not turn: return {} + retained_review = self._retained_result_review(row) + if retained_review is not None: + turn = store.load_turn(row["session_id"], retained_review["reservation"]["failed_turn_id"]) + if turn is None: + return {} if selected_binding is None and discovery_only: # A queue hint must not serialize provider identity IO across all # historical failures. Reuse the same typed planner with the local @@ -129,11 +146,20 @@ def _context_recovery_plan(self, row: dict[str, Any], *, if (reservation.get("routing_attempt") or reservation.get("quota_attempt") or reservation.get("preparation_attempt") or reservation.get("read_only_review") or reservation.get("work_attempt") or reservation.get("metadata_review")) - and reservation.get("failed_turn_id") == row["turn_id"] else None) + and reservation.get("failed_turn_id") == turn["turn_id"] + and (retained_review is None or reservation.get("work_attempt") == 1) else None) events = store.events_after(session["session_id"], turn["turn_id"], None) anchors = [event["payload"]["native_input"] for event in events if event.get("kind") == "turn.started" and event.get("payload", {}).get("native_input")] session = self._original_native_session(session, anchors) + live_work_owner = False + if retained_review is not None: + from .codex_adapter import CodexAppServerAdapter + with self.controller.lock: + adapter = self.controller.adapters.get(row["session_id"]) + live_work_owner = (isinstance(adapter, CodexAppServerAdapter) and not adapter.session.read_only_review + and adapter.upstream_thread_id == session["upstream_thread_id"]) + events += store.events_after(session["session_id"], retained_review["turn"]["turn_id"], None) # This only selects read-only result probes. Adoption still requires # the exact original anchor, native final, current source and typed rule. from ...chat import _review_openings @@ -156,6 +182,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *, 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 (retained_review is None or item.get("turn_id") != retained_review["turn"]["turn_id"]) and item.get("text") == row.get("message") and item.get("origin") in {None, "lark", "web"} and datetime.fromisoformat(item["created_at"]) > created for item in (discovery_reads.user_messages(session["session_id"]) @@ -191,6 +218,8 @@ def _context_recovery_plan(self, row: dict[str, Any], *, for item in (discovery_reads.source_controls() if discovery_reads is not None else self.source_controls())) if turn.get("error_code") == "antigravity_step_error" else None, "native_input": anchors[0] if len(anchors) == 1 else None, + **({"retained_result_review": retained_review, "live_work_owner": live_work_owner} + if retained_review is not None else {}), "response_literal_open": bool(open_literal), "metadata_capacity": metadata_capacity, "executor_configuration": steward_machine_defaults(self.controller) if metadata_capacity else None, @@ -349,7 +378,7 @@ def context_recovery_candidate(self, request_ref: str, *, from ...chat import _review_openings _, open_literal, _ = _review_openings(str((turn.get("response") or {}).get("message") or ""), 0) invalid_metadata = ((turn.get("response") or {}).get("gate") or {}).get("kind") == "response_metadata_invalid" - if not open_literal and not invalid_metadata: + if not open_literal and not invalid_metadata and not self._retained_result_review(row): return False # A lexical probe hint, never permission to adopt or send. return self._context_recovery_plan(row, discovery_only=True, discovery_reads=discovery_reads).get("recovery_candidate") is True @@ -375,8 +404,10 @@ def recover_failed_request(self, request_ref: str, *, if not plan.get("recovery_candidate"): return row session = self.controller.store.load_session(row["session_id"]) - turn = self.controller.store.load_turn(row["session_id"], row["turn_id"]) - assert session is not None and turn is not None + current_turn = self.controller.store.load_turn(row["session_id"], row["turn_id"]) + turn = (self.controller.store.load_turn(row["session_id"], plan["failed_turn_id"]) + if plan.get("completed_review_turn_id") else current_turn) + assert session is not None and turn is not None and current_turn is not None if plan.get("recover_native_result"): return self._recover_native_result(path, row, session, turn, source_observation) metadata_review = plan.get("metadata_review") is True @@ -497,7 +528,9 @@ def source_requests() -> list[dict[str, Any]]: # its execution and newer input may legitimately have advanced. # The typed planner below must still prove that exact receipt. if not plan.get("admitted_turn_id") and (self.controller.store.load_session(row["session_id"]) != session - or self.controller.store.load_turn(row["session_id"], row["turn_id"]) != turn + or self.controller.store.load_turn(row["session_id"], current_turn["turn_id"]) != current_turn + or plan.get("completed_review_turn_id") + and self.controller.store.load_turn(row["session_id"], turn["turn_id"]) != turn or source_requests() != requests): return row plan = self._context_recovery_plan(row, observation=proof, source_verified=source is not None, @@ -523,7 +556,7 @@ def source_requests() -> list[dict[str, Any]]: if recovery is None or recovery.get("client_turn_id") != plan["client_turn_id"]: if recovery is not None: row.setdefault("context_recovery_history", []).append(deepcopy(recovery)) - recovery = {"version": MANAGER_CONTEXT_VERSION, "failed_turn_id": row["turn_id"], + recovery = {"version": MANAGER_CONTEXT_VERSION, "failed_turn_id": turn["turn_id"], "native_observation": proof, "source_observation": source, "client_turn_id": plan["client_turn_id"], "failure_notice_delivery_verified": row["delivery_verified"]} @@ -548,6 +581,9 @@ def source_requests() -> list[dict[str, Any]]: recovery.update(work_attempt=1, unfinished_work=plan["unfinished_work"], prior_result_delivery_verified=row["delivery_verified"], presentation_key=f"{MANAGER_CONTEXT_VERSION}.work.1", not_before=plan["not_before"]) + if plan.get("completed_review_turn_id"): + recovery.update(completed_review_turn_id=plan["completed_review_turn_id"], + original_turn=deepcopy(turn), binding_snapshot=selected["binding"]) for kind in ("routing", "quota", "preparation"): if plan.get(f"{kind}_attempt") is not None: recovery.update({f"{kind}_attempt": plan[f"{kind}_attempt"], @@ -1185,6 +1221,18 @@ def prepare_continuation_input(self, session: dict[str, Any], turn: dict[str, An if not continuation or continuation.get("work_attempt") != 1: return message proof = continuation.get("native_observation") or {} + if continuation.get("completed_review_turn_id"): + row = self.read_request(continuation["request_ref"]) + retained = self._retained_result_review(row) + original = self.controller.store.load_turn(session["session_id"], continuation["failed_turn_id"]) + binding = self.bindings.resolve(binding_id=row["binding_id"], **row["source"])["binding"] + if (not retained or retained["turn"]["turn_id"] != continuation["completed_review_turn_id"] + or retained["turn"].get("status") != "completed" or original != continuation["original_turn"] + or binding != continuation["binding_snapshot"] + or self._source_stop_observed(row, original or {}, self.controller.store.events_after( + session["session_id"], continuation["failed_turn_id"], None))): + raise CodexChatAgentError("The retained work source, review or grant changed; continuation stopped.", + error_code="host_gate", gate=None) if (self.controller.store.queued_turns(session["session_id"]) or completed_work_readback(native, str(proof.get("turn_id") or ""), known_turn_ids=self._settled_native_turn_ids(session["session_id"])) != proof): diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index a3abecb673..b195c49cd0 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -567,7 +567,9 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject { // the reviewer says; neither a delivered review nor its failure is replay authority. const originalResult = reservation?.original_turn as JsonObject | undefined; const originalResponse = originalResult?.response as JsonObject | undefined; - const retainedWork = reservation?.read_only_review === true && reservation.answer_review !== true + const retainedWork = reservation != null && reservation.answer_review !== true + && (reservation.read_only_review === true || reservation.work_attempt === 1 + && typeof reservation.completed_review_turn_id === "string" && !pureResultText(response)) && turn?.client_turn_id === reservation.client_turn_id && originalResult?.status === "completed" && originalResult.session_id === row.session_id && originalResult.turn_id === reservation.failed_turn_id && originalResult.turn_id !== row.turn_id && originalResponse != null @@ -710,6 +712,9 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO || saved.project_ref !== selected.project_ref || (!steward && saved.grant !== selected.grant) || current.channel_id !== `${steward ? "manager.external.native" : "project.external"}.${selected.binding_id}.${saved.source_ref}`) return no; const failed = requireJsonObject(recovery.failed_turn, "original failed Turn"); + if (recovery.retained_result_review != null) { + return planWorkAfterResultReview(recovery, row, current, failed, selected, version); + } const runtimePreparation = planRuntimePreparationRecovery(recovery, row, current, failed, version); if (runtimePreparation !== null) return runtimePreparation; const preparation = steward ? planPreparationRecovery(recovery, row, current, failed, version) : null; @@ -1128,6 +1133,46 @@ function unfinishedProjectWork(result: JsonObject): JsonObject | null { return work; } +function planWorkAfterResultReview(recovery: JsonObject, row: JsonObject, current: JsonObject, + original: JsonObject, selected: JsonObject, version: number): JsonObject { + const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false}; + const retained = requireJsonObject(recovery.retained_result_review, "retained result review"); + const reservation = requireJsonObject(retained.reservation, "original review reservation"); + const reviewed = requireJsonObject(retained.turn, "completed result review"); + const previous = requireJsonObject(row.context_recovery, "current recovery reservation"); + const history = Array.isArray(row.context_recovery_history) ? row.context_recovery_history as JsonObject[] : []; + const continuing = previous.work_attempt === 1; + const expected = continuing ? history.at(-1) : previous; + const work = original.response != null && typeof original.response === "object" && !Array.isArray(original.response) + ? unfinishedProjectWork(original.response as JsonObject) : null; + // A delivered read-only report retains intent. Only a fresh observation by + // the original writable owner can admit the already bounded remaining work. + if (selected.context_kind !== "project" || current.agent_id !== "codex" || work === null + || !isDeepStrictEqual(expected, reservation) || reservation.read_only_review !== true + || reservation.answer_review === true || reservation.history_review === true || reservation.work_attempt != null + || reservation.version !== version || !isDeepStrictEqual(reservation.binding_snapshot, selected) + || reservation.client_turn_id !== `external-${row.request_ref}-recovery-${version}-review-1` + || !isDeepStrictEqual(reservation.original_turn, original) || !isDeepStrictEqual(reservation.unfinished_work, work) + || original.status !== "completed" || original.session_id !== current.session_id + || original.turn_id !== reservation.failed_turn_id || original.client_turn_id !== `external-${row.request_ref}` + || original.message !== row.message || original.origin !== row.origin + || reviewed.status !== "completed" || reviewed.session_id !== current.session_id + || reviewed.turn_id !== row.turn_id || reviewed.turn_id === original.turn_id + || reviewed.client_turn_id !== reservation.client_turn_id || reviewed.message !== row.message + || reviewed.origin !== row.origin || typeof reviewed.upstream_turn_id !== "string" || !reviewed.upstream_turn_id + || !pureResultText(reviewed.response as JsonObject | null) + || history.some(item => item.work_attempt != null) + || continuing && (previous.failed_turn_id !== original.turn_id + || previous.completed_review_turn_id !== reviewed.turn_id)) return no; + const admitted = continuing ? recovery.admitted_turn : null; + if (admitted == null && (recovery.live_work_owner !== true || recovery.later_equivalent_input_present !== false)) return no; + const proof = recovery.native_observation as JsonObject | null | undefined; + if (proof != null && (proof.completed_work !== true || proof.all_effects_settled !== true)) return no; + const plan = planUnfinishedWork({...recovery, admitted_turn: admitted}, {...row, turn_id: original.turn_id}, + current, original, continuing ? previous : null, version); + return plan.work_attempt === 1 ? {...plan, completed_review_turn_id: reviewed.turn_id} : no; +} + function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: JsonObject, completed: JsonObject, previous: JsonObject | null, version: number): JsonObject { const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false}; diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index 579584e174..abb3cd9822 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/tests/control_plane_ts/conversation_binding.test.ts @@ -166,6 +166,54 @@ test("a completed project partial result admits one restricted review while reta ...retained.pending_observation, stop_observed: true}}).pending, false); assert.equal(planBoundConversationRequest({...packet, request: retained.request, context_recovery: {...packet.context_recovery, failed_turn: reviewed}}).recovery_candidate, false); + + // A review does not consume the original single writable continuation. + // Its original live owner must independently settle the native effects first. + const reviewedTurn = {...reviewed, upstream_turn_id: "native-review", + response: {schema_version: "loopx_chat_agent_response_v0", ...reviewed.response}}; + const retainedReview = {reservation: {...reservation, binding_snapshot: selected}, turn: reviewedTurn}; + const settled = {thread_id: "native", turn_id: completed.upstream_turn_id, + completed_work: true, all_effects_settled: true, unfinished_work: work, + native_inventory_sha256: "2".repeat(64)}; + const afterReview = {...packet, request: {...retained.request, context_recovery: retainedReview.reservation}, + context_recovery: {...packet.context_recovery, retained_result_review: retainedReview, + live_work_owner: true, native_observation: settled}}; + const resumed = planBoundConversationRequest(afterReview); + assert.equal(resumed.recovery_eligible, true); + assert.equal(resumed.work_attempt, 1); + assert.equal(resumed.failed_turn_id, completed.turn_id); + assert.equal(resumed.completed_review_turn_id, reviewed.turn_id); + assert.equal(resumed.read_only_review, undefined); + for (const change of [{live_work_owner: false}, {source_verified: false}, {stop_observed: true}, + {steering_observed: true}, {queued_input_present: true}, {later_equivalent_input_present: true}, + {attachments_present: true}, {native_observation: proof}, + {native_observation: {...settled, all_effects_settled: false}}, + {native_observation: {...settled, turn_id: "different-original"}}, + {retained_result_review: {...retainedReview, turn: {...reviewedTurn, status: "failed"}}}, + {retained_result_review: {...retainedReview, reservation: {...retainedReview.reservation, answer_review: true}}}, + {retained_result_review: {...retainedReview, reservation: {...retainedReview.reservation, + unfinished_work: {...work, remaining: "Unrelated work"}}}}, + {retained_result_review: {...retainedReview, reservation: {...retainedReview.reservation, + binding_snapshot: {...selected, project_ref: "0".repeat(24)}}}}, + ]) assert.notEqual(planBoundConversationRequest({...afterReview, + context_recovery: {...afterReview.context_recovery, ...change}}).recovery_eligible, true); + const workReservation = {version: 22, failed_turn_id: completed.turn_id, work_attempt: 1, + unfinished_work: work, client_turn_id: resumed.client_turn_id, completed_review_turn_id: reviewed.turn_id, + original_turn: completed, binding_snapshot: selected}; + const admitted = {...reviewedTurn, turn_id: "remaining", client_turn_id: resumed.client_turn_id, status: "queued"}; + const crash = {...afterReview, request: {...afterReview.request, context_recovery: workReservation, + context_recovery_history: [retainedReview.reservation]}, + current_session: {...session, status: "busy", active_turn_id: admitted.turn_id}, + context_recovery: {...afterReview.context_recovery, native_observation: null, live_work_owner: false, + admitted_turn: admitted, queued_input_present: true, later_equivalent_input_present: true}}; + assert.equal(planBoundConversationRequest(crash).admitted_turn_id, admitted.turn_id); + assert.equal(planBoundConversationRequest({...crash, context_recovery: { + ...crash.context_recovery, source_verified: false}}).recovery_eligible, false); + assert.equal(planBoundConversationRequest({...crash, request: { + ...crash.request, turn_id: admitted.turn_id}}).recovery_candidate, false); + const failedContinuation = {...retained, request: {...crash.request, turn_id: admitted.turn_id}, + pending_observation: {turn: {...admitted, status: "failed", response: null, error_code: "host_gate"}, stop_observed: false}}; + assert.equal(planBoundConversationRequest(failedContinuation).pending, true); }); test("a restarted Codex steward retains one restricted answer instead of replaying business", () => { diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 312f0b183c..a835ce0e1d 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -2027,7 +2027,9 @@ 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"]) + "reservation-crash", "post-enqueue-crash", "queued-newer", "queued-items", "queued-input", "private-path", "still-incomplete", "cold-native", "dead-native", "foreign-native", "failed-effect", + "review-live", "review-live-source", "review-live-stop", "review-live-revoked", "review-live-equivalent", "review-live-unknown", + "review-live-reservation-crash", "review-live-enqueue-crash"]) 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 @@ -2135,7 +2137,7 @@ def test_original_worker_continues_explicit_unfinished_project_work_once_after_r if fault == "unsettled-command": snapshot["items"][1]["item"]["exitCode"] = None inventory.write_text(json.dumps(snapshot)) - if fault == "cold-native": + if fault == "cold-native" or str(fault).startswith("review-live"): runtime.adapters.pop(original["session_id"]).close_session() elif fault == "dead-native": runtime.adapters[original["session_id"]].close_session() @@ -2198,7 +2200,7 @@ def crash(**kwargs): artifact_before_recovery = artifact.read_bytes() selected.reconcile() continued = selected.core.read_request(ref) - reviewed = fault in {"cold-native", "failed-effect", "equivalent-later", "later-failed-effect", "unsettled-command"} + reviewed = fault in {"cold-native", "failed-effect", "equivalent-later", "later-failed-effect", "unsettled-command"} or str(fault).startswith("review-live") if fault in {"changed-source", "revoked", "stop", "unknown-native", "background", "later-user", "dead-native", "foreign-native"}: assert continued["turn_id"] == original["turn_id"] and not continued.get("context_recovery") assert artifact.read_bytes() == artifact_before_recovery @@ -2232,6 +2234,56 @@ def crash(**kwargs): assert permissions["filesystem"][":workspace_roots"] == {".": "read"} assert permissions["network"]["enabled"] is False assert sum(row.get("method") == "turn/start" for row in recovery_calls) == 1 + if str(fault).startswith("review-live"): + # An ordinary source request naturally restores the writable owner. + # Merely opening a cold reader must never manufacture effects proof. + later_message = event["content"] if fault == "review-live-equivalent" else "Newer follow-up: inspect another result." + later_event = provider.event("notes-app", "after-result-review", later_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) + assert runtime.wait_for_turn(session_id=later["session_id"], turn_id=later["turn_id"], timeout_sec=10)["status"] == "completed" + transport.reconcile_request(transport.root / f"{later_ref}.json") + before = artifact.read_bytes() + if fault == "review-live-source": + provider.messages[event["message_id"]]["content"] = "Changed original source" + elif fault == "review-live-stop": + transport.admit("notes-app", provider.event("notes-app", "after-review-stop", "/stop")) + elif fault == "review-live-revoked": + transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"]) + elif fault == "review-live-unknown": + snapshot = json.loads(inventory.read_text()) + snapshot["items"].append({"turnId": turn["upstream_turn_id"], "item": {"id": "unknown", "type": "futureEffect"}}) + inventory.write_text(json.dumps(snapshot)) + selected = worker() + if fault in {"review-live-reservation-crash", "review-live-enqueue-crash"}: + enqueue = runtime.enqueue_turn + def crash_after_review(**kwargs): + if fault == "review-live-enqueue-crash": + enqueue(**kwargs) + raise OSError("synthetic post-review admission crash") + monkeypatch.setattr(runtime, "enqueue_turn", crash_after_review) + with pytest.raises(OSError, match="post-review admission crash"): + selected.reconcile() + monkeypatch.setattr(runtime, "enqueue_turn", enqueue) + selected = worker() + selected.reconcile() + resumed = selected.core.read_request(ref) + if fault in {"review-live", "review-live-reservation-crash", "review-live-enqueue-crash"}: + assert resumed["turn_id"] != continued["turn_id"] + assert resumed["context_recovery"]["work_attempt"] == 1 + assert resumed["context_recovery"]["failed_turn_id"] == original["turn_id"] + assert runtime.wait_for_turn(session_id=resumed["session_id"], turn_id=resumed["turn_id"], timeout_sec=10)["status"] == "completed" + selected.reconcile_request(path) + assert artifact.read_text() == "first\nremaining\n" + assert json.loads(path.read_text())["deliveries"][presentation_phase(resumed, "terminal")]["verified"] + after_writes = list(provider.writes) + worker().reconcile() + assert provider.writes == after_writes + else: + assert resumed["turn_id"] == continued["turn_id"] + assert artifact.read_bytes() == before + assert store._turn_path(original["session_id"], original["turn_id"]).read_bytes() == original_turn_bytes else: assert continued["turn_id"] != original["turn_id"] assert continued["session_id"] == original["session_id"] @@ -2269,8 +2321,9 @@ def crash(**kwargs): assert "Verified remaining work completed." in bodies assert store._turn_path(original["session_id"], original["turn_id"]).read_bytes() == original_turn_bytes after_recovery_calls = [json.loads(line) for line in capture.read_text().splitlines()] - expected_new = {"initialize": int(reviewed) + int(fault == "cold-native"), - "thread/start": 0, "thread/resume": int(reviewed)} + restored_owner = int(str(fault).startswith("review-live")) + expected_new = {"initialize": int(reviewed) + int(fault == "cold-native") + 2 * restored_owner, + "thread/start": 0, "thread/resume": int(reviewed) + restored_owner} for method, count in expected_new.items(): assert sum(row.get("method") == method for row in after_recovery_calls[len(before_recovery_calls):]) == count