diff --git a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-validation-publication-recovery.md b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-validation-publication-recovery.md
new file mode 100644
index 0000000000..55ff876d6d
--- /dev/null
+++ b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-validation-publication-recovery.md
@@ -0,0 +1,20 @@
+# Validated Todo publication recovery
+
+Source: #5007. This closes a local create/revision delivery gap in the native
+mutation path: provider commit could precede private validator publication, so a
+lost response stranded the entire Goal's Markdown projection.
+
+The host now durably prepares immutable, digest-addressed private declarations
+before dispatch. Only the canonical Todo's digest selects authority; prepared
+but rejected content is inert. Projection and completion share this reader,
+with strict legacy sidecar compatibility. TS remains the single owner of Todo
+admission, CAS and operation receipts. `todo add --operation-id` provides exact
+create recovery without count-derived identities or a second Python journal.
+Historical retries cannot revert the current validator.
+
+File/SQLite regression cases cover lost responses, abrupt process exit with
+public CLI recovery, later edits/revisions, corrupt selected declarations and
+preparation failure before commit. These are isolated synthetic stores, not a
+live Goal migration or a PostgreSQL publication test. The change leaves
+provider selection, D1–D3 qualification and cross-host distribution of private
+commands unchanged. See the [caller contract](../../../../reference/canonical-todo-completion-update.md).
diff --git a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-validation-publication-recovery.zh-CN.md b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-validation-publication-recovery.zh-CN.md
new file mode 100644
index 0000000000..7e7497f04f
--- /dev/null
+++ b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-validation-publication-recovery.zh-CN.md
@@ -0,0 +1,15 @@
+# 已验证 Todo 的声明发布恢复
+
+来源：#5007。此交付修复原生创建/验证器修订路径的一个缺口：provider 已提交，
+私有验证声明尚未保存时丢失响应，会使整个 Goal 的 Markdown 投影无法重建。
+
+宿主现在先持久保存按摘要寻址的不可变私有声明，再发送请求。只有权威 Todo 选择的
+摘要才有效，被拒绝请求留下的内容不授予任何权威。投影和完成验证共享同一读取器，
+并严格兼容旧 sidecar。Todo 准入、CAS 和操作回执仍由 TS 单独持有；
+`todo add --operation-id` 复用该回执实现精确恢复，不再依赖当前 Todo 数量生成身份，
+也不增加 Python 提交日志。历史创建重试不能回滚当前验证器。
+
+隔离的 File/SQLite 合成用例覆盖响应丢失、进程突然退出后的公共 CLI 恢复、后续编辑与
+验证器修订、被选声明损坏，以及提交前保存失败。这不等于活动 Goal 迁移或 PostgreSQL
+私有声明发布验证；provider 默认值、D1–D3 和私有命令跨主机分发保持原边界。
+操作方法见[调用方合同](../../../../reference/canonical-todo-completion-update.md)。
diff --git a/docs/reference/canonical-todo-completion-update.md b/docs/reference/canonical-todo-completion-update.md
index 2ab3d6d9ce..452c2c9a07 100644
--- a/docs/reference/canonical-todo-completion-update.md
+++ b/docs/reference/canonical-todo-completion-update.md
@@ -33,9 +33,9 @@ loopx todo update --goal-id example --todo-id todo_observation \
 
 The TypeScript transaction compares the current declaration digest, commits the
 new digest, monotonic revision and public-safe audit receipt under one provider
-CAS, and rejects terminal, archived or stale edits. The Python boundary stores
-the private command declaration only after provider success and verifies its
-readback. Reuse the same operation id, expected revision and replacement after
+CAS, and rejects terminal, archived or stale edits. The Python boundary durably prepares digest-addressed private command content
+before the provider can reference it. Canonical readback selects that exact
+digest; a lost response does not leave projection waiting for a sidecar. Reuse the same operation id, expected revision and replacement after
 a lost response; a different intent requires a new operation id and a fresh
 read. Validator replacement cannot be combined with another Todo edit.
 
@@ -52,7 +52,8 @@ introduced.
 读取当前 provider revision，再把新命令作为独立的 reviewed edit 提交。TypeScript
 事务在同一次 provider CAS 中核对旧声明摘要，并提交新摘要、单调递增的 revision 和
 公开安全的审计回执；已完成、已归档或基于旧 revision 的修改会被拒绝。Python 边界
-只在 provider 成功后保存私有命令声明，并校验读回结果。丢失响应时复用相同的
+在 provider 提交前持久保存按摘要寻址的私有命令声明；权威读回只选择匹配的摘要，
+因此丢失响应不再阻塞投影。丢失响应时复用相同的
 operation id、expected revision 和替换内容；新的意图必须使用新的 operation id 并
 重新读取。验证器修改不能和其他 Todo 编辑合并提交。
 
@@ -210,3 +211,42 @@ projection delivery before downgrading. Older binaries reject request v3 and
 cannot recover this operation through the old update route. Existing durable
 Todo/lease records, historical receipts and permanent import/export obligations
 are not removed by this change; never revive stale Markdown as authority.
+
+
+## Retrying canonical Todo creation
+
+For an already promoted File/SQLite Goal, provide a stable caller operation id:
+
+```sh
+loopx todo add --goal-id example --role agent --claimed-by agent-a \
+  --text 'Validate the artifact' --operation-id artifact-create-1 \
+  --validation-command-json '["python3","-m","pytest","-q","tests/test_artifact.py"]'
+loopx todo receipt --goal-id example --operation-id artifact-create-1
+```
+
+Retry the same `todo add` intent with the same id after a lost response. The
+TypeScript receipt recovers the original Todo even if its text or validator
+has since changed. Changing the intent under the same id is rejected. An
+omitted id is generated and returned on success or ambiguous timeout; callers
+that must survive process termination should choose the id before dispatch.
+Legacy Markdown creation rejects this option instead of pretending to provide
+canonical idempotency.
+
+Validation content is prepared privately before create/revision dispatch. Its
+presence alone never activates a validator: the authoritative Todo selects its
+exact digest. Corrupt selected content fails closed. Legacy per-Todo sidecars
+remain readable when no digest-addressed content exists. Rejected requests may
+leave unreferenced private content; this change introduces no automatic deletion
+of declarations that historical receipts may still reference. An old create
+retry cannot replace the current canonical validator. This repairs local
+publication recovery, not cross-host distribution of private validation commands.
+
+对已晋升的 File/SQLite Goal，调用方可在 `todo add` 传入稳定的
+`--operation-id`。响应丢失后用同一编号和同一意图重试，TS 回执返回原 Todo，
+不会因 Todo 后来改名、完成或修订验证器而重复创建。相同编号搭配不同意图会被拒绝。
+省略编号时会自动生成并在成功或不确定超时错误中返回；需要应对进程终止的调用方
+应在发送前自行确定编号。旧 Markdown 路径不支持此参数。
+
+私有声明先持久保存，权威摘要再引用它；没有被权威 Todo 引用的内容不会成为验证要求。
+被选中内容损坏时仍拒绝执行。旧 sidecar 可继续读取，历史创建回执不能回滚新验证器。
+此改动不提供私有验证命令的跨主机分发，也不会自动清理未引用内容。
diff --git a/loopx/cli_commands/todo.py b/loopx/cli_commands/todo.py
index 64c3227c27..03016cd39d 100644
--- a/loopx/cli_commands/todo.py
+++ b/loopx/cli_commands/todo.py
@@ -305,6 +305,7 @@ def handle_todo_command(
                 runtime_root_arg=runtime_root_arg,
             )
             payload = add_goal_todo(
+                operation_id=args.operation_id,
                 registry_path=registry_path,
                 runtime_root_arg=runtime_root_arg,
                 goal_id=args.goal_id,
diff --git a/loopx/cli_commands/todo_argument_validation.py b/loopx/cli_commands/todo_argument_validation.py
index db764f9a68..c1153c6750 100644
--- a/loopx/cli_commands/todo_argument_validation.py
+++ b/loopx/cli_commands/todo_argument_validation.py
@@ -514,8 +514,8 @@ def validate_todo_archive_completed_options(args: argparse.Namespace) -> None:
 
 
 def validate_shared_todo_options(args: argparse.Namespace) -> None:
-    if getattr(args, "operation_id", None) and args.todo_command != "receipt":
-        raise ValueError("--operation-id is supported only by todo receipt")
+    if getattr(args, "operation_id", None) and args.todo_command not in {"receipt", "add"}:
+        raise ValueError("--operation-id is supported only by todo receipt and canonical todo add")
     if args.result_file and args.todo_command != "complete":
         raise ValueError("--result-file is supported only by todo complete")
     agent_id_allowed_for_user_authoring = (
diff --git a/loopx/cli_commands/todo_registration.py b/loopx/cli_commands/todo_registration.py
index 89ff07270c..7c6ddf5b00 100644
--- a/loopx/cli_commands/todo_registration.py
+++ b/loopx/cli_commands/todo_registration.py
@@ -57,7 +57,7 @@ def register_todo_command(
     todo_parser.add_argument("--todo-id", help="Structured todo id from status/quota, such as todo_ab12cd34ef56.")
     todo_parser.add_argument(
         "--operation-id",
-        help="For todo receipt, read the exact historical canonical operation after an ambiguous response; this does not grant a retry or lease.",
+        help="For canonical todo add, reuse this identity with unchanged intent after an ambiguous response. For todo receipt, read the exact historical operation; a receipt grants no lease.",
     )
     todo_parser.add_argument(
         "--update-operation-id",
diff --git a/loopx/control_plane/todos/completion_validation.py b/loopx/control_plane/todos/completion_validation.py
index ee06291036..41e9916151 100644
--- a/loopx/control_plane/todos/completion_validation.py
+++ b/loopx/control_plane/todos/completion_validation.py
@@ -451,14 +451,15 @@ def resolve_private_completion_validation_declaration(
         raise ValueError(
             "canonical Todo requires completion validation but omits its digest"
         )
-    # A missing sidecar is an availability case: the digest-bound Markdown or
-    # event projection may rehydrate it below. A present sidecar that fails its
-    # identity or digest checks is corruption/tamper evidence and deliberately
-    # raises instead of falling back, so a second source cannot mask the fault.
+    # The canonical digest selects immutable private content, including after
+    # a lost write response. Older installs may still use a per-Todo sidecar.
+    # Corruption in the selected source fails closed; only absence permits
+    # digest-checked rehydration from a materialized projection below.
     declaration = read_completion_validation_declaration(
         runtime_root=runtime_root,
         goal_id=goal_id,
         todo_id=todo_id,
+        expected_digest=expected,
     )
     if declaration is None:
         source = _materialized_todo_item(
diff --git a/loopx/control_plane/todos/completion_validation_store.py b/loopx/control_plane/todos/completion_validation_store.py
index 6d5c18b4e2..9b76e2fca8 100644
--- a/loopx/control_plane/todos/completion_validation_store.py
+++ b/loopx/control_plane/todos/completion_validation_store.py
@@ -9,6 +9,8 @@
 from typing import Any
 
 from ...registry import atomic_write_json, read_json
+from ...file_lock import exclusive_cross_runtime_file_lock
+from .active_state_editing import fsync_state_directory
 from .completion_validation_projection import (
     completion_validation_declaration,
     completion_validation_declaration_sha256,
@@ -47,7 +49,9 @@ def persist_completion_validation_declaration(
     normalized = completion_validation_declaration(dict(declaration))
     if normalized is None:
         raise ValueError("completion validation declaration is empty")
-    digest = completion_validation_declaration_sha256(normalized)
+    digest = prepare_completion_validation_declaration(
+        runtime_root=runtime_root, goal_id=goal_id, declaration=normalized
+    )
     path = completion_validation_declaration_path(
         runtime_root=runtime_root,
         goal_id=goal_id,
@@ -68,14 +72,64 @@ def persist_completion_validation_declaration(
     return digest
 
 
+def prepare_completion_validation_declaration(
+    *, runtime_root: Path, goal_id: str, declaration: Mapping[str, Any],
+) -> str:
+    """Persist immutable private content before a canonical digest can reference it.
+
+    A prepared blob grants no Todo authority. Only a canonical record selecting
+    its exact digest can consume it. Rejected creates may leave unreferenced blobs.
+    """
+    normalized = completion_validation_declaration(dict(declaration))
+    if normalized is None:
+        raise ValueError("completion validation declaration is empty")
+    digest = completion_validation_declaration_sha256(normalized)
+    path = completion_validation_declaration_path(
+        runtime_root=runtime_root, goal_id=goal_id, todo_id="blobs",
+    ).parent / "blobs" / f"{digest}.json"
+    payload = {"schema_version": "loopx_todo_validation_blob_v0", "goal_id": goal_id,
+               "declaration_sha256": digest, "declaration": normalized}
+    with exclusive_cross_runtime_file_lock(path, operation="prepare_validation_declaration"):
+        if path.exists():
+            if read_json(path) != payload:
+                raise ValueError("prepared validation declaration digest mismatch")
+        else:
+            atomic_write_json(path, payload)
+        # Also re-establish directory durability on an idempotent retry.
+        fsync_state_directory(path)
+    return digest
+
+
+def _read_prepared_declaration(path: Path, goal_id: str, digest: str) -> dict[str, Any] | None:
+    if not re.fullmatch(r"[a-f0-9]{64}", digest):
+        raise ValueError("canonical validation digest must be SHA-256")
+    try:
+        value = read_json(path.parent / "blobs" / f"{digest}.json")
+    except FileNotFoundError:
+        return None
+    if not isinstance(value, Mapping) or not isinstance(value.get("declaration"), dict):
+        raise ValueError("prepared validation declaration is malformed")
+    declaration = completion_validation_declaration(value["declaration"])
+    if (value.get("schema_version") != "loopx_todo_validation_blob_v0"
+            or value.get("goal_id") != goal_id or value.get("declaration_sha256") != digest
+            or declaration is None or completion_validation_declaration_sha256(declaration) != digest):
+        raise ValueError("prepared validation declaration digest or identity mismatch")
+    return declaration
+
+
 def read_completion_validation_declaration(
-    *, runtime_root: Path, goal_id: str, todo_id: str
+    *, runtime_root: Path, goal_id: str, todo_id: str,
+    expected_digest: str | None = None,
 ) -> dict[str, Any] | None:
     path = completion_validation_declaration_path(
         runtime_root=runtime_root,
         goal_id=goal_id,
         todo_id=todo_id,
     )
+    if expected_digest is not None:
+        prepared = _read_prepared_declaration(path, goal_id, expected_digest)
+        if prepared is not None:
+            return prepared
     try:
         value = read_json(path)
     except FileNotFoundError:
@@ -96,6 +150,8 @@ def read_completion_validation_declaration(
     digest = completion_validation_declaration_sha256(normalized)
     if value.get("declaration_sha256") != digest:
         raise ValueError("completion validation declaration store digest mismatch")
+    if expected_digest is not None and digest != expected_digest:
+        raise ValueError("private completion validation declaration does not match canonical Todo digest")
     return normalized
 
 
@@ -114,6 +170,7 @@ def load_completion_validation_declarations(
             runtime_root=runtime_root,
             goal_id=goal_id,
             todo_id=todo_id,
+            expected_digest=str(todo.get("completion_validation_sha256") or ""),
         )
         if declaration is not None:
             loaded[todo_id] = declaration
@@ -125,5 +182,6 @@ def load_completion_validation_declarations(
     "completion_validation_declaration_path",
     "load_completion_validation_declarations",
     "persist_completion_validation_declaration",
+    "prepare_completion_validation_declaration",
     "read_completion_validation_declaration",
 ]
diff --git a/loopx/control_plane/todos/provider_create.py b/loopx/control_plane/todos/provider_create.py
index b57f17d8a0..e4c4c383c3 100644
--- a/loopx/control_plane/todos/provider_create.py
+++ b/loopx/control_plane/todos/provider_create.py
@@ -5,6 +5,8 @@
 from pathlib import Path
 from typing import Any, cast
 from uuid import uuid4
+import hashlib
+import re
 
 from ...agent_registry import registered_agent_ids_from_registry
 from ..coordination.authority_source_capture import authority_registry_source
@@ -14,9 +16,8 @@
     LocalCoordinationAuthorityUnavailable,
     read_canonical_todos_if_promoted,
 )
-from ..effect_runtime import effect_runtime_result
+from ..effect_runtime import EffectRuntimeResponseAmbiguous, effect_runtime_result
 from .contract import (
-    build_todo_id,
     normalize_todo_metadata_for_write,
     normalize_todo_task_class,
     todo_done_for_status,
@@ -28,6 +29,7 @@
 )
 from .completion_validation_store import (
     persist_completion_validation_declaration,
+    prepare_completion_validation_declaration,
     read_completion_validation_declaration,
 )
 from .provider_projection import settle_canonical_todo_projection
@@ -38,17 +40,19 @@ def create_canonical_todo_if_promoted(
     text: str, status: str, actor_agent_id: str | None,
     claimed_by: str | None, metadata: dict[str, Any], dry_run: bool,
     project: Path | None = None, state_file: Path | None = None,
+    operation_id: str | None = None,
 ) -> dict[str, Any] | None:
     canonical = read_canonical_todos_if_promoted(
         runtime_root=runtime_root, goal_id=goal_id
     )
     if canonical is None:
         return None
-    section = "Agent Todo" if role == "agent" else "User Todo"
-    same_role_count = sum(1 for item in canonical["todos"] if item["role"] == role)
-    todo_id = build_todo_id(
-        role=role, source_section=section, index=same_role_count + 1, text=text
-    )
+    if operation_id is not None and not re.fullmatch(r"[A-Za-z0-9_.:-]+", operation_id):
+        raise ValueError("operation_id must be a non-empty public-safe token")
+    operation_id = operation_id if operation_id is not None else f"todo-create:{uuid4().hex}"
+    # Identity is independent of current Todo count: exact retries still address
+    # the original operation after other creates, edits, completion or archive.
+    todo_id = "todo_" + hashlib.sha256(operation_id.encode("utf-8")).hexdigest()[:24]
     normalized_metadata = normalize_todo_metadata_for_write(metadata)
     validation_source = {
         **normalized_metadata,
@@ -65,6 +69,8 @@ def create_canonical_todo_if_promoted(
     }
     validation_declaration = completion_validation_declaration(validation_source)
     provider_metadata = project_completion_validation_authority(validation_source)
+    # TS owns the commit timestamp; transport retries must not change intent.
+    provider_metadata.pop("updated_at", None)
     todo = {
         "schema_version": "todo_domain_record_v0",
         "todo_id": todo_id,
@@ -83,21 +89,33 @@ def create_canonical_todo_if_promoted(
     }
     with authority_registry_source(registry_path) as registry_source:
         registered = registered_agent_ids_from_registry(registry_path, goal_id)
-    result = effect_runtime_result(
-        "coordination.local_authority.todo_create",
-        {
-            "schema_version": "loopx_local_coordination_todo_create_request_v1",
-            "runtime_root": str(runtime_root.resolve()),
-            "goal_id": goal_id,
-            "todo": todo,
-            "actor_agent_id": actor_agent_id,
-            "registered_agents": registered,
-            "registry_source": registry_source,
-            "operation_id": f"todo-create:{uuid4().hex}",
-            "dry_run": dry_run,
-            "observed_at": now_local(),
-        },
-    )
+    if validation_declaration is not None and not dry_run:
+        prepare_completion_validation_declaration(runtime_root=runtime_root,
+            goal_id=goal_id, declaration=validation_declaration)
+    try:
+        result = effect_runtime_result(
+            "coordination.local_authority.todo_create",
+            {
+                "schema_version": "loopx_local_coordination_todo_create_request_v1",
+                "runtime_root": str(runtime_root.resolve()),
+                "goal_id": goal_id,
+                "todo": todo,
+                "actor_agent_id": actor_agent_id,
+                "registered_agents": registered,
+                "registry_source": registry_source,
+                "operation_id": operation_id,
+                "dry_run": dry_run,
+                "observed_at": now_local(),
+            },
+        )
+    except EffectRuntimeResponseAmbiguous as error:
+        raise LocalCoordinationAuthorityUnavailable(
+            "Todo create may have committed; inspect todo receipt with this operation id, "
+            "then retry the same todo add intent with --operation-id",
+            code="todo_create_response_ambiguous",
+            payload={"goal_id": goal_id, "operation_id": operation_id,
+                     "recovery": {"operation_id": operation_id, "retry_with_same_operation_id": True}},
+        ) from error
     if not isinstance(result, dict) or result.get("status") not in {
         "applied", "recovered", "replayed", "no_change", "planned",
     } or result.get("source_authority") not in LOCAL_AUTHORITY_SOURCES or (
@@ -108,7 +126,7 @@ def create_canonical_todo_if_promoted(
         raise LocalCoordinationAuthorityUnavailable(
             str(payload.get("reason") or "canonical Todo create failed; reread before retry"),
             code=str(payload.get("reason_code") or payload.get("conflict_kind")
-                     or "todo_create_failed"), payload=payload,
+                     or "todo_create_failed"), payload={**payload, "operation_id": operation_id},
         )
     canonical_todo_id = str(result.get("todo_id") or "")
     canonical_todo = result.get("todo")
@@ -127,28 +145,35 @@ def create_canonical_todo_if_promoted(
                 "accepted canonical Todo does not match its private validation declaration",
                 code="todo_create_validation_publication_mismatch",
                 payload={
-                    "source_authority": "file_v0",
+                    "source_authority": result.get("source_authority"),
                     "goal_id": goal_id,
                     "todo_id": canonical_todo_id or None,
                     "provider_status": result.get("status"),
                 },
             )
-        persist_completion_validation_declaration(
-            runtime_root=runtime_root,
-            goal_id=goal_id,
-            todo_id=canonical_todo_id,
-            declaration=validation_declaration,
-        )
+        # A replay describes the original create, not the current validator.
+        # Never deliberately regress the compatibility alias after a revision.
+        current_todo = canonical_todo
+        if result.get("status") in {"recovered", "replayed"}:
+            current = read_canonical_todos_if_promoted(runtime_root=runtime_root, goal_id=goal_id)
+            current_todo = next((item for item in (current or {}).get("todos", [])
+                                 if item.get("todo_id") == canonical_todo_id), {})
+        if current_todo.get("completion_validation_sha256") == expected_digest:
+            persist_completion_validation_declaration(
+                runtime_root=runtime_root, goal_id=goal_id, todo_id=canonical_todo_id,
+                declaration=validation_declaration,
+            )
         if read_completion_validation_declaration(
             runtime_root=runtime_root,
             goal_id=goal_id,
             todo_id=canonical_todo_id,
+            expected_digest=expected_digest,
         ) != validation_declaration:
             raise LocalCoordinationAuthorityUnavailable(
                 "accepted Todo validation declaration failed private-store readback",
                 code="todo_create_validation_publication_readback_mismatch",
                 payload={
-                    "source_authority": "file_v0",
+                    "source_authority": result.get("source_authority"),
                     "goal_id": goal_id,
                     "todo_id": canonical_todo_id,
                     "provider_status": result.get("status"),
@@ -156,6 +181,7 @@ def create_canonical_todo_if_promoted(
             )
     settled = settle_canonical_todo_projection({
         "ok": True,
+        "operation_id": operation_id,
         "goal_id": goal_id,
         "role": role,
         "todo_id": canonical_todo_id or todo_id,
diff --git a/loopx/control_plane/todos/provider_update.py b/loopx/control_plane/todos/provider_update.py
index 1d4b72f2ed..78b3b9d5a6 100644
--- a/loopx/control_plane/todos/provider_update.py
+++ b/loopx/control_plane/todos/provider_update.py
@@ -31,6 +31,7 @@
 )
 from .completion_validation_store import (
     persist_completion_validation_declaration,
+    prepare_completion_validation_declaration,
     read_completion_validation_declaration,
 )
 from .completion_validation_projection import (
@@ -128,6 +129,7 @@ def _publish_completion_validation_revision(
         runtime_root=runtime_root,
         goal_id=goal_id,
         todo_id=todo_id,
+        expected_digest=expected_digest,
     ) != declaration:
         raise LocalCoordinationAuthorityUnavailable(
             "completion validation revision committed but private declaration readback failed",
@@ -215,6 +217,10 @@ def update_canonical_todo_if_promoted(
            if validation_revision is not None else {}),
         **({"monitor_observation": asdict(monitor_observation)} if monitor_observation is not None else {}),
     }
+    if completion_validation_revision is not None and not dry_run:
+        prepare_completion_validation_declaration(
+            runtime_root=runtime_root, goal_id=goal_id, declaration=completion_validation_revision
+        )
     result = effect_runtime_result("coordination.local_authority.todo_update", request)
     completion_validation_executed = False
     if isinstance(result, dict) and result.get("status") == "execute_validation":
diff --git a/loopx/todos.py b/loopx/todos.py
index 005964eab4..fb2165c5a7 100644
--- a/loopx/todos.py
+++ b/loopx/todos.py
@@ -739,6 +739,7 @@ def add_goal_todo(
     project: Path | None = None,
     state_file: Path | None = None,
     dry_run: bool = False,
+    operation_id: str | None = None,
 ) -> dict[str, Any]:
     shadow_runtime_root = effective_runtime_root(registry_path, runtime_root_arg)
     if role not in TODO_SECTION_HEADINGS:
@@ -852,6 +853,7 @@ def add_goal_todo(
         enforce_boundedness=True,
     )
     canonical_create = create_canonical_todo_if_promoted(
+        operation_id=operation_id,
         registry_path=registry_path,
         runtime_root=shadow_runtime_root,
         goal_id=goal_id,
@@ -897,6 +899,8 @@ def add_goal_todo(
     )
     if canonical_create is not None:
         return canonical_create
+    if operation_id is not None:
+        raise ValueError("todo add --operation-id requires promoted canonical authority")
     resolved_project, resolved_state_file = resolve_todo_state_path(
         registry_path=registry_path,
         goal_id=goal_id,
diff --git a/tests/control_plane/canonical_authority_fixture.py b/tests/control_plane/canonical_authority_fixture.py
index 0f008b14ff..a786215d12 100644
--- a/tests/control_plane/canonical_authority_fixture.py
+++ b/tests/control_plane/canonical_authority_fixture.py
@@ -67,3 +67,50 @@ def single_snapshot_page(result: dict, goal_id: str = "goal-a") -> dict:
         ) if key in result},
         "next": None,
     }
+
+
+def promoted_create_fixture(tmp_path: Path, *, provider: str = "file") -> tuple[Path, Path, Path]:
+    from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+    from loopx.control_plane.coordination.coordination_state_contract import (
+        TODO_DOMAIN_READ_RECORD_SCHEMA_VERSION, TODO_DOMAIN_RECORD_FIELDS,
+    )
+
+    runtime_root = tmp_path / "runtime"
+    project = tmp_path / "project"
+    state_file = project / ".codex/goals/goal-a/ACTIVE_GOAL_STATE.md"
+    state_file.parent.mkdir(parents=True)
+    state_file.write_text(
+        "# Goal\n\n## User Todo / Owner Review Reading Queue\n\n"
+        "## Agent Todo\n\n## Completed Work Archive\n",
+        encoding="utf-8",
+    )
+    registry_path = tmp_path / "registry.json"
+    registry_path.write_text(
+        json.dumps(
+            {
+                "schema_version": 1,
+                "common_runtime_root": str(runtime_root),
+                "goals": [
+                    {
+                        "id": "goal-a",
+                        "repo": str(project),
+                        "state_file": ".codex/goals/goal-a/ACTIVE_GOAL_STATE.md",
+                        "coordination": {"registered_agents": ["agent-a"]},
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    projection = build_todo_runtime_shadow_projection(
+        goal_id="goal-a", todos=[], handoff_mode="soft_claim"
+    )
+    projection["todo_read_model"] = {
+        **projection["todo_read_model"],
+        "schema_version": TODO_DOMAIN_READ_RECORD_SCHEMA_VERSION,
+        "contract_fields": list(TODO_DOMAIN_RECORD_FIELDS),
+    }
+    initialize_canonical_authority(
+        runtime_root, "goal-a", projection, state_path=state_file, provider=provider
+    )
+    return registry_path, runtime_root, state_file
diff --git a/tests/control_plane/test_local_coordination_authority.py b/tests/control_plane/test_local_coordination_authority.py
index 89d7f77ef2..d3b531c0fe 100644
--- a/tests/control_plane/test_local_coordination_authority.py
+++ b/tests/control_plane/test_local_coordination_authority.py
@@ -8,7 +8,10 @@
 from threading import Barrier
 
 import pytest
-from canonical_authority_fixture import initialize_canonical_authority, single_snapshot_page
+from canonical_authority_fixture import (
+    initialize_canonical_authority, single_snapshot_page,
+    promoted_create_fixture as _promoted_create_fixture,
+)
 
 from loopx.control_plane.coordination import local_authority as local_authority_module
 from loopx.control_plane.coordination.coordination_state_contract import (
@@ -371,47 +374,6 @@ def _create(method: str, params: dict[str, object]) -> dict[str, object]:
     assert calls[0][0] == "coordination.local_authority.todo_create"
 
 
-def _promoted_create_fixture(tmp_path: Path) -> tuple[Path, Path, Path]:
-    runtime_root = tmp_path / "runtime"
-    project = tmp_path / "project"
-    state_file = project / ".codex/goals/goal-a/ACTIVE_GOAL_STATE.md"
-    state_file.parent.mkdir(parents=True)
-    state_file.write_text(
-        "# Goal\n\n## User Todo / Owner Review Reading Queue\n\n"
-        "## Agent Todo\n\n## Completed Work Archive\n",
-        encoding="utf-8",
-    )
-    registry_path = tmp_path / "registry.json"
-    registry_path.write_text(
-        json.dumps(
-            {
-                "schema_version": 1,
-                "common_runtime_root": str(runtime_root),
-                "goals": [
-                    {
-                        "id": "goal-a",
-                        "repo": str(project),
-                        "state_file": ".codex/goals/goal-a/ACTIVE_GOAL_STATE.md",
-                        "coordination": {"registered_agents": ["agent-a"]},
-                    }
-                ],
-            }
-        ),
-        encoding="utf-8",
-    )
-    projection = build_todo_runtime_shadow_projection(
-        goal_id="goal-a", todos=[], handoff_mode="soft_claim"
-    )
-    projection["todo_read_model"] = {
-        **projection["todo_read_model"],
-        "schema_version": TODO_DOMAIN_READ_RECORD_SCHEMA_VERSION,
-        "contract_fields": list(TODO_DOMAIN_RECORD_FIELDS),
-    }
-    initialize_canonical_authority(
-        runtime_root, "goal-a", projection, state_path=state_file
-    )
-    return registry_path, runtime_root, state_file
-
 
 def test_rejected_validated_create_publishes_no_private_sidecar(
     tmp_path: Path,
diff --git a/tests/control_plane/test_validated_create_recovery.py b/tests/control_plane/test_validated_create_recovery.py
new file mode 100644
index 0000000000..84b12dfd26
--- /dev/null
+++ b/tests/control_plane/test_validated_create_recovery.py
@@ -0,0 +1,201 @@
+"""Lost responses must not orphan validated Todos or roll back newer validators."""
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+
+import pytest
+
+from canonical_authority_fixture import isolate_sqlite_runtime, promoted_create_fixture
+from loopx.control_plane.coordination.local_authority import (
+    LocalCoordinationAuthorityUnavailable, read_canonical_todos_if_promoted,
+)
+from loopx.control_plane.effect_runtime import EffectRuntimeResponseAmbiguous
+from loopx.control_plane.todos import provider_create, provider_update
+from loopx.control_plane.todos.completion_validation_store import (
+    completion_validation_declaration_path, read_completion_validation_declaration,
+    prepare_completion_validation_declaration,
+)
+from loopx.todos import add_goal_todo, update_goal_todo
+
+
+@pytest.fixture(params=["file", "sqlite"])
+def promoted(request, tmp_path, monkeypatch):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    return promoted_create_fixture(tmp_path, provider=request.param)
+
+
+def read(runtime):
+    return read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a")
+
+
+def intent(registry):
+    return dict(registry_path=registry, goal_id="goal-a", role="agent",
+                text="Validate a durable artifact", claimed_by="agent-a", agent_id="agent-a",
+                operation_id="create-durable-artifact", validation_label="artifact check",
+                validation_command_json=json.dumps([sys.executable, "-c", "raise SystemExit(0)"]))
+
+
+def project(registry, runtime):
+    before = read(runtime)
+    result = subprocess.run(
+        [sys.executable, "-m", "loopx.cli", "--registry", str(registry), "--format", "json",
+         "todo", "project-markdown", "--goal-id", "goal-a", "--provider-revision",
+         before["provider_revision"], "--execute"], capture_output=True, text=True, timeout=60,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr
+    assert read(runtime) == before
+
+
+def test_lost_response_replays_exact_create_after_another_edit(promoted, monkeypatch):
+    registry, runtime, state = promoted
+    real = provider_create.effect_runtime_result
+    def lose_response(method, request):
+        real(method, request)
+        raise EffectRuntimeResponseAmbiguous(method, timeout=1)
+    monkeypatch.setattr(provider_create, "effect_runtime_result", lose_response)
+    with pytest.raises(LocalCoordinationAuthorityUnavailable) as error:
+        add_goal_todo(**intent(registry))
+    assert error.value.code == "todo_create_response_ambiguous"
+    assert error.value.payload["operation_id"] == "create-durable-artifact"
+    committed = read(runtime)
+    assert len(committed["todos"]) == 1
+    todo = committed["todos"][0]
+    alias = completion_validation_declaration_path(runtime_root=runtime, goal_id="goal-a", todo_id=todo["todo_id"])
+    assert not alias.exists()
+    declaration = read_completion_validation_declaration(
+        runtime_root=runtime, goal_id="goal-a", todo_id=todo["todo_id"],
+        expected_digest=todo["completion_validation_sha256"],
+    )
+    assert declaration["validation_command_argv"][-1] == "raise SystemExit(0)"
+    project(registry, runtime)  # Recovery works in another process before retrying creation.
+    assert todo["todo_id"] in state.read_text()
+    monkeypatch.setattr(provider_create, "effect_runtime_result", real)
+    update_goal_todo(registry_path=registry, goal_id="goal-a", todo_id=todo["todo_id"],
+                     role="agent", agent_id="agent-a", text="A subsequent work declaration")
+    before_retry = read(runtime)
+    replay = add_goal_todo(**intent(registry))
+    assert replay["status"] == "replayed"
+    assert replay["todo_id"] == todo["todo_id"]
+    assert read(runtime) == before_retry  # No new commit or stale declaration overwrite.
+    with pytest.raises(LocalCoordinationAuthorityUnavailable):
+        add_goal_todo(**{**intent(registry), "text": "Different intent with the same operation"})
+    assert read(runtime) == before_retry
+
+
+def test_create_replay_preserves_revised_validator(promoted):
+    registry, runtime, _ = promoted
+    created = add_goal_todo(**intent(registry))
+    todo_id = created["todo_id"]
+    update_goal_todo(registry_path=registry, goal_id="goal-a", todo_id=todo_id,
+                     role="agent", agent_id="agent-a", update_operation_id="revise-check",
+                     update_expected_provider_revision=read(runtime)["provider_revision"],
+                     validation_command_json=json.dumps([sys.executable, "-c", "raise SystemExit(7)"]),
+                     validation_label="revised artifact check")
+    revised = read(runtime)
+    add_goal_todo(**intent(registry))
+    assert read(runtime) == revised
+    for digest in (None, revised["todos"][0]["completion_validation_sha256"]):
+        stored = read_completion_validation_declaration(
+            runtime_root=runtime, goal_id="goal-a", todo_id=todo_id, expected_digest=digest)
+        assert stored["validation_command_argv"][-1] == "raise SystemExit(7)"
+    project(registry, runtime)
+
+
+def test_revision_lost_response_still_projects_exact_new_declaration(promoted, monkeypatch):
+    registry, runtime, _ = promoted
+    created = add_goal_todo(**intent(registry))
+    real = provider_update.effect_runtime_result
+    def lose_response(method, request):
+        real(method, request)
+        raise EffectRuntimeResponseAmbiguous(method, timeout=1)
+    monkeypatch.setattr(provider_update, "effect_runtime_result", lose_response)
+    with pytest.raises(EffectRuntimeResponseAmbiguous):
+        update_goal_todo(registry_path=registry, goal_id="goal-a", todo_id=created["todo_id"],
+                         role="agent", agent_id="agent-a", update_operation_id="revise-after-loss",
+                         update_expected_provider_revision=read(runtime)["provider_revision"],
+                         validation_command_json=json.dumps([sys.executable, "-c", "raise SystemExit(3)"]))
+    project(registry, runtime)
+    canonical = read(runtime)["todos"][0]
+    stored = read_completion_validation_declaration(runtime_root=runtime, goal_id="goal-a",
+             todo_id=created["todo_id"], expected_digest=canonical["completion_validation_sha256"])
+    assert stored["validation_command_argv"][-1] == "raise SystemExit(3)"
+
+
+def test_prepare_failure_prevents_commit_and_dry_run_publishes_nothing(promoted, monkeypatch):
+    registry, runtime, _ = promoted
+    before = read(runtime)
+    add_goal_todo(**intent(registry), dry_run=True)
+    assert not (runtime / "goals/goal-a/todo-validation-declarations").exists()
+    def fail(**kwargs):
+        raise OSError("private storage unavailable")
+    monkeypatch.setattr(provider_create, "prepare_completion_validation_declaration", fail)
+    with pytest.raises(OSError, match="private storage"):
+        add_goal_todo(**intent(registry))
+    assert read(runtime) == before
+
+
+@pytest.mark.parametrize("corruption", ["content", "identity"])
+def test_selected_blob_corruption_cannot_fall_back_to_alias(promoted, corruption):
+    registry, runtime, _ = promoted
+    todo_id = add_goal_todo(**intent(registry))["todo_id"]
+    todo = read(runtime)["todos"][0]
+    alias = completion_validation_declaration_path(runtime_root=runtime, goal_id="goal-a", todo_id=todo_id)
+    blob = alias.parent / "blobs" / (todo["completion_validation_sha256"] + ".json")
+    if os.name == "posix":
+        assert blob.stat().st_mode & 0o777 == 0o600
+    payload = json.loads(blob.read_text())
+    if corruption == "content":
+        payload["declaration"]["validation_command_argv"] = ["false"]
+    else:
+        payload["goal_id"] = "another-goal"
+    blob.write_text(json.dumps(payload))
+    with pytest.raises(ValueError, match="digest or identity"):
+        read_completion_validation_declaration(runtime_root=runtime, goal_id="goal-a", todo_id=todo_id,
+                                               expected_digest=todo["completion_validation_sha256"])
+
+
+def test_unselected_prepared_content_does_not_change_canonical_authority(promoted):
+    registry, runtime, _ = promoted
+    before = read(runtime)
+    prepare_completion_validation_declaration(runtime_root=runtime, goal_id="goal-a",
+                                             declaration={"validation_command_argv": ["false"]})
+    project(registry, runtime)
+    assert read(runtime) == before
+
+
+def test_process_exit_after_commit_recovers_through_public_cli(promoted):
+    registry, runtime, _ = promoted
+    script = '''
+import json, os, sys
+from pathlib import Path
+from loopx.control_plane.todos import provider_create
+from loopx.todos import add_goal_todo
+real = provider_create.effect_runtime_result
+def exit_after_commit(method, request):
+    real(method, request)
+    os._exit(77)
+provider_create.effect_runtime_result = exit_after_commit
+add_goal_todo(registry_path=Path(sys.argv[1]), goal_id="goal-a", role="agent",
+    text="Survive process exit", claimed_by="agent-a", agent_id="agent-a",
+    operation_id="process-exit-create", validation_command_json='["python3", "-c", "pass"]')
+'''
+    child = subprocess.run([sys.executable, "-c", script, str(registry)],
+                           capture_output=True, text=True, timeout=60)
+    assert child.returncode == 77, child.stdout + child.stderr
+    before = read(runtime)
+    assert len(before["todos"]) == 1
+    project(registry, runtime)
+    retried = subprocess.run([
+        sys.executable, "-m", "loopx.cli", "--registry", str(registry), "--format", "json",
+        "todo", "add", "--goal-id", "goal-a", "--role", "agent", "--text", "Survive process exit",
+        "--claimed-by", "agent-a", "--operation-id", "process-exit-create",
+        "--validation-command-json", '["python3", "-c", "pass"]',
+    ], capture_output=True, text=True, timeout=60)
+    assert retried.returncode == 0, retried.stdout + retried.stderr
+    result = json.loads(retried.stdout)
+    assert result["status"] == "replayed"
+    assert result["todo_id"] == before["todos"][0]["todo_id"]
+    assert read(runtime) == before