diff --git a/docs/reference/canonical-todo-completion-update.md b/docs/reference/canonical-todo-completion-update.md
index 58d1bdcfab..f16705623e 100644
--- a/docs/reference/canonical-todo-completion-update.md
+++ b/docs/reference/canonical-todo-completion-update.md
@@ -386,6 +386,26 @@ that must survive process termination should choose the id before dispatch.
 Legacy Markdown creation rejects this option instead of pretending to provide
 canonical idempotency.
 
+Public creation validates task class, role and User gate scope together through
+the existing typed create-authoring plan before provider dispatch. It no longer
+makes a separate class-only preflight call. The same plan returns normalized
+author and claim identities from one registry snapshot, removing repeated Python
+registration reads; the canonical transaction retains its fresh source fence.
+The standalone Markdown add codec
+retains its class check for callers outside that facade. Priority normalization
+still preserves the create request bound by historical operation receipts;
+removing a redundant check does not change replay identity or authorize a write.
+Requests with several invalid fields still fail; the reported error follows the
+complete create and input-validation order.
+
+Caller retirement must also preserve each field's original request behavior:
+creation compacts text and resolves priority/binding, preserves note bytes, and
+deduplicates capabilities in first-occurrence order. Sorting that list or using
+update-note compaction for an original create changes the receipt digest. Real
+File/SQLite CLI cases retry independently specified v1 requests after later
+canonical edits, checking the original receipt and unchanged newer data. This
+is a caller-compatibility gate, not full writer-retirement qualification.
+
 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
@@ -401,6 +421,18 @@ publication recovery, not cross-host distribution of private validation commands
 省略编号时会自动生成并在成功或不确定超时错误中返回；需要应对进程终止的调用方
 应在发送前自行确定编号。旧 Markdown 路径不支持此参数。
 
+公开创建入口在 provider 调用之前，通过现有 TS 创建规划一并检查任务类别、角色与
+User gate 范围，移除重复的类别预检调用。独立 Markdown 添加 codec 仍为直接调用方
+保留类别检查。完整创建规划从一次注册表快照返回规范化的作者与认领身份，删除
+Python 重复注册读取；canonical 事务仍保留自己的新鲜来源检查。优先级规范化继续
+保持历史创建回执绑定的请求形态，不改变重放身份或
+写入准入。多个字段同时无效时，仍拒绝写入，错误由完整创建检查及输入检查顺序决定。
+
+退役调用方也须保留各字段的原请求行为：创建路径压缩文本、解析优先级与绑定，保留
+note 字节，按首次出现顺序对能力去重。能力排序或将更新时的 note 压缩套到历史创建
+会改变回执摘要。File/SQLite 真实 CLI 回归先提交独立指定的 v1 请求，在后续权威
+修改后重试，核对原回执与新数据保持；这是调用方兼容门，不代表完整 writer 退役。
+
 私有声明先持久保存，权威摘要再引用它；没有被权威 Todo 引用的内容不会成为验证要求。
 被选中内容损坏时仍拒绝执行。旧 sidecar 可继续读取，历史创建回执不能回滚新验证器。
 此改动不提供私有验证命令的跨主机分发，也不会自动清理未引用内容。
diff --git a/loopx/control_plane/todos/authoring_scope.ts b/loopx/control_plane/todos/authoring_scope.ts
index 502b9ebc27..2c30b93870 100644
--- a/loopx/control_plane/todos/authoring_scope.ts
+++ b/loopx/control_plane/todos/authoring_scope.ts
@@ -261,16 +261,23 @@ function validateScope(role: string, taskClass: string | null, scope: Scope, age
 }
 
 function planScope(command: string, role: string, taskClass: string | null, todo: JsonObject,
-  intent: JsonObject, agents: string[], goalId: string): Scope {
+  intent: JsonObject, agents: string[], goalId: string): Scope & {
+    actor_agent_id: string | null; claimed_by: string | null;
+  } {
   const registered = (field: string): string | null => {
     if (!intent[field]) return null;
     const value = normalizeTodoAgent(intent[field], field);
+    if (command === "create" && !agents.length && ["actor_agent_id", "claimed_by"].includes(field)) {
+      fail(`${field}='${value}' cannot be used because goal '${goalId}' ` +
+      "has no coordination.registered_agents list. Register this peer identity first: " +
+      `loopx configure-goal --goal-id ${goalId} --registered-agent ${value} --execute`);
+    }
     if (!value || !agents.includes(value)) fail(`${field}='${value}' is not registered for goal '${goalId}'; registered_agents=${agents.join(", ")}`);
     return value;
   };
   const requestedBound = registered("bound_agent");
   const requestedBlocks = registered("blocks_agent");
-  registered("claimed_by");
+  const claim = registered("claimed_by");
   for (const excluded of (intent.excluded_agents ?? []) as string[]) {
     if (!agents.includes(excluded)) fail(`excluded_agents='${excluded}' is not registered for goal '${goalId}'`);
   }
@@ -311,7 +318,8 @@ function planScope(command: string, role: string, taskClass: string | null, todo
     if (requestedBound || intent.goal_bound) fail("bound_agent and goal_bound are only valid for user todos");
     bound = null; goal = null;
   }
-  return {bound_agent: bound, goal_bound: goal, blocks_agent: blocks, global_gate: global};
+  return {bound_agent: bound, goal_bound: goal, blocks_agent: blocks, global_gate: global,
+    actor_agent_id: actor, claimed_by: claim};
 }
 
 export function planTodoAuthoringScope(value: unknown): JsonObject {
@@ -341,7 +349,11 @@ export function planTodoAuthoringScope(value: unknown): JsonObject {
   if (command === "create" && status === "done") fail("todo add cannot create completed work; add it open and use `loopx todo complete`");
   if (command === "update" && role === "agent" && intent.status && status === "done") fail("agent todo completion must use complete_goal_todo " +
     "(CLI: `loopx todo complete`) so completion policy, successor, and no-follow-up contracts are enforced");
-  const scope = planScope(command ?? "", role, taskClass, todo, intent, agents, string(request.goal_id, "goal_id") ?? "");
+  // Create owns the draft class check as well as the resolved-scope invariant.
+  // Keep its diagnostic before scope inference without a separate caller RPC.
+  if (command === "create") requireTaskClass(role, taskClass, intent.blocks_agent, intent.global_gate);
+  const {actor_agent_id: actor, claimed_by: claim, ...scope} = planScope(
+    command ?? "", role, taskClass, todo, intent, agents, string(request.goal_id, "goal_id") ?? "");
   const exclusions = intent.excluded_agents ?? todo.excluded_agents;
   const ownership = todoOwnershipViolations(role, intent.clear_claim ? null : intent.claimed_by || todo.claimed_by, exclusions);
   if (TODO_OWNERSHIP_INTENT_FIELDS.some(field => intent[field] != null && intent[field] !== false)
@@ -363,6 +375,7 @@ export function planTodoAuthoringScope(value: unknown): JsonObject {
   const effectiveResume = intent.clear_resume_when ? null : resume || existingResume;
   if (status === "deferred" && !effectiveResume) fail("transition to deferred requires --resume-when with a supported condition");
   return {schema_version: TODO_AUTHORING_SCOPE_RESULT_SCHEMA, ...scope, task_class: taskClass,
+    ...(command === "create" ? {actor_agent_id: actor, claimed_by: claim} : {}),
     status, normalized_resume_when: resume, effective_resume_when: effectiveResume,
     clear_user_binding: role !== "user" && Boolean(todo.bound_agent || todo.goal_bound != null)};
 }
diff --git a/loopx/todos.py b/loopx/todos.py
index e6f782ceb1..25a69f7362 100644
--- a/loopx/todos.py
+++ b/loopx/todos.py
@@ -507,12 +507,6 @@ def add_goal_todo(
     shadow_runtime_root = effective_runtime_root(registry_path, runtime_root_arg)
     if role not in TODO_SECTION_HEADINGS:
         raise ValueError("todo role must be one of: user, agent")
-    require_user_todo_task_class(
-        role=role,
-        task_class=task_class,
-        blocks_agent=blocks_agent,
-        global_gate=True if global_gate else None,
-    )
     replan_obligation_id = require_replan_successor_scope(
         role=role,
         task_class=task_class,
@@ -549,17 +543,6 @@ def add_goal_todo(
                 "--validation-timeout-seconds must be between 1 and "
                 f"{completion_validation_module.COMPLETION_VALIDATION_TIMEOUT_MAX_SECONDS}"
             )
-    effective_claimed_by = (
-        require_registered_agent_id(
-            registry_path=registry_path, goal_id=goal_id, agent_id=claimed_by,
-        ) if claimed_by else None
-    )
-    effective_agent_id = (
-        require_registered_agent_id(
-            registry_path=registry_path, goal_id=goal_id, agent_id=agent_id,
-            field="agent_id",
-        ) if agent_id else None
-    )
     registered_agents = registered_agent_ids_from_registry(registry_path, goal_id)
     effective_excluded_agents = (
         require_todo_excluded_agents(excluded_agents)
@@ -569,14 +552,18 @@ def add_goal_todo(
     authoring_scope = plan_todo_authoring_scope(
         command="create", role=role, goal_id=goal_id, registered_agents=registered_agents,
         intent={
-            "task_class": task_class, "status": status, "actor_agent_id": effective_agent_id,
-            "claimed_by": effective_claimed_by, "bound_agent": bound_agent, "goal_bound": goal_bound,
+            "task_class": task_class, "status": status, "actor_agent_id": agent_id,
+            "claimed_by": claimed_by, "bound_agent": bound_agent, "goal_bound": goal_bound,
             "blocks_agent": blocks_agent, "global_gate": global_gate,
             "excluded_agents": effective_excluded_agents, "resume_when": resume_when,
             "task_repository": task_repository, "task_domain": task_domain,
             "capability_binding_ref": capability_binding_ref,
         },
     )
+    # The shared plan already validates and normalizes both identities from
+    # this registry snapshot; the transaction still rechecks its own source.
+    effective_claimed_by = authoring_scope["claimed_by"]
+    effective_agent_id = authoring_scope["actor_agent_id"]
     effective_blocks_agent = authoring_scope["blocks_agent"]
     effective_bound_agent = authoring_scope["bound_agent"]
     effective_goal_bound = authoring_scope["goal_bound"]
diff --git a/tests/control_plane/test_todo_authoring_scope.py b/tests/control_plane/test_todo_authoring_scope.py
index fbe6dbe855..e94457059e 100644
--- a/tests/control_plane/test_todo_authoring_scope.py
+++ b/tests/control_plane/test_todo_authoring_scope.py
@@ -5,7 +5,7 @@
 
 from loopx.control_plane.testing.canary_harness import run_json_cli_result, write_fixture_registry
 from loopx.control_plane.todos.active_state_editing import find_todo_block
-from loopx.todos import add_goal_todo, update_goal_todo
+from loopx.todos import add_goal_todo, add_todo_to_lines, update_goal_todo
 from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
 from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
 
@@ -127,7 +127,7 @@ def test_execution_exclusion_registration_and_claim_conflict_preserve_source(exe
     _, state, cli = execution_exclusion_goal
     create = ("add", "--role", "agent", "--text", "Review the current change",
               "--task-class", "advancement_task",
-              "--claimed-by", "agent-a")
+              "--claimed-by", "\u001cAGENT\u0085A\u001f")
     before = state.read_bytes()
     for excluded in ("unknown-agent", "agent-a", "invalid/token"):
         code, rejected = cli(*create, "--excluded-agent", excluded)
@@ -135,7 +135,7 @@ def test_execution_exclusion_registration_and_claim_conflict_preserve_source(exe
         assert state.read_bytes() == before
         assert cli("list")[1]["todo_count"] == 0
 
-    code, created = cli(*create, "--excluded-agent", " agent-b ", "--excluded-agent", "agent-b")
+    code, created = cli(*create, "--excluded-agent", "\u001cAGENT\u0085B\u001f", "--excluded-agent", "agent-b")
     assert code == 0, created
     todo_id = created["todo_id"]
     code, listed = cli("list", "--todo-id", todo_id)
@@ -211,3 +211,41 @@ def test_create_role_restrictions_refuse_before_source_write(
     code, listed = cli("list")
     assert code == 0, listed
     assert listed["todo_count"] == 0
+
+
+@pytest.mark.parametrize("role, task_class, error", [
+    ("user", None, "user todo requires explicit --task-class"),
+    ("agent", "user_gate", "user_action and user_gate task_class are only valid for --role user"),
+])
+def test_standalone_markdown_codec_keeps_class_admission(role, task_class, error):
+    lines = ["# Goal", "", "## Agent Todo", "", "## User Todo", ""]
+    before = list(lines)
+    with pytest.raises(ValueError, match=error):
+        add_todo_to_lines(lines, role=role, task_class=task_class, text="Decide the next step")
+    assert lines == before
+
+
+@pytest.mark.parametrize("role, task_class, flags, error", [
+    ("user", None, ("--bound-agent", "agent-a"), "user todo requires explicit --task-class"),
+    ("user", "advancement_task", ("--bound-agent", "agent-a"), "user todo requires explicit --task-class"),
+    ("user", "user_action", ("--blocks-agent", "agent-a"), "user_action is non-blocking"),
+    ("user", "user_action", ("--global-gate",), "user_action is non-blocking"),
+    ("agent", "user_gate", (), "user_action and user_gate task_class are only valid for --role user"),
+])
+def test_create_class_rules_reject_before_any_provider_write(
+    execution_exclusion_goal, role, task_class, flags, error,
+):
+    _, state, cli = execution_exclusion_goal
+    before = state.read_bytes()
+    code, initial = cli("list")
+    assert code == 0, initial
+    declaration = ("--task-class", task_class) if task_class else ()
+    code, rejected = cli("add", "--role", role, "--text", "Decide the next step",
+                         *declaration, *flags)
+    assert code != 0, rejected
+    assert error in rejected["error"], rejected
+    assert state.read_bytes() == before
+    code, listed = cli("list")
+    assert code == 0, listed
+    assert listed["todo_count"] == 0
+    assert listed.get("authority_read") == initial.get("authority_read")
diff --git a/tests/control_plane/test_validated_create_recovery.py b/tests/control_plane/test_validated_create_recovery.py
index 35b8c84080..9cd184cfdb 100644
--- a/tests/control_plane/test_validated_create_recovery.py
+++ b/tests/control_plane/test_validated_create_recovery.py
@@ -1,6 +1,7 @@
 """Lost responses must not orphan validated Todos or roll back newer validators."""
 from __future__ import annotations
 
+import hashlib
 import json
 import os
 import subprocess
@@ -14,6 +15,8 @@
     LocalCoordinationAuthorityUnavailable, read_canonical_todos_if_promoted,
 )
 from loopx.control_plane.effect_runtime import EffectRuntimeResponseAmbiguous
+from loopx.control_plane.effect_runtime import effect_runtime_result
+from loopx.control_plane.coordination.authority_source_capture import authority_registry_source
 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,
@@ -39,6 +42,106 @@ def intent(registry):
                 validation_command_json=json.dumps([sys.executable, "-c", "raise SystemExit(0)"]))
 
 
+@pytest.mark.parametrize("options, normalized", [
+    (
+        {"role": "user", "task_class": "user_action", "priority": "P3",
+         "text": "Verify\u001c recovery", "note": "Keep\u001f this"},
+        {"role": "user", "task_class": "user_action", "bound_agent": "agent-a",
+         "text": "[P3] Verify recovery", "priority": "P3", "title": "Verify recovery",
+         "note": "Keep\u001f this"},
+    ),
+    (
+        {"role": "user", "task_class": "user_action", "text": "[p3] Verify recovery"},
+        {"role": "user", "task_class": "user_action", "bound_agent": "agent-a",
+         "text": "[P3] Verify recovery", "priority": "P3", "title": "Verify recovery"},
+    ),
+    (
+        {"role": "user", "task_class": "user_gate", "text": "Choose the next step"},
+        {"role": "user", "task_class": "user_gate", "bound_agent": "agent-a",
+         "blocks_agent": "agent-a", "text": "Choose the next step"},
+    ),
+    (
+        {"role": "user", "task_class": "user_action", "goal_bound": True,
+         "text": "Read the result"},
+        {"role": "user", "task_class": "user_action", "goal_bound": True,
+         "text": "Read the result"},
+    ),
+    (
+        {"role": "agent", "task_class": "advancement_task", "action_kind": "implement",
+         "claimed_by": "agent-a", "required_capabilities": ["shell", "filesystem_read", "shell"],
+         "text": "Deliver the artifact"},
+        {"role": "agent", "task_class": "advancement_task", "action_kind": "implement",
+         "claimed_by": "agent-a", "required_capabilities": ["shell", "filesystem_read"],
+         "text": "Deliver the artifact"},
+    ),
+    (
+        {"role": "agent", "action_kind": "verify", "claimed_by": "agent-a",
+         "priority": "P1", "text": "Check the artifact"},
+        {"role": "agent", "task_class": "advancement_task", "action_kind": "verify",
+         "claimed_by": "agent-a", "priority": "P1", "title": "Check the artifact",
+         "text": "[P1] Check the artifact"},
+    ),
+    (
+        {"role": "user", "task_class": "user_action", "text": "Read the artifact",
+         "agent_id": "\u001cAGENT\u0085A\u001f"},
+        {"role": "user", "task_class": "user_action", "text": "Read the artifact",
+         "bound_agent": "agent-a"},
+    ),
+    (
+        {"role": "agent", "text": "Validate the artifact", "claimed_by": "\u001cAGENT\u0085A\u001f"},
+        {"role": "agent", "text": "Validate the artifact", "claimed_by": "agent-a",
+         "task_class": "advancement_task"},
+    ),
+])
+def test_public_cli_replays_prior_normalized_create_without_replacing_new_data(promoted, options, normalized):
+    registry, runtime, _ = promoted
+    operation = "prior-normalized-create"
+    todo_id = "todo_" + hashlib.sha256(operation.encode()).hexdigest()[:24]
+    # Independent compatibility inputs, not a snapshot generated by the current
+    # facade. Creation preserves note bytes and first-occurrence capability order;
+    # text/priority/binding have their own compatibility normalization. Persist
+    # the old v1 request through the real native transaction;
+    # the public CLI must still address its exact receipt after a later edit.
+    todo = {"schema_version": "todo_domain_record_v0", "todo_id": todo_id,
+            "status": "open", "done": False, "archive_state": "active", **normalized}
+    with authority_registry_source(registry) as source:
+        committed = effect_runtime_result("coordination.local_authority.todo_create", {
+            "schema_version": "loopx_local_coordination_todo_create_request_v1",
+            "runtime_root": str(runtime), "goal_id": "goal-a", "todo": todo,
+            "actor_agent_id": "agent-a", "registered_agents": ["agent-a"],
+            "registry_source": source, "operation_id": operation, "dry_run": False,
+            "observed_at": "2026-01-01T00:00:00Z",
+        })
+    assert committed["status"] == "applied", committed
+    identity = {"goal_id": "goal-a", "todo": todo, "actor_agent_id": "agent-a", "dry_run": False}
+    expected_digest = hashlib.sha256(json.dumps(identity, ensure_ascii=False, sort_keys=True,
+                                                separators=(",", ":")).encode()).hexdigest()
+    receipt = committed["original_receipt"]
+    assert receipt["request_sha256"] == expected_digest
+    update_goal_todo(registry_path=registry, goal_id="goal-a", todo_id=todo_id,
+                     role=options["role"], agent_id="agent-a", note="New data after the original create")
+    current = read(runtime)
+    assert current["todos"][0]["note"] == "New data after the original create"
+    command = [sys.executable, "-m", "loopx.cli", "--registry", str(registry), "--format", "json",
+               "todo", "add", "--goal-id", "goal-a", "--operation-id", operation]
+    for field, value in options.items():
+        flag = "--" + field.replace("_", "-")
+        if value is True:
+            command.append(flag)
+        else:
+            for entry in value if isinstance(value, list) else [value]:
+                # CLI's capability option is singular and repeatable.
+                command.extend(["--required-capability" if field == "required_capabilities" else flag, entry])
+    if options["role"] == "user" and "agent_id" not in options:
+        command.extend(["--agent-id", "agent-a"])
+    retried = subprocess.run(command, capture_output=True, text=True, timeout=60)
+    assert retried.returncode == 0, retried.stdout + retried.stderr
+    replay = json.loads(retried.stdout)
+    assert replay["status"] == "replayed", replay
+    assert replay["original_receipt"] == receipt
+    assert read(runtime) == current
+
+
 def project(registry, runtime):
     before = read(runtime)
     result = subprocess.run(
diff --git a/tests/control_plane_ts/todo_authoring_scope.test.ts b/tests/control_plane_ts/todo_authoring_scope.test.ts
index a231024b98..54d29249e8 100644
--- a/tests/control_plane_ts/todo_authoring_scope.test.ts
+++ b/tests/control_plane_ts/todo_authoring_scope.test.ts
@@ -41,6 +41,25 @@ test("explicit all-agent intent is accepted without turning authorship into lane
   }
 });
 
+test("create returns normalized author and claim without granting execution", () => {
+  const raw = "\u001cAGENT\u0085A\u001f";
+  const result = plan({actor_agent_id: raw, claimed_by: raw}, {role: "agent"});
+  assert.equal(result.actor_agent_id, "agent-a");
+  assert.equal(result.claimed_by, "agent-a");
+  assert.equal(result.bound_agent, null);
+  const anonymous = plan({}, {role: "agent"});
+  assert.equal(anonymous.actor_agent_id, null);
+  assert.equal(anonymous.claimed_by, null);
+  for (const field of ["actor_agent_id", "claimed_by"]) {
+    assert.throws(() => plan({[field]: "missing-agent"}, {role: "agent"}), /not registered/);
+    assert.throws(() => plan({[field]: raw}, {role: "agent", registered_agents: []}),
+      /loopx configure-goal --goal-id goal-a --registered-agent agent-a --execute/);
+  }
+  const edited = plan({}, {command: "update", role: "agent"});
+  assert.equal(Object.hasOwn(edited, "actor_agent_id"), false);
+  assert.equal(Object.hasOwn(edited, "claimed_by"), false);
+});
+
 test("explicit continuation and gate constraints cannot silently overwrite each other", () => {
   for (const command of ["create", "update"]) for (const intent of [
     {global_gate: true, bound_agent: "agent-a"},
@@ -81,6 +100,15 @@ test("authoring plan cannot grant terminal, executor, or non-user gate semantics
   assert.equal(plan({}, {command: "update", todo: {status: "done", task_class: "legacy"}}).status, "done");
 });
 
+test("create reports draft class violations before inferring response scope", () => {
+  assert.throws(() => plan({bound_agent: "agent-a"}), /user todo requires explicit --task-class/);
+  for (const flags of [{global_gate: true}, {blocks_agent: "agent-a"}]) {
+    assert.throws(() => plan({task_class: "user_action", ...flags}), /user_action is non-blocking/);
+  }
+  assert.throws(() => plan({task_class: "user_gate"}, {role: "agent"}),
+    /user_action and user_gate task_class are only valid for --role user/);
+});
+
 test("deferred state requires a supported condition and clears remain explicit", () => {
   const todo = {task_class: "advancement_task", status: "deferred", resume_when: "capacity_available:network"};
   assert.equal(plan({}, {command: "update", role: "agent", todo}).effective_resume_when, todo.resume_when);