fix(turn): bind completion to the admitted Todo text

docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md

@@ -626,6 +626,18 @@ fresh Turn settling once. Boundary-length valid results retain their behavior.
 This is result-transport correctness; it does not qualify model adoption, undo
 Host work already performed, or close the three convergence streams.
 
+Turn completion now carries an exact text commitment from the admitted full
+Todo source to the existing atomic terminal owner. Changed requirements refuse
+completion; note-only and unrelated Todo changes remain valid. Provider CAS
+fences concurrent replacement, and historical receipt recovery precedes fresh
+source checks. Saved Turns without full-source proof refuse new completion;
+already accepted legacy receipts retain their fingerprints and recover first.
+File/SQLite CLI and shared File/SQLite/PostgreSQL conformance
+cover refusal and recovery. Final no-followup retains its existing settlement
+order: a terminal refusal preserves any earlier writeback/spend receipts.
+This qualifies task-body completion safety, not settlement-wide source atomicity,
+installed host adoption, live models or the remaining convergence acceptance.
+
 The first ordinary-CLI consolidation replaces an exactly identical
 `heartbeat_recommendation.blocked_priority_fallback` mirror with
 `blocked_priority_fallback_ref="$.blocked_priority_fallback"`. The complete root

docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md

@@ -431,6 +431,14 @@ File/SQLite 原生 CLI
 修正后的新 Turn 只结算一次。边界长度的合法结果保持原行为。这只证明结果传输
 正确性，不证明模型采用、撤销 Host 已执行工作或关闭原三批收敛验收。
 
+Turn 完成写入现将已准入 Todo 全文的精确摘要交给既有原子终态 owner。正文要求
+变化拒绝完成，单独备注或其他 Todo 变化仍有效；provider CAS 阻止并发替换，历史
+receipt 恢复先于当前来源检查。已保存 Turn 缺少全文来源证明时拒绝新完成写入，
+已接受的旧 receipt 则保留原指纹并优先恢复。真实 File/SQLite CLI 和共享 File/SQLite/PostgreSQL
+conformance 覆盖拒绝与恢复。最终 no-followup 保留既有结算顺序：终态拒绝保留此前
+已经提交的 writeback/spend receipt。本批只证明任务正文的完成安全性，不证明整个
+结算的来源原子性、安装态宿主采用、真实模型或其余收敛验收完成。
+
 首个普通 CLI 整合将完全相同的
 `heartbeat_recommendation.blocked_priority_fallback` 副本替换为
 `blocked_priority_fallback_ref="$.blocked_priority_fallback"`。根字段的完整事实、

loopx/cli_commands/turn_run_once.py

@@ -54,7 +54,7 @@
 from .turn_dsh_host import build_dsh_host_runner, build_dsh_operation_host_runner
 from .turn_rendering import build_turn_error_payload
 from .turn_post_writeback import drain_committed_turn_post_writeback_hooks
-from .turn_todo_writeback import write_turn_validated_completion
+from .turn_todo_writeback import captured_todo_text_sha256, write_turn_validated_completion
 
 EXACT_SETTLEMENT_READBACK_NOT_FOUND = (
     "exact settlement readback unexpectedly returned not-found"
@@ -122,6 +122,9 @@ def execute_turn_run_once(
             else {}
         )
         selected_todo = selected_turn_todo(envelope)
+        expected_todo_text_sha256 = captured_todo_text_sha256(
+            envelope, todo_id=str(selected_todo.get("todo_id") or "")
+        )
         transaction = (
             payload.get("transaction")
             if isinstance(payload.get("transaction"), Mapping)
@@ -350,6 +353,7 @@ def todo_completion(
                 goal_id=args.goal_id,
                 todo_id=todo_id,
                 completion_turn_key=settlement_identity.turn_instance_id,
+                expected_todo_text_sha256=expected_todo_text_sha256,
                 evidence=(
                     "LoopX Turn validated completion: "
                     + str(result.get("summary") or result["classification"])
@@ -359,6 +363,8 @@ def todo_completion(
                 completion_delivery_workspace=completion_delivery_workspace,
                 completion_validation_workspace_path=delivery_workspace_path,
             )
+            if completion.get("ok") is not True:
+                return completion
             # Project the continuation the Todo lifecycle durably recorded,
             # never a host-normalized continuation. Contradictory or
             # dangling durable state fails closed before any further

loopx/cli_commands/turn_todo_writeback.py

@@ -1,10 +1,64 @@
 """Todo writeback adapters that preserve the Turn's effective runtime root."""
+
 from __future__ import annotations
 
 from pathlib import Path
 from typing import Any, Mapping
+from hashlib import sha256
 
 from ..control_plane.todos.mutation_api import complete_goal_todo
+from ..control_plane.coordination.local_authority import (
+    LocalCoordinationAuthorityRejection,
+)
+
+
+def captured_todo_text_sha256(
+    envelope: Mapping[str, Any], *, todo_id: str
+) -> str | None:
+    """Resolve admitted full-source references; hash bytes without rereading live state.
+
+    Older journal envelopes have no verification marker and retain their
+    original completion request. New producers mark source validation before
+    body deduplication, so a display preview alone can never become the source.
+    """
+    action = envelope.get("action") or {}
+    selected = action.get("selected_todo") or {}
+    if "text_source_verified" not in selected:
+        return None
+    if (
+        selected["text_source_verified"] is not True
+        or selected.get("todo_id") != todo_id
+    ):
+        raise ValueError("Turn task source does not match its selected Todo")
+    context = envelope.get("work_context") or {}
+    if not isinstance(context, Mapping) or not isinstance(
+        context.get("sources", []), list
+    ):
+        raise ValueError("Turn completion requires its admitted full task source")
+    records = []
+    for source in context.get("sources", []):
+        if not isinstance(source, Mapping) or source.get("source") != "selected_todo":
+            continue
+        content = source.get("content")
+        record = content.get("todo") if isinstance(content, Mapping) else None
+        if (
+            isinstance(record, Mapping)
+            and record.get("todo_id") == todo_id
+            and isinstance(record.get("text"), str)
+        ):
+            records.append(record["text"])
+    if len(records) > 1:
+        raise ValueError("Turn task source is ambiguous")
+    text = records[0] if records else None
+    if text is None and (
+        not context or context.get("selected_todo_ref") == "selected_todo"
+    ):
+        text = selected.get("text")
+        if text is None and selected.get("text_ref") == "action.recommended_action":
+            text = action.get("recommended_action")
+    if not isinstance(text, str) or not text:
+        raise ValueError("Turn completion requires its admitted full task source")
+    return sha256(text.encode("utf-8")).hexdigest()
 
 
 def write_turn_validated_completion(
@@ -17,24 +71,32 @@ def write_turn_validated_completion(
     evidence: str,
     note: str,
     agent_id: str | None,
+    expected_todo_text_sha256: str | None = None,
     completion_delivery_workspace: Mapping[str, Any] | None = None,
     completion_validation_workspace_path: Path | None = None,
 ) -> dict[str, Any]:
     """Complete one validated Todo under the effective runtime root."""
 
-    return complete_goal_todo(
-        registry_path=registry_path,
-        goal_id=goal_id,
-        todo_id=todo_id,
-        role="agent",
-        completion_turn_key=completion_turn_key,
-        completion_identity_source="turn_settlement",
-        completion_delivery_workspace=completion_delivery_workspace,
-        completion_validation_workspace_path=completion_validation_workspace_path,
-        evidence=evidence,
-        note=note,
-        agent_id=agent_id,
-        project=None,
-        dry_run=False,
-        runtime_root_arg=runtime_root_arg,
-    )
+    try:
+        return complete_goal_todo(
+            registry_path=registry_path,
+            goal_id=goal_id,
+            todo_id=todo_id,
+            role="agent",
+            completion_turn_key=completion_turn_key,
+            completion_identity_source="turn_settlement",
+            expected_todo_text_sha256=expected_todo_text_sha256,
+            require_captured_todo_text=True,
+            completion_delivery_workspace=completion_delivery_workspace,
+            completion_validation_workspace_path=completion_validation_workspace_path,
+            evidence=evidence,
+            note=note,
+            agent_id=agent_id,
+            project=None,
+            dry_run=False,
+            runtime_root_arg=runtime_root_arg,
+        )
+    except LocalCoordinationAuthorityRejection as exc:
+        # A typed owner rejection proves that this attempt did not write. Lost
+        # responses and unavailable authority still raise for receipt recovery.
+        return {**exc.payload, "ok": False, "appended": False, "completed": False}

loopx/control_plane/coordination/local_authority_runtime.ts

@@ -1322,6 +1322,10 @@ export async function terminalLifecycleLocalCoordinationTodo(
       const researchEvidence = new ResearchTerminalEvidenceHost(root, goalId, requireJsonObject(input.registry_source, "registry source"));
       try {
       return {...await executeCoordinationTodoTerminalLifecycle(store, {
+        ...(input.expected_todo_text_sha256 === undefined ? {} : {
+          expected_todo_text_sha256: input.expected_todo_text_sha256 === null ? null
+            : requireAuthorityStoreId(input.expected_todo_text_sha256, "expected Todo text digest"),
+        }),
         validation_source_provider_revision: input.validation_source_provider_revision == null
           ? null : requireAuthorityStoreId(input.validation_source_provider_revision, "validation source provider revision"),
         validation_declaration_sha256: input.validation_declaration_sha256 == null

loopx/control_plane/coordination/todo_terminal_lifecycle.ts

@@ -96,6 +96,9 @@ type TodoRole = typeof TODO_ROLES[number];
 type CompletionIdentitySource = typeof COMPLETION_IDENTITY_SOURCES[number];
 
 interface CoordinationTodoTerminalLifecycleBaseInput {
+  /** Captured task digest. Null refuses fresh completion without source proof;
+   * omission preserves unscoped callers. Historical receipts recover first. */
+  readonly expected_todo_text_sha256?: string | null;
   readonly review_basis?: {readonly provider_revision: string; readonly registry_sha256: string};
   /** Presence selects source-bound validation; null means no effect issued yet. */
   readonly validation_source_provider_revision?: string | null;
@@ -368,6 +371,10 @@ function validateSuccessorSemantics(
 function normalizeTerminalInput(
   raw: CoordinationTodoTerminalLifecycleInput,
 ): CoordinationTodoTerminalLifecycleInput {
+  if (raw.expected_todo_text_sha256 != null &&
+      (typeof raw.expected_todo_text_sha256 !== "string" || !BARE_SHA256_PATTERN.test(raw.expected_todo_text_sha256))) {
+    throw new AuthorityStoreProtocolError("expected Todo text commitment must be a SHA-256 digest");
+  }
   if (raw.review_basis !== undefined) {
     const basis = canonicalAuthorityObject(raw.review_basis, "terminal review basis");
     if (Object.keys(basis).some(key => !["provider_revision", "registry_sha256"].includes(key)) ||
@@ -500,6 +507,11 @@ function terminalRequestSha(input: CoordinationTodoTerminalLifecycleInput): stri
     // command binds both its approved snapshot and its complete prose intent.
     ...(input.review_basis === undefined ? {} : {review_basis: input.review_basis,
       note: input.note, evidence: input.evidence, reason: input.reason}),
+    // Null means the native Turn lacks source proof. Preserve the historical
+    // fingerprint so an accepted old receipt can still recover before refusal.
+    ...(input.expected_todo_text_sha256 == null ? {} : {
+      expected_todo_text_sha256: input.expected_todo_text_sha256,
+    }),
     ...(input.user_update === undefined ? {} : {user_update: {
       patch: input.user_update.patch, clear_fields: input.user_update.clear_fields,
       planning_intent: input.user_update.planning_intent ?? {},
@@ -1296,6 +1308,20 @@ export async function executeCoordinationTodoTerminalLifecycle(
       "decision_rejection",
     );
   }
+  // Exact operation receipts above recover historical effects before this
+  // fresh-source check. The provider CAS below also prevents a source change
+  // between this comparison and the terminal write.
+  if (input.expected_todo_text_sha256 === null) {
+    return terminalFailure("todo_text_source_unavailable",
+      "Turn completion lacks its admitted full task source; reread the task and run a fresh Turn",
+      {todo_id: input.todo_id}, "decision_rejection");
+  }
+  if (input.expected_todo_text_sha256 !== undefined &&
+      (typeof todo.text !== "string" || createHash("sha256").update(todo.text, "utf8").digest("hex") !== input.expected_todo_text_sha256)) {
+    return terminalFailure("todo_text_changed",
+      "Todo requirements changed after Turn admission; reread the task and run a fresh Turn",
+      {todo_id: input.todo_id}, "decision_rejection");
+  }
   if (todo.archive_state !== "active") {
     return terminalFailure(
       "todo_archived",

loopx/control_plane/quota/turn_envelope.ts

@@ -237,6 +237,8 @@ function selectedTodo(payload: JsonObject, recommendedAction: string | null): Js
   ]) {
     if (source[field] !== null && source[field] !== undefined) compact[field] = source[field];
   }
+  const context = object(object(object(payload.interaction_contract).agent_channel).work_context);
+  if (context.selected_todo_text_verified === true) compact.text_source_verified = true;
   // Work declarations can end in acceptance/stop conditions. Keep their exact
   // text; a display prefix or case-folded recommendation cannot stand in for it.
   const rendered = scalarString(source.text, "selected_todo.text");
@@ -743,7 +745,10 @@ function actionProjection(payload: JsonObject, protocolActionFields: JsonObject)
   );
   const context = object(interaction.agent_context);
   if (Object.keys(context).length > 0) projection.agent_context = structuredClone(context);
-  const workContext = object(object(interaction.agent_channel).work_context);
+  // Source verification belongs to the selected task above. Preserve one carrier while
+  // retaining the existing context elision and historical packet signatures.
+  const {selected_todo_text_verified: _taskTextVerified, ...workContext} =
+    object(object(interaction.agent_channel).work_context);
   const selectedTodoContextIsAlreadyInAction = workContext.complete === true
     && workContext.selected_todo_ref === "selected_todo"
     && Object.keys(workContext).every(field =>

loopx/control_plane/todos/mutation_api.py

@@ -449,6 +449,8 @@ def complete_goal_todo(
     completion_turn_key: str | None = None,
     completion_identity_source: str | None = None,
     terminal_review_basis: Mapping[str, Any] | None = None,
+    expected_todo_text_sha256: str | None = None,
+    require_captured_todo_text: bool = False,
     completion_delivery_workspace: Mapping[str, Any] | None = None,
     completion_validation_workspace_path: Path | None = None,
     task_lease_idempotency_key: str | None = None,

loopx/control_plane/todos/provider_terminal_lifecycle.py

@@ -125,6 +125,8 @@ def route_terminal_call(command: str, call: Mapping[str, Any]) -> dict[str, Any]
             call.get("completion_identity_source") if complete else None
         ),
         review_basis=call.get("terminal_review_basis"),
+        expected_todo_text_sha256=call.get("expected_todo_text_sha256"),
+        require_captured_todo_text=call.get("require_captured_todo_text") is True,
         completion_delivery_workspace=(
             call.get("completion_delivery_workspace") if complete else None
         ),
@@ -275,6 +277,8 @@ def terminal_canonical_todo_if_promoted(
     self_merged: bool,
     dry_run: bool,
     review_basis: Mapping[str, Any] | None = None,
+    expected_todo_text_sha256: str | None = None,
+    require_captured_todo_text: bool = False,
     project: Path | None = None,
     state_file: Path | None = None,
 ) -> dict[str, Any] | None:
@@ -356,6 +360,8 @@ def terminal_canonical_todo_if_promoted(
         request = {
             "schema_version": _TERMINAL_REQUEST_SCHEMA,
             **({"review_basis": dict(review_basis)} if review_basis is not None else {}),
+            **({"expected_todo_text_sha256": expected_todo_text_sha256}
+               if expected_todo_text_sha256 is not None or require_captured_todo_text else {}),
             "validation_source_provider_revision": None,
             "runtime_root": str(runtime_root.expanduser().resolve(strict=False)),
             "goal_id": goal_id,

loopx/control_plane/work_items/interaction_contract.ts

@@ -426,6 +426,7 @@ export function projectInteractionWorkContext(request: JsonObject): JsonObject {
   const pending: JsonObject[] = [], sources: JsonObject[] = [], failures: JsonObject[] = [];
   let selectedTodoRef = false;
   let selectedTodoAuthority: string | null = null;
+  let selectedTodoTextVerified = false;
   for (const value of request.required_reads) {
     const read = requireJsonObject(value, "required read");
     const matches = results.filter(result => result.command === read.command);
@@ -472,6 +473,12 @@ export function projectInteractionWorkContext(request: JsonObject): JsonObject {
       // full read instead of guessing which prose is current Goal authority.
       pending.push(read);
     } else {
+      if (read.source === "selected_todo" && typeof selectedTodoRecord?.text === "string") {
+        // Bind completion to the admitted full source, before body deduplication
+        // or display compaction. Provider-wide revisions also change for notes
+        // and unrelated work; those do not replace this task's requirements.
+        selectedTodoTextVerified = true;
+      }
       if (read.source === "selected_todo" && typeof selected.text === "string"
           && selectedTodoRecord?.text === selected.text
           && selectedTodoRecord?.todo_id === selected.todo_id) {
@@ -518,6 +525,7 @@ export function projectInteractionWorkContext(request: JsonObject): JsonObject {
   const includeUserTodos = users && !users.error_code
     && (!Array.isArray(users.todos) || users.todos.length > 0 || users.authority_read != null);
   return {required_reads: pending, work_context: {complete: failures.length === 0,
+    ...(selectedTodoTextVerified ? {selected_todo_text_verified: true} : {}),
     ...(selectedTodoRef ? {selected_todo_ref: "selected_todo"} : {}),
     ...(selectedTodoAuthority ? {selected_todo_authority: selectedTodoAuthority} : {}),
     ...(sources.length ? {sources} : {}),

loopx/presentation/renderers/turn_envelope_markdown.py

@@ -6,7 +6,8 @@
 
 def work_context_lines(context: dict[str, Any]) -> list[str]:
     """Complete source content once; arbitrary source text stays JSON-escaped."""
-    if set(context) <= {"complete", "selected_todo_ref", "selected_todo_authority", "instruction"}:
+    # The machine-only source witness does not add another instruction body.
+    if set(context) <= {"complete", "selected_todo_ref", "selected_todo_authority", "instruction", "selected_todo_text_verified"}:
         lines = [
             "- work_context: "
             f"complete={context.get('complete')} "