fix(turn): project durable completion continuation in built-in CLI provider

docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md

@@ -577,6 +577,14 @@ only the `load` / `compare_and_put` provider. This prerequisite does not create
 a generic storage abstraction over registry, run history, quota, scheduler, or
 evidence; those ledgers retain the owners defined in Section 3.
 
+Durable completion continuation read-back
+(`durable_completion.py`: `read_persisted_todo_record` /
+`project_durable_completion_outcome`) is a provider read point: it re-reads
+the persisted lifecycle record (Markdown first, event projection fallback)
+after the completion write and before settlement. Once a remote provider
+becomes canonical, this seam flips to provider-first without changing the
+typed outcome contract below.
+
 ### P0: contract and deterministic proof
 
 - this ownership matrix and explicit shared-mode boundary;
@@ -593,6 +601,10 @@ evidence; those ledgers retain the owners defined in Section 3.
 
 - lease renewal, explicit release, expired-lease reclaim, and stale-fence
   writeback rejection, all required before production shared mode;
+- durable completion continuation projection
+  (`successor | no_followup | active_goal`) with fail-closed contradiction
+  rules (`no_followup` + successors, dangling declared successor), reproduced
+  by the provider with identical semantics;
 - atomic `complete_todo_with_successor` and accepted evidence pointers;
 - transfer and restricted delegated assignment;
 - delivery/wake integration through Agent IM;

docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md

@@ -503,6 +503,13 @@ gate/dependency ref、claim/lease field 与按 privacy class 标注的 opaque po
 history、quota、scheduler 或 evidence 的通用存储抽象，这些账继续遵守第 3 节的 owner
 边界。
 
+完成续接（continuation）的持久化读回
+（`durable_completion.py`：`read_persisted_todo_record` /
+`project_durable_completion_outcome`）是一个 provider read point：它在完成写入之后、
+settlement 之前重新读取已落盘的 lifecycle record（Markdown 优先，event projection
+兜底）。一旦远端 provider 成为 canonical，这个 seam 翻转为 provider-first，且不改变
+下述 typed outcome 合同。
+
 ### P0：合同与 deterministic proof
 
 - 本 ownership matrix 与显式 shared-mode boundary；
@@ -518,6 +525,10 @@ history、quota、scheduler 或 evidence 的通用存储抽象，这些账继续
 
 - Lease renewal、显式 release、过期 lease reclaim 与 stale-fence writeback rejection；
   production shared mode 前必须完成这些能力；
+- durable completion continuation projection
+  （`successor | no_followup | active_goal`），带 fail-closed 矛盾规则
+  （`no_followup` + successors、悬挂的 declared successor），并由 provider 以相同
+  语义复现；
 - atomic `complete_todo_with_successor` 与 accepted evidence pointer；
 - transfer 与受限 delegated assignment；
 - 经 Agent IM 的 delivery/wake integration；

loopx/cli_commands/turn.py

@@ -13,6 +13,10 @@
 from ..control_plane.runtime.status_projection_cache import (
     resolve_status_projection_cache_runtime_root,
 )
+from ..control_plane.todos.durable_completion import (
+    project_durable_completion_outcome,
+    read_persisted_todo_record,
+)
 from ..control_plane.scheduler.execution_context import (
     scheduler_execution_context_for_turn,
 )
@@ -523,6 +527,38 @@ def completion_writeback(result: dict[str, object]) -> dict[str, object]:
                     project=None,
                     dry_run=False,
                 )
+                # Project the continuation the Todo lifecycle durably recorded,
+                # never a host-normalized continuation. Contradictory or
+                # dangling durable state fails closed before any further
+                # writeback so the typed settlement sees the truthful outcome.
+                state_file = completion.get("state_file")
+                if not isinstance(state_file, str) or not state_file:
+                    return {
+                        "ok": False,
+                        "appended": False,
+                        "reason": (
+                            "validated completion lifecycle did not report its "
+                            "durable Todo state file"
+                        ),
+                    }
+                try:
+                    durable_todo, existing_todo_ids = read_persisted_todo_record(
+                        Path(state_file),
+                        todo_id=todo_id,
+                        registry_path=registry_path,
+                        goal_id=args.goal_id,
+                    )
+                    completion_outcome = project_durable_completion_outcome(
+                        todo=durable_todo,
+                        expected_todo_id=todo_id,
+                        existing_todo_ids=existing_todo_ids,
+                    )
+                except ValueError as exc:
+                    return {
+                        "ok": False,
+                        "appended": False,
+                        "reason": f"durable completion projection failed: {exc}",
+                    }
                 refresh = writeback(
                     result,
                     completion_todo_id=todo_id,
@@ -536,10 +572,7 @@ def completion_writeback(result: dict[str, object]) -> dict[str, object]:
                         refresh.get("appended")
                     ),
                     "classification": refresh.get("classification"),
-                    "completion": {
-                        "todo_id": completion.get("todo_id"),
-                        "continuation": "active_goal",
-                    },
+                    "completion": completion_outcome,
                 }
 
             def current_status() -> dict[str, object]:

loopx/control_plane/todos/durable_completion.py

@@ -0,0 +1,163 @@
+"""Project the durable completion continuation for the built-in Turn CLI provider.
+
+``loopx turn run-once`` completes the selected Todo through
+:func:`loopx.todos.complete_goal_todo` and must report the continuation the Todo
+lifecycle durably recorded (``successor``, ``no_followup``, or ``active_goal``)
+instead of normalizing every completion to ``active_goal``.
+
+Projection reads the persisted Todo lifecycle record only; it never trusts host
+result JSON and never creates successors from host text. Contradictory or
+dangling durable state fails closed so the typed settlement never sees a
+fabricated continuation.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Collection
+from pathlib import Path
+from typing import Any, Mapping
+
+from .active_state_editing import (
+    TODO_SECTION_HEADINGS,
+    find_todo_block,
+    section_bounds,
+    todo_blocks,
+)
+from .contract import (
+    TODO_STATUS_DONE,
+    normalize_todo_id,
+    normalize_todo_id_list,
+    normalize_todo_no_followup,
+    normalize_todo_status,
+)
+from .event_writeback import event_projection_todo_context
+
+
+def read_persisted_todo_record(
+    state_file: Path,
+    *,
+    todo_id: str,
+    registry_path: Path | None = None,
+    goal_id: str | None = None,
+) -> tuple[dict[str, Any], set[str]]:
+    """Read one durable Todo record and the goal-state Todo id universe.
+
+    Returns the parsed Todo record for ``todo_id`` together with the set of
+    Todo ids present in the persisted goal state. Like
+    ``complete_goal_todo``, the record is looked up in the active Markdown
+    blocks first and falls back to the event projection for event-sourced
+    goals. Raises ``ValueError`` when the Todo cannot be located in the
+    durable lifecycle, so the caller can fail closed instead of projecting a
+    fabricated continuation.
+    """
+    lines = state_file.read_text(encoding="utf-8").splitlines()
+    block: dict[str, Any] | None = None
+    match = find_todo_block(lines, todo_id=todo_id)
+    if match is not None:
+        _role, _section, _start, _end, block = match
+    existing_todo_ids: set[str] = set()
+    for candidate_role in TODO_SECTION_HEADINGS:
+        bounds = section_bounds(lines, candidate_role)
+        if bounds is None:
+            continue
+        start, end, _heading = bounds
+        for item in todo_blocks(lines, start, end, role=candidate_role):
+            item_todo_id = normalize_todo_id(item.get("todo_id"))
+            if item_todo_id:
+                existing_todo_ids.add(item_todo_id)
+    if block is None and registry_path is not None and goal_id is not None:
+        context = event_projection_todo_context(
+            registry_path=registry_path,
+            goal_id=goal_id,
+            state_path=state_file,
+            todo_id=todo_id,
+            role=None,
+        )
+        if context is not None:
+            block = dict(context["item"])
+            for candidate_role in ("user", "agent"):
+                summary = context["fields"].get(f"{candidate_role}_todos")
+                items = summary.get("items") if isinstance(summary, dict) else []
+                for item in items if isinstance(items, list) else []:
+                    if not isinstance(item, dict):
+                        continue
+                    item_todo_id = normalize_todo_id(item.get("todo_id"))
+                    if item_todo_id:
+                        existing_todo_ids.add(item_todo_id)
+    if block is None:
+        normalized_todo_id = normalize_todo_id(todo_id) or todo_id
+        raise ValueError(
+            f"durable completion todo_id {normalized_todo_id!r} was not found "
+            "in persisted Todo lifecycle state"
+        )
+    return block, existing_todo_ids
+
+
+def project_durable_completion_outcome(
+    *,
+    todo: Mapping[str, Any],
+    expected_todo_id: str,
+    existing_todo_ids: Collection[str] | None = None,
+) -> dict[str, Any]:
+    """Project the typed completion continuation recorded by the Todo lifecycle.
+
+    ``todo`` is the durable Todo record re-read from persisted lifecycle state
+    after ``complete_goal_todo``. The outcome is one of:
+
+    - ``successor`` when the Todo durably declares ``successor_todo_ids``;
+      every declared successor must exist in the goal state
+      (``existing_todo_ids``) or projection fails closed;
+    - ``no_followup`` when the Todo durably records ``no_followup``;
+    - ``active_goal`` otherwise.
+
+    Raises ``ValueError`` (fail closed) when the durable record contradicts
+    itself (both ``no_followup`` and successors), when a declared successor is
+    dangling, when the Todo id does not match ``expected_todo_id``, or when the
+    Todo is not durably done.
+    """
+    normalized_expected_todo_id = normalize_todo_id(expected_todo_id)
+    if not normalized_expected_todo_id:
+        raise ValueError("durable completion requires a public Todo id")
+    durable_todo_id = normalize_todo_id(todo.get("todo_id"))
+    if durable_todo_id != normalized_expected_todo_id:
+        raise ValueError(
+            "durable completion todo_id does not match the selected Todo"
+        )
+    if normalize_todo_status(todo.get("status")) != TODO_STATUS_DONE:
+        raise ValueError(
+            "durable completion requires the selected Todo to be durably done"
+        )
+    successor_todo_ids = normalize_todo_id_list(todo.get("successor_todo_ids"))
+    no_followup = normalize_todo_no_followup(todo.get("no_followup"))
+    if no_followup is True:
+        if successor_todo_ids:
+            raise ValueError(
+                "durable completion records both no_followup and "
+                "successor_todo_ids"
+            )
+        return {
+            "todo_id": normalized_expected_todo_id,
+            "continuation": "no_followup",
+        }
+    if successor_todo_ids:
+        if existing_todo_ids is not None:
+            existing = set(existing_todo_ids)
+            missing_todo_ids = [
+                successor_todo_id
+                for successor_todo_id in successor_todo_ids
+                if successor_todo_id not in existing
+            ]
+            if missing_todo_ids:
+                raise ValueError(
+                    "durable completion declares missing successor Todo ids: "
+                    + ", ".join(missing_todo_ids)
+                )
+        return {
+            "todo_id": normalized_expected_todo_id,
+            "continuation": "successor",
+            "successor_todo_ids": successor_todo_ids,
+        }
+    return {
+        "todo_id": normalized_expected_todo_id,
+        "continuation": "active_goal",
+    }

loopx/control_plane/turn_driver/executor.py

@@ -1047,7 +1047,14 @@ def writeback_effect() -> Mapping[str, Any]:
                 return {
                     "ok": False,
                     "appended": False,
-                    "reason": "todo lifecycle adapter returned an invalid completion outcome",
+                    "reason": str(
+                        callback_payload.get("reason")
+                        or callback_payload.get("error")
+                        or (
+                            "todo lifecycle adapter returned an invalid "
+                            "completion outcome"
+                        )
+                    ),
                 }
             return {
                 **callback_payload,