fix(replan): recover exact successor proof for original Turn settlement

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>

Author: huangruiteng

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

@@ -115,6 +115,23 @@ stays on the original Turn, and in-flight continuation remains unchanged.
 JSON 写作契约复用 vision 校验器，不新增 ACK 仪式，也不改变既有 successor、blocker、
 terminal 出口。语义接受、checkpoint 满足、Turn 结算与 Goal 完成仍须分别验证。
 
+An exact runnable-successor transition can settle the original Turn's selected
+replan without completing its still-open validation Todo. The shared frontier
+keeps the history-obligation receipt separate from a newly derived frontier
+duty; the TypeScript semantic gate recovers only the receipt matching the
+durable Turn guard. It reuses current canonical runnable/ownership checks, not
+Todo prose or an old creation response. Refresh and spend retain the original
+Goal/Agent/Todo/Turn identity and their existing replay behavior. Other Vision
+acceptance gaps remain visible; an unrelated Turn without that selected duty
+still requires the Todo's declared completion validation.
+
+精确绑定且仍可执行的 successor 可以结算原 Turn 选定的重规划义务，但不把原先
+未完成的验证 Todo 标为完成。共享 frontier 分别保留历史义务的成功凭证和新派生
+的义务，由 TS 语义门禁仅恢复与持久 Turn guard 匹配的凭证；资格仍来自当前
+canonical Todo 的可执行性与归属校验，不来自描述或旧创建响应。写回和扣额保持
+原 Goal/Agent/Todo/Turn 身份及既有幂等行为。其他 Vision 验收缺口仍可见；未选定
+该义务的另一 Turn 不能复用这次成功来跳过 Todo 的完成验证。
+
 Long-chain review also accepts `fresh_vision_path_outcome` and now projects this
 JSON route. An acceptance summary plus an evidence-linked `continue`, `no_change`
 or `replan` path can retain existing runnable work; no extra planning Todo or

loopx/control_plane/goals/goal_frontier/__init__.py

@@ -1648,6 +1648,10 @@ def build_goal_frontier_projection_context_from_status(
         replan_obligation=replan_obligation,
         agent_todo_items=agent_todo_source_items,
     )
+    # Keep the validated history-obligation transition distinct from any
+    # successor frontier obligation derived below. Exact Turn settlement may
+    # still owe this receipt even when the next decision has another duty.
+    run_replan_transition_ack = replan_transition_ack
     obligation_ack = replan_transition_ack or effective_replan_ack
     if (
         autonomous_replan_is_required(replan_obligation)
@@ -1766,6 +1770,7 @@ def build_goal_frontier_projection_context_from_status(
         "latest_replan_ack": latest_agent_replan_ack,
         "projected_replan_ack": projected_replan_ack,
         "replan_transition_ack": replan_transition_ack,
+        "run_replan_transition_ack": run_replan_transition_ack,
     }
 
 

loopx/control_plane/work_items/progress_observation.py

@@ -342,6 +342,27 @@ def required_semantic_outcomes(obligation: Mapping[str, Any]) -> list[str]:
     return list(replan_writeback_requirements(obligation)["required_any_of"])
 
 
+def guarded_replan_transition_delta(
+    *, guard_scoped: bool, selected_obligation_id: str | None,
+    transition_acks: list[dict[str, Any] | None],
+) -> dict[str, Any] | None:
+    """Adapt revalidated canonical receipts to the TS-owned exact Turn gate."""
+    try:
+        result = effect_runtime_result("work_item.replan_semantics.project", {
+            "operation": "turn_transition", "guard_scoped": guard_scoped,
+            "selected_obligation_id": selected_obligation_id,
+            "transition_acks": transition_acks,
+        })
+    except EffectRuntimeRejected as exc:
+        raise ValueError(str(exc)) from None
+    if not isinstance(result, Mapping) or "semantic_delta" not in result:
+        raise RuntimeError("TypeScript guarded replan transition shape mismatch")
+    delta = result["semantic_delta"]
+    if delta is not None and not isinstance(delta, Mapping):
+        raise RuntimeError("TypeScript guarded replan delta must be an object or null")
+    return dict(delta) if delta is not None else None
+
+
 def replan_obligation_trigger_kinds(
     obligation: Mapping[str, Any],
 ) -> list[str]:

loopx/control_plane/work_items/replan_semantics.ts

@@ -1,6 +1,6 @@
 import type { JsonObject } from "../effect_program.ts";
 import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
-import { requireJsonObject } from "../runtime_decode.ts";
+import { optionalNonEmptyString, requireJsonObject } from "../runtime_decode.ts";
 import { visionAuthoringContract } from "../goals/vision_checkpoint.ts";
 
 const PROGRESS_OUTCOMES = [
@@ -49,6 +49,38 @@ function isExternalReview(obligation: JsonObject): boolean {
   return triggerKinds(obligation).some(kind => EXTERNAL_REVIEW_TRIGGERS.has(kind));
 }
 
+/** Route only a revalidated canonical transition to its original Turn guard.
+ * A successor frontier duty stays with the next decision; this receipt cannot
+ * discharge an unrelated duty or certify completion of the settlement Todo.
+ */
+function projectTurnTransition(request: JsonObject): JsonObject {
+  if (typeof request.guard_scoped !== "boolean" || !Array.isArray(request.transition_acks)) {
+    throw new EffectRuntimeRequestError("turn transition requires a scoped guard and transition receipts");
+  }
+  const selected = optionalNonEmptyString(request.selected_obligation_id, "selected_obligation_id");
+  if (selected !== null && !/^replan-[a-f0-9]{16}$/.test(selected)) {
+    throw new EffectRuntimeRequestError("selected replan obligation id is malformed");
+  }
+  let delta: JsonObject | null = null;
+  if (request.guard_scoped && selected !== null) {
+    for (const value of request.transition_acks) {
+      const ack = object(value);
+      const candidate = object(ack.semantic_delta);
+      if (ack.schema_version === "autonomous_replan_ack_v0" &&
+          ack.recorded === true && ack.source === "todo_replan_successor_transition" &&
+          candidate.schema_version === "replan_semantic_delta_v0" &&
+          candidate.accepted === true && candidate.obligation_id === selected &&
+          strings(candidate.outcomes).includes("new_runnable_successor") &&
+          strings(candidate.satisfying_outcomes).includes("new_runnable_successor") &&
+          String(candidate.successor_todo_id ?? "").trim()) {
+        delta = candidate;
+        break;
+      }
+    }
+  }
+  return {semantic_delta: delta};
+}
+
 /** One outcome policy for host projection and write-time discharge. */
 export function requiredSemanticOutcomes(obligation: JsonObject): SemanticOutcome[] {
   const kinds = triggerKinds(obligation);
@@ -108,6 +140,7 @@ function writebackProjection(required: SemanticOutcome[], externalReview: boolea
 
 export function projectReplanSemantics(value: unknown): JsonObject {
   const request = requireJsonObject(value, "work_item.replan_semantics params");
+  if (request.operation === "turn_transition") return projectTurnTransition(request);
   const obligation = requireJsonObject(request.obligation, "obligation");
   const required = requiredSemanticOutcomes(obligation);
   const externalReview = isExternalReview(obligation);

loopx/control_plane/work_items/semantic_replan_writeback.py

@@ -31,7 +31,7 @@
     build_autonomous_replan_cli_actions,
     project_todo_lifecycle_settlement_reentry,
 )
-from .progress_observation import semantic_delta_from_writeback
+from .progress_observation import guarded_replan_transition_delta, semantic_delta_from_writeback
 from .repair_delta import (
     build_repair_delta_contract,
     repair_delta_kinds_have_accountable_progress,
@@ -198,6 +198,8 @@ def qualify_replan_writeback(
     completion_turn_key: str | None = None,
     todo_fields: dict[str, Any] | None = None,
     external_progress_review: Mapping[str, Any] | None = None,
+    guard_scoped: bool = False,
+    guard_semantic_replan_obligation_id: str | None = None,
 ) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
     """Return the shared open obligation and the writeback's typed delta.
 
@@ -295,6 +297,17 @@ def qualify_replan_writeback(
             else None
         ),
     )
+    if guard_scoped and guard_semantic_replan_obligation_id:
+        transition_delta = guarded_replan_transition_delta(
+            guard_scoped=guard_scoped,
+            selected_obligation_id=guard_semantic_replan_obligation_id,
+            transition_acks=[context.get("run_replan_transition_ack"),
+                             context.get("replan_transition_ack")],
+        )
+        if transition_delta is not None:
+            # This closes only the selected Turn obligation. The freshly
+            # derived frontier obligation remains visible to the next guard.
+            return None, transition_delta
     obligation = context.get("replan_obligation")
     if not obligation:
         # A validated Todo transition can discharge the read-model obligation
@@ -373,6 +386,8 @@ def enforce_open_replan_writeback(
         completion_todo_id=completion_todo_id,
         completion_turn_key=completion_turn_key,
         todo_fields=todo_fields,
+        guard_scoped=guard_scoped,
+        guard_semantic_replan_obligation_id=guard_semantic_replan_obligation_id,
     )
     if not obligation:
         if isinstance(semantic_delta, dict) and semantic_delta.get("accepted") is True:

tests/control_plane/test_replan_successor_durable_ack.py

@@ -83,7 +83,7 @@ def test_cli_successor_refresh_resets_periodic_window(tmp_path: Path, capsys) ->
         [*reversed(new_runs), compact, *runs], agent_todos={}, agent_id=AGENT) is not None
 
 
-@pytest.mark.parametrize("guard", [None, "replan-different"])
+@pytest.mark.parametrize("guard", [None, "replan-0000000000000000"])
 def test_transition_cannot_settle_a_different_turn_guard(guard) -> None:
     runs = history()
     obligation = autonomous_replan_obligation_from_runs(runs, agent_todos={}, agent_id=AGENT)
@@ -106,6 +106,145 @@ def test_transition_settles_only_exact_guard_and_owner() -> None:
             state_text=successor_state(obligation["obligation_id"], owner="another-agent"), **kwargs)
 
 
+@pytest.mark.parametrize("invalid", ["done", "deferred", "missing_target"])
+def test_non_runnable_successor_cannot_settle_original_guard(invalid: str) -> None:
+    runs = history()
+    obligation = autonomous_replan_obligation_from_runs(runs, agent_todos={}, agent_id=AGENT)
+    state = successor_state(obligation["obligation_id"])
+    state = (state.replace("target_key=source-audit ", "") if invalid == "missing_target"
+        else state.replace("status=open", f"status={invalid}"))
+    with pytest.raises(ReplanWritebackRejected):
+        enforce_open_replan_writeback(newest_first_runs=runs, state_text=state,
+            agent_id=AGENT, goal_id=GOAL, guard_scoped=True,
+            guard_semantic_replan_obligation_id=obligation["obligation_id"])
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_canonical_periodic_successor_settles_original_open_validation_todo(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str,
+) -> None:
+    """A new task changes the path, not the original task's completion."""
+    import subprocess
+    import sys
+
+    from canonical_authority_fixture import (
+        initialize_canonical_authority, isolate_sqlite_runtime,
+    )
+    from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+    from loopx.control_plane.goals.goal_vision import compact_goal_vision_packet, normalize_goal_vision_packet
+    from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos
+
+    if provider == "sqlite":
+        isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project = tmp_path / "project"
+    project.mkdir()
+    state = project / "ACTIVE_GOAL_STATE.md"
+    original_todo = "todo_original_validation"
+    state.write_text(
+        "---\nstatus: active\n---\n\n# Goal\n\n## Agent Todo\n\n"
+        "- [ ] [P1] Validate the original artifact.\n"
+        f"  <!-- loopx:todo todo_id={original_todo} status=open "
+        f"task_class=advancement_task claimed_by={AGENT} action_kind=validate "
+        "validation_command=pytest -->\n"
+    )
+    runtime = tmp_path / "runtime"
+    index = runtime / "goals" / GOAL / "runs" / "index.jsonl"
+    index.parent.mkdir(parents=True)
+    runs = history()
+    baseline_json = index.parent / "synthetic-baseline.json"
+    baseline_markdown = index.parent / "synthetic-baseline.md"
+    baseline_json.write_text(json.dumps({"ok": True, "fixture": "synthetic-replan"}))
+    baseline_markdown.write_text("# Synthetic prior delivery\n")
+    for row in runs:
+        row.update(json_path=str(baseline_json), markdown_path=str(baseline_markdown))
+    runs[0]["agent_vision"] = compact_goal_vision_packet(normalize_goal_vision_packet({
+        "goal_id": GOAL, "agent_id": AGENT, "state": "vision_drift_detected",
+        "todo_delta": [f"retain:{original_todo}"],
+        "vision_patch": {
+            "acceptance_summary": "Independently validate the source artifact.",
+            "replan_trigger_summary": "The source acceptance remains open.",
+            "advancement_policy": "repeat_until_closed",
+        },
+    }, goal_id=GOAL, agent_id=AGENT))
+    runs[0]["vision_checkpoint"] = {
+        "agent_id": AGENT, "required": True, "satisfied": False,
+        "triggers": [{"kind": "material_delivery_outcome", "delivery_outcome": "outcome_progress"}],
+    }
+    index.write_text("".join(json.dumps(row) + "\n" for row in reversed(runs)))
+    registry = tmp_path / "registry.json"
+    registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [{
+        "id": GOAL, "status": "active", "repo": str(project), "state_file": state.name,
+        "domain": "synthetic-replan",
+        "adapter": {"kind": "fixture_connected_delivery_v0", "status": "connected-delivery"},
+        "quota": {"compute": 1.0, "window_hours": 24},
+        "coordination": {"agent_model": "peer_v1", "registered_agents": [AGENT]},
+    }]}))
+    todos = parse_active_state_todos(state.read_text(), item_limit=None)["agent_todos"]["items"]
+    initialize_canonical_authority(runtime, GOAL, build_todo_runtime_shadow_projection(
+        goal_id=GOAL, todos=todos, handoff_mode="soft_claim", leases=[],
+    ), state_path=state, provider=provider)
+
+    def call(*args: str, expected_error: str | None = None) -> dict:
+        result = subprocess.run([sys.executable, "-m", "loopx.cli", "--registry", str(registry),
+            "--runtime-root", str(runtime), "--format", "json", *args], cwd=project,
+            capture_output=True, text=True, timeout=60)
+        payload = json.loads(result.stdout)
+        if expected_error is not None:
+            assert result.returncode == 1, payload
+            assert expected_error in payload["error"], payload
+        else:
+            assert result.returncode == 0, (payload.get("error"), payload.get("reason"), payload.get("status"))
+        return payload
+
+    binding = ["--goal-id", GOAL, "--agent-id", AGENT, "--todo-id", original_todo,
+               "--turn-instance-id", "turn-original-periodic-review"]
+    guard = call("quota", "should-run", "--codex-app", "--goal-id", GOAL,
+        "--agent-id", AGENT, "--turn-instance-id", "turn-original-periodic-review")
+    assert guard["selected_todo"]["todo_id"] == original_todo
+    obligation = guard["autonomous_replan_obligation"]
+    added = call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--claimed-by", AGENT,
+        "--text", "Verify an independent source artifact",
+        "--task-class", "advancement_task", "--action-kind", "validate",
+        "--target-key", "independent-source-artifact", "--operation-id", "periodic-successor",
+        "--replan-obligation-id", obligation["obligation_id"])
+    assert added["replan_transition"]["recorded"] is True
+    refresh_args = ("refresh-state", *binding, "--classification", "bounded_replan_progress",
+        "--delivery-batch-scale", "single_surface", "--delivery-outcome", "outcome_progress",
+        "--vision-unchanged-reason", "The original validation remains open; the independent successor changes the path.",
+        "--no-global-sync", "--suppress-external-sinks")
+    refreshed = call(*refresh_args)
+    persisted = json.loads(Path(refreshed["json_path"]).read_text())
+    delta = persisted["autonomous_replan_ack"]["semantic_delta"]
+    assert delta["obligation_id"] == obligation["obligation_id"]
+    assert delta["successor_todo_id"] == added["todo_id"]
+    spend_args = ("quota", "spend-slot", *binding, "--slots", "1",
+                  "--source", "heartbeat", "--execute")
+    spent = call(*spend_args)
+    assert spent["appended"] is True
+    assert spent["settlement_progress"]["state"] == "settled"
+    assert call(*refresh_args)["appended"] is False
+    assert call(*spend_args)["appended"] is False
+    settlement_runs = [json.loads(line) for line in index.read_text().splitlines()
+        if json.loads(line).get("turn_instance_id") == "turn-original-periodic-review"]
+    assert len(settlement_runs) == 2
+    assert sum(row.get("classification") == "quota_slot_spent" for row in settlement_runs) == 1
+    original = call("todo", "list", "--goal-id", GOAL, "--todo-id", original_todo)["todo"]
+    assert original["status"] == "open"
+    # Review settlement does not establish Vision acceptance or Todo completion.
+    next_guard = call("quota", "should-run", "--codex-app", "--goal-id", GOAL,
+        "--agent-id", AGENT, "--turn-instance-id", "turn-next-vision-review")
+    frontier = next_guard["goal_frontier_projection"]
+    assert "vision_outcome_checkpoint_required" in [gap["kind"] for gap in frontier["acceptance_gaps"]]
+    assert frontier["vision_continuation_audit"]["decision"] == "acceptance_gap_open"
+    call("quota", "should-run", "--codex-app", "--goal-id", GOAL,
+        "--agent-id", AGENT, "--todo-id", original_todo,
+        "--turn-instance-id", "turn-next-vision-review")
+    call("refresh-state", "--goal-id", GOAL, "--agent-id", AGENT, "--todo-id", original_todo,
+        "--turn-instance-id", "turn-next-vision-review", "--classification", "evidence_validated",
+        "--delivery-outcome", "outcome_progress", "--no-global-sync", "--suppress-external-sinks",
+        expected_error="controller-declared completion validation")
+
+
 @pytest.mark.parametrize("route", ["writeback", "successor", "canonical-successor"])
 def test_long_chain_ack_survives_real_cli_history_and_peer_claim(tmp_path: Path, route: str) -> None:
     import subprocess