fix(control-plane): recover receipt-bound terminal lifecycle settlement

docs/reference/protocols/goal-vision-replan-contract-v0.md

@@ -95,6 +95,34 @@ replan 持久记录已接受的 `new_runnable_successor` 后，新一轮 quota 
 Agent、匹配确切来源义务并通过准入。方案被替换或任务失效时恢复普通选择；
 显式选择、已绑定回执及现有 Gate 保持优先。纯文字 Vision 不能标识要推荐的任务。
 
+## Completed Todo lifecycle reentry
+
+An admitted lifecycle review retains its original Turn identity when its exact
+completed Todos are legally closed with `todo complete --no-follow-up`. The
+canonical completion identities and same-agent terminal-reentry facts must
+reconstruct the admitted obligation; ambiguous or changed sources fail closed.
+Lifecycle reentry updates `updated_at`, preserving the original completion time,
+identity and validation evidence. Terminal rows closed before admission are
+excluded. Recovery restores original
+completion order before the bounded trigger selection; it does not order sources
+by their later closeout times. Whole-second Todo timestamps allow a same-second
+closeout candidate only when the original obligation digest still matches.
+The shared typed replan owner projects settlement only, with a fresh
+evidence-linked Vision/path writeback still required before one quota debit.
+No original validator is rerun, no successor is executed in that Turn, and a
+new frontier remains the next Turn's responsibility. Lifecycle obligation
+identities now include Todo and completion identities; older ambiguous receipts
+are not rewritten or automatically admitted by this recovery.
+
+已准入的生命周期复核中，确切 Todo 合法执行 `--no-follow-up` 后，原轮身份保持
+不变。恢复须由同一 Agent 的规范终态及原完成身份重建原义务；来源不明确或
+变化时仍拒绝。恢复只允许收尾，仍须提交带证据的 Vision/path 回写后扣额一次；
+接续收尾更新 `updated_at`，保留原完成时间、身份及验证证据。准入前已收尾的
+历史记录须排除，来源按原完成顺序恢复后再截取触发项，不能按
+后来收尾的时间排序。Todo 时间戳精度为秒，同秒候选仍须匹配原义务摘要。
+不重跑原验证、不在本轮执行后续任务，也不抹除下一轮的新验收缺口。该类义务
+身份增加 Todo 与完成身份，旧的模糊回执不会被改写或自动放行。
+
 ## CLI Budget
 
 Per-agent vision is an executable control-plane field, so the CLI/write API must

loopx/control_plane/coordination/todo_terminal_reentry.ts

@@ -84,5 +84,6 @@ export function prepareTerminalReentry(todo: JsonObject, input: CoordinationTodo
   const state = selectTodoCompletionState({schema_version: TODO_COMPLETION_STATE_REQUEST_SCHEMA, todo,
     requested_no_followup: true, has_successor: false, completion_identity_source: "lifecycle_reentry"});
   return {...todo, no_followup: true, completion_continuation: state.continuation,
-    completion_recovery: state.recovery, completion_turn_key: input.requested_completion_turn_key};
+    completion_recovery: state.recovery, completion_turn_key: input.requested_completion_turn_key,
+    updated_at: input.now.toISOString().replace(/\.\d{3}Z$/u, "Z")};
 }

loopx/control_plane/goals/goal_frontier/__init__.py

@@ -1499,6 +1499,7 @@ def build_goal_frontier_projection_context_from_status(
     goal_status: str | None = None,
     agent_profile: dict[str, Any] | None = None,
     receipt_bound_replan_obligation_id: str | None = None,
+    receipt_bound_guard_recorded_at: str | None = None,
 ) -> dict[str, Any]:
     """Build the quota-facing goal-frontier read model.
 
@@ -1718,6 +1719,21 @@ def build_goal_frontier_projection_context_from_status(
             )
             frontier_transition_candidate = {"obligation": source, "ack": source_ack}
 
+    lifecycle_candidate = None
+    if receipt_bound_replan_obligation_id and receipt_bound_guard_recorded_at:
+        from ...work_items.lifecycle_recovery import receipt_bound_lifecycle_candidate
+        lifecycle_candidate = receipt_bound_lifecycle_candidate(
+            selected_obligation_id=receipt_bound_replan_obligation_id,
+            guard_recorded_at=receipt_bound_guard_recorded_at, agent_id=agent_id,
+            items=agent_todo_source_items or [],
+            derive=lambda summary: derive_goal_frontier_replan_obligation_from_summaries(
+                user_todo_summary=user_todo_summary, agent_todo_summary=summary,
+                work_lane_contract=work_lane_contract, agent_id=agent_id,
+                existing_replan_obligation=None, agent_todo_source_items=agent_todo_source_items,
+                _frontier_counts=frontier_counts,
+            ),
+        )
+
     if (
         frontier_replan_obligation
         and autonomous_replan_ack_satisfies_obligation(
@@ -1795,7 +1811,7 @@ def build_goal_frontier_projection_context_from_status(
         "projected_replan_ack": projected_replan_ack,
         "replan_transition_ack": replan_transition_ack,
         "run_replan_transition_ack": run_replan_transition_ack,
-        "replan_transition_candidates": [run_transition_candidate, frontier_transition_candidate],
+        "replan_transition_candidates": [run_transition_candidate, frontier_transition_candidate, *([lifecycle_candidate] if lifecycle_candidate else [])],
     }
 
 

loopx/control_plane/quota/live_decision.py

@@ -17,6 +17,7 @@
 from .effect_program import ReceiptBoundReplayPhase
 from .blocked_retry import overlay_active_turn_retries
 from .settlement import (
+    QuotaSettlementReadback,
     attach_settlement_progress,
     read_heartbeat_settlement,
 )
@@ -40,6 +41,23 @@
 BoundedResearchFrontierProjector = Callable[..., Mapping[str, Any] | None]
 
 
+def _receipt_bound_guard_fields(
+    readback: QuotaSettlementReadback | None,
+    obligation_id: str | None,
+) -> dict[str, Any]:
+    """Adapt the recorded guard identity and time for the shared decision owner."""
+    guard = getattr(readback, "semantic_replan_guard", None)
+    receipt = getattr(readback, "heartbeat_receipt", None) or {}
+    return {
+        "receipt_bound_replan_guard_scoped": bool(
+            obligation_id and isinstance(guard, Mapping)
+            and guard.get("scope") == "turn_guard"
+            and guard.get("selected_obligation_id") == obligation_id
+        ),
+        "receipt_bound_guard_recorded_at": receipt.get("recorded_at"),
+    }
+
+
 def _apply_retained_action_selection_reentry(
     payload: dict[str, Any],
     *,
@@ -570,12 +588,6 @@ def build_live_quota_should_run_decision(
     receipt_bound_replay_phase = (
         settlement_readback.replay_phase if settlement_readback else None
     )
-    semantic_guard = getattr(settlement_readback, "semantic_replan_guard", None)
-    receipt_bound_replan_guard_scoped = bool(
-        receipt_bound_replan_obligation_id and isinstance(semantic_guard, Mapping)
-        and semantic_guard.get("scope") == "turn_guard"
-        and semantic_guard.get("selected_obligation_id") == receipt_bound_replan_obligation_id
-    )
     fresh_operator_inbox_read = _fresh_operator_inbox_read_required(
         turn_start_hook_dispatch
     )
@@ -631,7 +643,9 @@ def build_live_quota_should_run_decision(
         receipt_bound_monitor_phase=receipt_bound_monitor_phase,
         receipt_bound_replay_phase=receipt_bound_replay_phase,
         receipt_bound_replan_obligation_id=receipt_bound_replan_obligation_id,
-        receipt_bound_replan_guard_scoped=receipt_bound_replan_guard_scoped,
+        **_receipt_bound_guard_fields(
+            settlement_readback, receipt_bound_replan_obligation_id,
+        ),
         turn_instance_id=turn_instance_id,
         runtime_root=runtime_root,
         workspace_path=workspace_path,

loopx/control_plane/quota/should_run.py

@@ -315,6 +315,7 @@ def build_quota_should_run(
     receipt_bound_terminal_phase: ReceiptBoundTerminalPhase | None = None,
     receipt_bound_replan_obligation_id: str | None = None,
     receipt_bound_replan_guard_scoped: bool = False,
+    receipt_bound_guard_recorded_at: str | None = None,
     turn_instance_id: str | None = None,
     runtime_root: str | Path | None = None,
     workspace_path: Path | None = None,
@@ -389,6 +390,7 @@ def build_quota_should_run(
             receipt_bound_replay_phase=receipt_bound_replay_phase,
             receipt_bound_replan_obligation_id=receipt_bound_replan_obligation_id,
             receipt_bound_replan_guard_scoped=receipt_bound_replan_guard_scoped,
+            receipt_bound_guard_recorded_at=receipt_bound_guard_recorded_at,
         )
         route = _resolve_quota_should_run_route(prepared)
         route = _apply_selected_todo_guards(prepared, route, workspace_path=workspace_path)

loopx/control_plane/quota/should_run_prepare.py

@@ -519,6 +519,7 @@ def _prepare_quota_should_run_item(
     receipt_bound_replay_phase: ReceiptBoundReplayPhase | None,
     receipt_bound_replan_obligation_id: str | None,
     receipt_bound_replan_guard_scoped: bool = False,
+    receipt_bound_guard_recorded_at: str | None = None,
 ) -> _QuotaDecisionPreparation:
     quota = item.get("quota") if isinstance(item.get("quota"), dict) else {}
     state = str(quota.get("state") or "unknown")
@@ -837,6 +838,7 @@ def _prepare_quota_should_run_item(
         goal_status=str(registry_goal.get("status") or ""),
         agent_profile=_quota_agent_profile(agent_identity),
         receipt_bound_replan_obligation_id=receipt_bound_replan_obligation_id,
+        receipt_bound_guard_recorded_at=receipt_bound_guard_recorded_at,
     )
     replan_obligation = (
         None

loopx/control_plane/work_items/autonomous_replan_obligation.py

@@ -228,6 +228,7 @@ def ensure_replan_novelty_policy(
                 "monitor_target_id",
                 "progress_fingerprint",
                 "agent_id",
+                *(("todo_id", "completion_turn_key") if trigger.get("kind") == "completed_advancement_without_successor" else ()),
                 *(
                     ("latest_generated_at", "oldest_counted_generated_at")
                     if trigger.get("kind")

loopx/control_plane/work_items/lifecycle_recovery.py

@@ -0,0 +1,36 @@
+"""Adapt canonical lifecycle facts to the existing typed replan owner."""
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+from ..effect_runtime import effect_runtime_result
+
+
+def receipt_bound_lifecycle_candidate(
+    *,
+    selected_obligation_id: str,
+    guard_recorded_at: str,
+    agent_id: str | None,
+    items: list[dict[str, Any]],
+    derive: Callable[[dict[str, Any]], dict[str, Any] | None],
+) -> dict[str, Any] | None:
+    facts = effect_runtime_result("work_item.replan_semantics.project", {
+        "operation": "lifecycle_source", "guard_recorded_at": guard_recorded_at,
+        "agent_id": agent_id, "items": items,
+    })
+    if not facts.get("settled"):
+        return None
+    gaps = facts["items"]
+    source = derive({"completed_without_successor_items": gaps,
+                     "completed_without_successor_count": len(gaps)})
+    if not source or source.get("obligation_id") != selected_obligation_id:
+        return None
+    result = effect_runtime_result("work_item.replan_semantics.project", {
+        "operation": "lifecycle_obligation", "obligation": source,
+        "selected_obligation_id": selected_obligation_id,
+        "guard_recorded_at": guard_recorded_at, "agent_id": agent_id, "items": items,
+    })
+    return {"obligation": result["obligation"], "source_obligation": source,
+            "guard_recorded_at": guard_recorded_at, "items": items,
+            "ack": None, "lifecycle_settled": True}

loopx/control_plane/work_items/replan_semantics.ts

@@ -94,6 +94,70 @@ function projectTurnTransition(request: JsonObject): JsonObject {
   return {semantic_delta: delta};
 }
 
+/** Reconstruct only the exact lifecycle source which could be open at admission.
+ * Canonical no-follow-up reentry preserves the original completion identity.
+ * The selected obligation digest additionally binds Todo and completion identities.
+ */
+function lifecycleSource(request: JsonObject): JsonObject {
+  const agent = optionalNonEmptyString(request.agent_id, "agent_id");
+  const admitted = typeof request.guard_recorded_at === "string"
+    ? parseTodoTimestampMicros(request.guard_recorded_at) : null;
+  if (!agent || admitted === null || !Array.isArray(request.items)) return {items: [], settled: false};
+  // Canonical Todo updates use whole seconds. A closeout in the admission
+  // second is possible; the exact obligation digest must still prove identity.
+  // Older terminal rows were already absent from the admitted frontier.
+  const admissionSecond = admitted / 1_000_000n * 1_000_000n;
+  const items = request.items.map(object).filter(item => {
+    const completed = typeof item.completed_at === "string" ? parseTodoTimestampMicros(item.completed_at) : null;
+    const updated = typeof item.updated_at === "string" ? parseTodoTimestampMicros(item.updated_at) : null;
+    return item.status === "done" && item.done === true &&
+      (item.archive_state ?? "active") === "active" && item.task_class === "advancement_task" &&
+      item.claimed_by === agent && item.last_actor_agent_id === agent &&
+      completed !== null && completed <= admitted &&
+      strings(item.successor_todo_ids).length === 0 && !item.superseded_by &&
+      (item.no_followup !== true || (item.completion_recovery === "lifecycle_reentry_terminal_closeout" &&
+        updated !== null && updated >= admissionSecond));
+  }).sort((left, right) => {
+    // Terminal reentry changes updated_at, never the original completed_at.
+    // Restore the original summary's ordering before its three-trigger bound.
+    const a = parseTodoTimestampMicros(String(left.completed_at))!;
+    const b = parseTodoTimestampMicros(String(right.completed_at))!;
+    return a === b ? Number(right.index ?? 0) - Number(left.index ?? 0) : a > b ? -1 : 1;
+  });
+  const settled = items.length > 0 && items.slice(0, 3).every(item =>
+    item.no_followup === true && item.completion_continuation === "no_followup" &&
+    typeof item.todo_id === "string" && !!item.todo_id &&
+    typeof item.completion_turn_key === "string" && !!item.completion_turn_key &&
+    item.completion_recovery === "lifecycle_reentry_terminal_closeout");
+  return {items, settled};
+}
+
+function lifecycleObligation(request: JsonObject): JsonObject | null {
+  const obligation = object(request.obligation);
+  if (obligation.obligation_id !== request.selected_obligation_id ||
+      obligation.resolution_mode !== "todo_lifecycle_settlement") return null;
+  const source = lifecycleSource(request);
+  const items = source.items as JsonObject[];
+  const triggers = Array.isArray(obligation.triggers) ? obligation.triggers.map(object) : [];
+  if (source.settled !== true || triggers.length !== Math.min(3, items.length) ||
+      triggers.some((trigger, index) => trigger.kind !== "completed_advancement_without_successor" ||
+        trigger.agent_id !== request.agent_id || trigger.todo_id !== items[index].todo_id ||
+        trigger.completion_turn_key !== items[index].completion_turn_key)) return null;
+  const reason = "The exact completed Todo lifecycle is settled. Reassess the original acceptance with evidence, then write back and settle this Turn once; do not repeat the original work.";
+  return {...obligation, required: true, selection_binding: "heartbeat_receipt",
+    resolution_mode: "receipt_bound_replan_settlement", satisfying_semantic_outcomes: [...VISION_OUTCOMES],
+    recommended_action: reason, todo_actions: [], guidance_actions: [reason],
+    settlement_action_packet: {
+      schema_version: "replan_action_packet_v0", obligation_id: obligation.obligation_id,
+      decision: "vision_outcome_checkpoint_required", settlement_only: true,
+      required_outcome: "durable_writeback_receipt", semantic_delta: null,
+      writeback_contract: {preferred_input: "evidence_linked_vision_path",
+        cli_semantic_args: VISION_CLI_ARGS, vision_authoring: visionAuthoringContract(),
+        required_fields: ["vision_patch.acceptance_summary", "path_delta.outcome", "path_delta.evidence_refs"],
+        path_outcomes: [...FRESH_PATH_DISPOSITIONS], rule: reason},
+    }};
+}
+
 /** A changed frontier cannot rebind work already admitted by a Turn receipt.
  * Current canonical transition candidates must carry their original source
  * obligation. After validated writeback the settlement owner's replay phase,
@@ -114,6 +178,11 @@ function preserveReceiptBoundObligation(request: JsonObject): JsonObject {
   for (const value of request.transition_candidates) {
     const candidate = object(value);
     const obligation = object(candidate.obligation);
+    if (request.guard_scoped && candidate.lifecycle_settled === true) {
+      const recovered = lifecycleObligation({...candidate, obligation: candidate.source_obligation,
+        selected_obligation_id: selected, agent_id: request.agent_id});
+      if (recovered) return {obligation: recovered};
+    }
     if (obligation.obligation_id !== selected || !requiredSemanticOutcomes(obligation).includes("new_runnable_successor")) continue;
     const result = projectTurnTransition({guard_scoped: request.guard_scoped,
       selected_obligation_id: selected, transition_acks: [candidate.ack]});
@@ -242,6 +311,12 @@ function acknowledgeVisionEvidence(request: JsonObject): JsonObject {
 
 export function projectReplanSemantics(value: unknown): JsonObject {
   const request = requireJsonObject(value, "work_item.replan_semantics params");
+  if (request.operation === "lifecycle_source") return lifecycleSource(request);
+  if (request.operation === "lifecycle_obligation") {
+    const obligation = lifecycleObligation(request);
+    if (!obligation) throw new EffectRuntimeRequestError("lifecycle source does not match the admitted obligation");
+    return {obligation};
+  }
   if (request.operation === "vision_ack") return acknowledgeVisionEvidence(request);
   if (request.operation === "turn_transition") return projectTurnTransition(request);
   if (request.operation === "receipt_bound_obligation") return preserveReceiptBoundObligation(request);