Skip to content

fix(chat): recover unused result-review preparation without replay #8349

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
loopx-agent merged 3 commits into from
Nov 8, 2026
Merged

fix(chat): recover unused result-review preparation without replay #8349

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e55d87b
fix(chat): preserve result evidence across host preparation failures
loopx-agent Nov 8, 2026
61630ca
style(chat): normalize result evidence module ending
loopx-agent Nov 8, 2026
ac34333
fix(chat): hold read-only preparation receipts on older runtimes
loopx-agent Nov 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2065,6 +2065,19 @@ intent. The original worker independently checks current artifacts and returns
one read-only answer under this source's own idempotent identity. A legacy review
cannot seed another legacy review; changed evidence and exhausted attempts hold.

Host event-compaction filesystem stamps are excluded from the two retained Turn
evidence locations before crossing the JSON/JavaScript boundary; nanoseconds
must not become rounded answer or authority evidence. All semantic fields stay
pinned. An already failed legacy preparation can recover once only when the
canonical trace is exactly queue then local `host_gate`, with no native input,
delta or response, and fresh native inventory excludes hidden dispatch. This
retains the failed Turn and reservation in history and uses a distinct, pinned
preparation receipt/phase. It does not renew the single model invocation: an
actual dispatch, unknown native work, changed source/evidence, Stop, or a second
preparation failure remains held. Reservation/queue crashes adopt that receipt.
Its existing `preparation-1` key carries the explicit read-only preparation tag;
older readers reject that policy combination before model dispatch on rollback.

A new result review now requires an explicit observation that no original
writable adapter is attached. An inconclusive effect readback while that owner
is still attached remains a hold; it must not replace the owner with a model
Expand Down
10 changes: 10 additions & 0 deletions docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1357,6 +1357,16 @@ binding 和受众下，更早的独立来源提出完全相同的问题，且其
原 worker 独立检查当前产物，沿本条原消息自己的幂等身份补回一次只读答案。
旧请求复核不能再作为其他旧请求的准入依据；证据变化或次数用尽继续保持等待。

两处保留 Turn 证据在跨 JSON/JavaScript 边界前排除宿主事件压缩用的文件系统标记，
避免纳秒数值舍入被当成答案或授权变化；全部语义字段仍固定。旧准备阶段已失败时，
只有规范轨迹恰为入队后本地 `host_gate`、没有原生输入、delta 或答复，且新鲜原生
清单排除隐藏派发，才允许一次准备恢复。失败 Turn 与预约继续留在历史中，恢复使用
独立固定的准备回执及投递阶段，不重新授予一次模型调用。已经派发、原生工作未知、
来源或证据变化、停止，以及第二次准备失败继续保持等待；预约和入队崩溃只认回
同一个准备回执。
它沿用 `preparation-1` 标识并携带显式只读准备标记，旧版本在回滚后会于模型
派发前拒绝这一策略组合。

新的结果复核现在要求明确观测到没有挂接原可写 adapter。原进程仍挂接时，效果
读回暂时没有结论，应保持等待，不能为复述已投递的部分答复而用模型复核替换
原进程。已有回复 worker 的探测等待间隔继续生效；后续取得新鲜效果结清证明后，
Expand Down
80 changes: 36 additions & 44 deletions loopx/capabilities/native_chat/external_conversations.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
unavailable_automation_cadence_readback,
)
from ...file_lock import exclusive_file_lock, try_exclusive_file_lock
from . import result_review_evidence
from .codex_context import checked_session_home
from .file_discovery import FileChatDiscovery

Expand Down Expand Up @@ -116,43 +117,11 @@ def _retained_result_review(self, row: dict[str, Any]) -> dict[str, Any] | None:
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"])
if reviewed is None and reservation.get("review_preparation_attempt") == 1:
reviewed = self.controller.store.load_turn(row["session_id"], reservation["preparation_failed_turn_id"])
return {"reservation": reservation, "turn": reviewed} if reviewed else None

def _related_result_review(self, row: dict[str, Any], original: dict[str, Any], *,
discovery_reads: _RecoveryDiscoveryReads | None = None) -> dict[str, Any] | None:
"""Collect same-question evidence, never equate distinct source identities.

Core qualifies the complete correlation. Discovery shares only a local
history index; admission and dispatch read the actual records afresh.
"""
store = self.controller.store
messages = (discovery_reads.matching_user_messages(row["session_id"], row["message"])
if discovery_reads is not None else [item for item in store.messages(row["session_id"], strict=True)
if item.get("role") == "user" and item.get("text") == row["message"]])
pinned = ((row.get("context_recovery") or {}).get("related_result_review") or {}).get("request", {}).get("request_ref")
for message in sorted(messages, key=lambda item: item["created_at"], reverse=True):
if (message.get("turn_id") == original["turn_id"]
or datetime.fromisoformat(message["created_at"]) <= datetime.fromisoformat(original["completed_at"])):
continue
reviewed = store.load_turn(row["session_id"], message.get("turn_id", ""))
if not reviewed or reviewed.get("status") != "completed":
continue
reservation = self.recovery_context({"session_id": row["session_id"]}, reviewed)
if not reservation or not reservation.get("unfinished_work") or reservation.get("legacy_result_review"):
continue
if pinned is not None and reservation["request_ref"] != pinned:
continue
sibling = self.read_request(reservation["request_ref"])
if (sibling["request_ref"] == row["request_ref"] or sibling.get("message") != row["message"]
or sibling.get("binding_id") != row["binding_id"] or sibling.get("source") != row["source"]
or sibling.get("turn_id") != reviewed["turn_id"] or not sibling.get("delivery_verified")):
continue
events = store.events_after(row["session_id"], reviewed["turn_id"], None)
events += store.events_after(row["session_id"], reservation["failed_turn_id"], None)
return {"request": {key: sibling.get(key) for key in ("request_ref", "binding_id", "source", "session_id",
"message", "origin", "command", "status", "turn_id", "delivery_verified", "context_recovery")},
"turn": reviewed, "steering_observed": any(event.get("kind") == "turn.steered" for event in events)}
return None

def _context_recovery_plan(self, row: dict[str, Any], *,
observation: dict[str, Any] | None = None,
Expand Down Expand Up @@ -200,7 +169,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
if event.get("kind") == "turn.started" and event.get("payload", {}).get("native_input")]
session = self._original_native_session(session, anchors)
unfinished = turn.get("status") == "completed" and bool((turn.get("response") or {}).get("unfinished_work"))
related_review = (self._related_result_review(row, turn, discovery_reads=discovery_reads)
related_review = (result_review_evidence.collect_related_result_review(self, row, turn, discovery_reads=discovery_reads)
if turn.get("status") == "completed" and not unfinished
and (not reservation or reservation.get("legacy_result_review")) else None)
live_work_owner = False
Expand Down Expand Up @@ -239,8 +208,19 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
for item in (discovery_reads.user_messages(session["session_id"])
if discovery_reads is not None else store.messages(session["session_id"])))
from ...chat_manager import steward_machine_defaults
recovery_row = {key: value for key, value in row.items() if key != "attachments"}
review_preparation = None
if reservation.get("legacy_result_review"):
recovery_row = deepcopy(recovery_row)
recovery_row["context_recovery"]["related_result_review"] = result_review_evidence.related_review_basis(reservation["related_result_review"])
if retained_review is not None:
failed_review = (store.load_turn(session["session_id"], reservation["preparation_failed_turn_id"])
if reservation.get("review_preparation_attempt") == 1 else retained_review["turn"])
if failed_review is not None:
review_preparation = {"turn": failed_review,
"events": store.events_after(session["session_id"], failed_review["turn_id"], None)}
return effect_runtime_result("collaboration.conversation.request", {
"request": {key: value for key, value in row.items() if key != "attachments"},
"request": recovery_row,
"current_session": session, "binding": selected_binding,
"context_recovery": {"version": MANAGER_CONTEXT_VERSION,
"failed_turn": ({key: value for key, value in turn.items() if key != "attachments"}
Expand Down Expand Up @@ -272,6 +252,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
**({"retained_result_review": retained_review} if retained_review is not None else {}),
**({"live_work_owner": live_work_owner} if unfinished or related_review is not None else {}),
**({"related_result_review": related_review} if related_review is not None else {}),
**({"result_review_preparation": review_preparation} if review_preparation 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,
Expand Down Expand Up @@ -431,7 +412,7 @@ def context_recovery_candidate(self, request_ref: str, *,
_, 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 and not self._retained_result_review(row)
and self._related_result_review(row, turn, discovery_reads=discovery_reads) is None):
and result_review_evidence.collect_related_result_review(self, row, turn, discovery_reads=discovery_reads) is None):
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
Expand Down Expand Up @@ -459,7 +440,7 @@ def recover_failed_request(self, request_ref: str, *,
session = self.controller.store.load_session(row["session_id"])
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)
if plan.get("completed_review_turn_id") or plan.get("original_result_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)
Expand Down Expand Up @@ -585,7 +566,7 @@ def source_requests() -> list[dict[str, Any]]:
# 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"], current_turn["turn_id"]) != current_turn
or plan.get("completed_review_turn_id")
or (plan.get("completed_review_turn_id") or plan.get("original_result_turn_id"))
and self.controller.store.load_turn(row["session_id"], turn["turn_id"]) != turn
or source_requests() != requests):
return row
Expand Down Expand Up @@ -627,6 +608,10 @@ def source_requests() -> list[dict[str, Any]]:
else:
recovery.update(legacy_result_review=True,
related_result_review=deepcopy(plan["related_result_review"]))
if plan.get("review_preparation_attempt") == 1:
recovery.update(review_preparation_attempt=1,
preparation_failed_turn_id=plan["preparation_failed_turn_id"],
presentation_key=f"{MANAGER_CONTEXT_VERSION}.review.1.preparation.1")
if plan.get("answer_review"):
recovery.update(answer_review=True, answer_executor="codex", original_turn=deepcopy(turn),
binding_snapshot=selected["binding"])
Expand Down Expand Up @@ -942,16 +927,20 @@ def recovery_context(self, session: dict[str, Any], turn: dict[str, Any]) -> dic
preparation = match[2] is not None and match[2].startswith("preparation-")
if (row.get("session_id") == session["session_id"] and recovery
and recovery["client_turn_id"] == turn["client_turn_id"]):
if (match[2] == "review-1") != (recovery.get("read_only_review") is True):
review_preparation = (match[2] == "preparation-1" and recovery.get("legacy_result_review") is True
and recovery.get("read_only_review") is True and recovery.get("review_preparation_attempt") == 1)
if (match[2] == "review-1" or review_preparation) != (recovery.get("read_only_review") is True):
raise ValueError("reserved result review policy changed")
if review_preparation != (recovery.get("review_preparation_attempt") == 1):
raise ValueError("reserved result review preparation changed")
if recovery.get("context_correction") and (recovery.get("history_review") is not True or recovery.get("context_version") != 1):
raise ValueError("historical correction requires its qualified context")
if (match[2] == "work-1") != (recovery.get("work_attempt") == 1):
raise ValueError("reserved unfinished work correlation changed")
if (match[2] == "metadata-1") != (recovery.get("metadata_review") is True):
raise ValueError("reserved metadata review correlation changed")
if preparation != (recovery.get("preparation_attempt") is not None) or (
preparation and str(recovery["preparation_attempt"]) != match[2][-1]):
if (preparation and not review_preparation) != (recovery.get("preparation_attempt") is not None) or (
preparation and not review_preparation and str(recovery["preparation_attempt"]) != match[2][-1]):
raise ValueError("reserved preparation recovery correlation changed")
return {"request_ref": row["request_ref"], **recovery}
if preparation or match[2] in {"review-1", "work-1", "metadata-1"}:
Expand Down Expand Up @@ -1240,16 +1229,19 @@ def _checked_completed_result_review(self, session: dict[str, Any], turn: dict[s
binding = self.bindings.resolve(binding_id=row["binding_id"], **row["source"])["binding"]
events = store.events_after(session["session_id"], turn["turn_id"], None)
events += store.events_after(session["session_id"], review["failed_turn_id"], None)
predecessor = (review.get("preparation_failed_turn_id")
if review.get("legacy_result_review") and review.get("review_preparation_attempt") == 1 else None)
if (not session.get("project_context") or review.get("answer_review")
or original != review.get("original_turn") or binding != review.get("binding_snapshot")
or current.get("active_turn_id") != turn["turn_id"] or current.get("status") == "closed"
or row["turn_id"] not in {turn["turn_id"], review["failed_turn_id"]}
or row["turn_id"] not in {turn["turn_id"], review["failed_turn_id"], predecessor}
or (store.turn_for_client(session["session_id"], review["client_turn_id"]) or {}).get("turn_id") != turn["turn_id"]
or store.queued_turns(session["session_id"]) or self._source_stop_observed(row, original or {}, events)
or any(event.get("kind") == "turn.steered" for event in events)):
raise CodexChatAgentError("The original result review source, grant or queue changed; review held.",
error_code="host_gate", gate=None)
if review.get("legacy_result_review") and self._related_result_review(row, original) != review.get("related_result_review"):
if (review.get("legacy_result_review") and result_review_evidence.collect_related_result_review(self, row, original)
!= result_review_evidence.related_review_basis(review["related_result_review"])):
raise CodexChatAgentError("The related result evidence changed; review held.", error_code="host_gate", gate=None)
if native is not None:
known = self._settled_native_turn_ids(session["session_id"], include_interrupted=True)
Expand Down
Loading
Loading