refactor(todo): reuse typed create validation and preserve replay identity

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 可继续读取，历史创建回执不能回滚新验证器。
 此改动不提供私有验证命令的跨主机分发，也不会自动清理未引用内容。

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)};
 }

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"]

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,15 +127,15 @@ 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)
         assert code != 0, rejected
         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")