fix(delegation): reuse canonical independent-task validation and member continuation

docs/architecture/rfcs/loopx-overall-roadmap-v0.md

@@ -499,6 +499,14 @@ coordinators and ordinary members use the same grant contract. Disabled stdio
 servers retain their original five non-executing tools. Configuration files
 compact provider launch arguments without changing default executor selection.
 
+Independent work outside owner-selected acceptance now uses the canonical
+Todo's explicit completion validator through the same TS validation plan.
+Covered work still requires its current owner association and every applicable
+check. Member completion keeps the Goal active, then resumes only the original
+Turn's settlement; it does not assert terminal no-follow-up. File/SQLite CLI
+and local host regressions qualify this boundary, not real-model research,
+requester synthesis or Lark parity. No binding grant or provider is activated.
+
 The [synthetic research example](../../../examples/managed-research-team/README.md)
 uses a local lead, two DSH members and two Ark members. One cloud reviewer adopts
 local analysis; another Ark member delegates to DSH before returning to local

docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md

@@ -395,6 +395,12 @@ P0 首批是负责人路由和真实 2–3-worker 协调：两轮并行汇合、
 委派逻辑。主协调员与普通成员使用同一授权合同；未启用执行配置的 stdio 服务保持
 原有五个非执行工具。文件形式的 provider 配置缩短启动参数，不改变默认执行器。
 
+owner 所选验收范围外的独立工作，现可通过同一 TS 校验计划使用规范 Todo 显式声明的
+完成校验。范围内仍须具备当前 owner 关联，且所有适用校验都须通过。成员完成保留
+Goal active，再仅恢复原 Turn 的结算，不声明 terminal no-follow-up。File/SQLite CLI
+及本地 host 回归只验收该边界，不代表真实模型投研、请求方综合或 Lark 等价；不激活
+任何 binding grant 或 provider。
+
 [合成投研示例](../../../examples/managed-research-team/README.md)由本地主 Agent
 组织两个 DSH 和两个 Ark 成员：云端核验员采用本地分析，另一 Ark 成员继续委派
 DSH 后向本地主 Agent 返回。五个稳定预授权任务一次绑定精确验收；Turn 验证与普通

docs/reference/local-delegation.md

@@ -7,8 +7,15 @@ existing Turn entrypoint; there is no steward-specific scheduler or task store.
 
 ## Activate
 
-First register the participating Agents and bind the intended canonical Todos
-to [owner-configured acceptance](goal-acceptance-observations.md). Prepare an
+First register the participating Agents and give each intended canonical Todo
+an explicit validation basis. Work covered by
+[owner-configured acceptance](goal-acceptance-observations.md) must retain its
+current owner binding. Independent work outside that scope (or with Goal
+acceptance disabled) instead requires its own canonical Todo completion
+validator, declared through the existing `todo add --validation-command-json`
+entrypoint. A missing or stale owner association never falls back to that
+validator; a Todo validator supplements owner criteria when both apply.
+Prepare an
 operator-owned JSON file **outside every delegated member workspace**. A
 coordinator may keep it as an ignored file under its Goal project at
 `.loopx/config/delegations.json`:
@@ -36,6 +43,35 @@ select `generic-cli`, `fresh`, and the optional adapter's `--config` invocation.
 Profiles, executables, workspace isolation and credential custody remain the
 operator's responsibility. No model tool accepts those values.
 
+Inspection, pre-launch admission, Turn validation and returned-artifact readback
+consume this same basis. Private commands must match the canonical Todo's
+declaration digest; verifier files declared by Goal acceptance are checked
+before and after execution. The ordinary Todo digest pins the command, not
+undeclared script dependencies. A successful validator still needs canonical
+completion, unchanged artifacts and receiver adoption. Inspection never starts
+work or configures owner acceptance. Disable by removing the exact binding
+from the operator configuration; existing operations retain their history and
+cannot re-execute or return accepted evidence under a revoked grant.
+
+Member completion uses the ordinary active-Goal continuation, including legacy
+non-hard-lease routes. It does not declare terminal `no_followup` for a
+requester-owned synthesis. This lets controller validation finish the Todo
+before resuming only the original Turn's settlement; the host is not rerun.
+Explicit terminal closeout still requires matching writeback/spend receipts.
+
+中文：受 Goal 验收范围覆盖的 Todo 保留当前 owner 关联；范围外的独立任务，或未启用
+Goal 验收的任务，必须通过既有 `todo add --validation-command-json` 声明规范 Todo
+完成校验。范围内关联缺失或过期不能退回普通校验；两者同时存在时须全部通过。
+预检、启动前准入、Turn 校验和结果读回复用同一依据，私有命令必须匹配规范声明
+摘要。普通 Todo 摘要固定命令，不固定未声明的脚本依赖；Goal 声明的校验文件在
+执行前后核对。校验通过仍不等于规范完成、产物未变或接收方采纳。预检不启动工作、
+不配置 owner 验收；移除原配置中的精确 binding 即撤销 grant，保留历史但拒绝重新
+执行或返回已撤权任务的有效结果。
+
+成员完成沿用普通 active-Goal 继续状态，旧的非 hard-lease 路径也如此；不会为仍由
+请求方负责的汇总声明 terminal `no_followup`。因此可以先通过 controller 校验完成
+Todo，再仅恢复原 Turn 的结算，不重跑 host。显式终结仍须具备匹配的写回和扣额回执。
+
 A Codex binding launches an independent, resumable Codex Agent Session through
 the same governed Turn path. Pin both fields when the worker must use an exact
 profile:

loopx/collaboration_mcp.py

@@ -26,9 +26,7 @@
     from mcp.server.fastmcp import FastMCP
 
 from .file_lock import exclusive_file_lock, LockAcquisitionPolicy, LockAcquireTimeoutError
-from .todos import list_goal_todos
 from .control_plane.effect_runtime import effect_runtime_result, EffectRuntimeRemoteError
-from .control_plane.goals.acceptance import inspect_goal_acceptance, validate_goal_task_acceptance, goal_task_validation_files_current
 from .control_plane.coordination.local_authority import local_authority_is_promoted
 from .control_plane.todos.handoff_mode import show_goal_handoff_mode
 from .control_plane.turn_driver.journal_store import (
@@ -40,7 +38,7 @@
 from .control_plane.collaboration.inbox import _hash, _read, _write, _root, _receipt
 from .control_plane.collaboration.peers import return_result
 from .control_plane.collaboration.inbox import acknowledge, _entry, normalize_request
-from .control_plane.collaboration import delegation_results
+from .control_plane.collaboration import delegation_results, delegation_validation
 from .control_plane.collaboration.peers import (
     _goal,
     consume_return,
@@ -303,10 +301,7 @@ def inspect(self, binding_id: str) -> dict:
         if not Path(binding["workspace"]).is_dir():
             raise ValueError("delegation workspace unavailable")
         try:
-            acceptance = inspect_goal_acceptance(registry_path=self.registry, goal_id=self.goal_id,
-                                                  runtime_root=str(self.root))
-            files_current = goal_task_validation_files_current(registry_path=self.registry,
-                runtime_root=str(self.root), goal_id=self.goal_id, agent_id=binding["agent_id"], todo_id=binding["todo_id"])
+            acceptance = delegation_validation.capture(self, binding)
         except (OSError, ValueError) as exc:
             # Authority admission is a readiness observation, not a reason for
             # inspection to invent a provider launch or collapse into a raw CLI error.
@@ -363,19 +358,14 @@ def inspect(self, binding_id: str) -> dict:
         preview = self._cli(binding, *arguments)
         if preview.get("status") != "preview":
             raise ValueError(f"delegation Turn preflight unavailable: {preview.get('error') or preview.get('status')}")
-        current = inspect_goal_acceptance(registry_path=self.registry, goal_id=self.goal_id,
-                                          runtime_root=str(self.root))
-        if (acceptance != current or self.binding(binding_id, require_active=True) != binding
-                or files_current != goal_task_validation_files_current(registry_path=self.registry,
-                    runtime_root=str(self.root), goal_id=self.goal_id,
-                    agent_id=binding["agent_id"], todo_id=binding["todo_id"])):
+        current = delegation_validation.capture(self, binding)
+        if acceptance != current or self.binding(binding_id, require_active=True) != binding:
             raise ValueError("delegation preflight source changed; retry inspection")
-        task = next((row for row in (acceptance.get("goal_acceptance_contract") or {}).get("tasks", [])
-                     if row.get("todo_id") == binding["todo_id"]), None)
         return effect_runtime_result("collaboration.delegation.preflight", {
             "binding": {key: binding[key] for key in ("id", "agent_id", "todo_id")},
             "authority": {"ready": True, "reason": None},
-            "preview": preview, "acceptance": task, "validation_files_current": files_current,
+            "preview": preview, "acceptance": acceptance["plan"],
+            "validation_files_current": acceptance["files_current"],
         })
 
     def start(self, binding_id: str, operation_id: str, brief: dict,
@@ -592,22 +582,12 @@ def _cli(self, binding: dict, *args: str, timeout: int = 60) -> dict:
             )
         return value
 
-    def _validate(self, binding: dict) -> None:
-        value = validate_goal_task_acceptance(registry_path=self.registry, runtime_root=str(self.root),
-            goal_id=self.goal_id, agent_id=binding["agent_id"], todo_id=binding["todo_id"])
-        if not value["passed"]:
-            raise ValueError("delegation task acceptance rejected")
+    def _validate(self, binding: dict) -> dict:
+        return delegation_validation.validate(self, binding)
 
     def _accepted(self, binding: dict) -> list[dict]:
-        self._validate(binding)
-        todos = list_goal_todos(registry_path=self.registry, goal_id=self.goal_id, runtime_root_arg=str(self.root))
-        basis = inspect_goal_acceptance(registry_path=self.registry, goal_id=self.goal_id, runtime_root=str(self.root))
-        if todos.get("authority_read", {}).get("provider_revision") != basis.get("provider_revision"):
-            raise ValueError("delegation canonical snapshot changed; retry readback")
-        todo = next((row for row in todos["todos"] if row["todo_id"] == binding["todo_id"]), {})
-        guard = next((row for row in basis["goal_acceptance_contract"]["tasks"]
-                      if row["todo_id"] == binding["todo_id"]), {})
-        if not todo.get("done") or todo.get("status") != "done" or guard.get("state") != "ready":
+        validation = self._validate(binding)
+        if not validation["plan"]["canonical_done"]:
             raise ValueError("delegation requires current canonical completion")
         workspace = Path(binding["workspace"]).resolve()
         artifacts = []
@@ -867,8 +847,10 @@ def _complete_delegated_todo(self, row: dict, binding: dict) -> None:
                 "--task-lease-expected-version",
                 str(lease["version"]),
             ]
-        else:
-            arguments.append("--no-follow-up")
+        # A bounded member task returns to its requester; it is not terminal
+        # Goal intent. Ordinary completion may precede its original Turn's
+        # accounting (controller validation requires that order). Do not add
+        # no-follow-up, which correctly requires already-settled receipts.
         result = self._cli(binding, *arguments)
         if result.get("ok") is not True:
             raise ValueError(
@@ -881,6 +863,9 @@ def _execute(self, path: Path, row: dict, binding: dict) -> None:
         execution = self._execution_arguments(binding, row["identity"]["operation_id"])
         try:
             if row["status"] == "prepared":
+                acceptance = delegation_validation.capture(self, binding)
+                if acceptance["plan"]["state"] != "ready" or not acceptance["files_current"]:
+                    raise ValueError("delegation task acceptance rejected before host launch")
                 row["turn_instance_id"] = self._turn_instance_id(row)
                 self._write_delegation_bootstrap(row, binding)
                 self._acquire_delegation_lease(path, row, binding)

loopx/control_plane/collaboration/delegation.ts

@@ -3,13 +3,60 @@
 import type {JsonObject} from "../effect_program.ts";
 import {requireJsonObject} from "../runtime_decode.ts";
 import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
+import {canonicalAuthoritySha256} from "../coordination/authority_store_codec.ts";
+import {acceptanceValidationEffects, type AcceptanceCompletionRequirements} from "../goals/acceptance_contract.ts";
+import {normalizeTodoCompletionValidationDeclaration} from "../todos/completion_validation_declaration.ts";
 
 function requireThat(ok: unknown, message: string): asserts ok {
   if (!ok) throw new EffectRuntimeRequestError(message);
 }
 function text(value: unknown): value is string {
   return typeof value === "string" && value.length > 0 && value.length <= 4096;
 }
+
+/** The canonical acceptance reader resolves owner scope before this plan.
+ * A null requirement is out of scope/disabled, never an unbound-task fallback.
+ * Private declarations must match the current Todo authority, also on readback. */
+export function delegationValidationPlan(params: JsonObject): JsonObject {
+  const binding = requireJsonObject(params.binding, "delegation binding");
+  const basis = requireJsonObject(params.basis, "canonical validation basis");
+  const todo = requireJsonObject(basis.todo, "canonical delegation Todo");
+  requireThat(basis.status === "loaded" && text(basis.provider_revision)
+    && todo.todo_id === binding.todo_id, "delegation requires a current matching canonical Todo");
+  requireThat(Object.hasOwn(basis, "completion_requirements"), "canonical acceptance scope required");
+  const requirements = basis.completion_requirements === null ? null
+    : requireJsonObject(basis.completion_requirements, "canonical acceptance requirements");
+  if (requirements !== null) requireThat(requirements.todo_id === todo.todo_id
+    && Array.isArray(requirements.criteria) && requirements.criteria.length > 0,
+  "delegation requires matching owner acceptance criteria");
+  const unavailable = (reason: string) => ({todo_id: todo.todo_id, state: "unbound",
+    source: null, reason, effects: [], canonical_done: false});
+  const effects: JsonObject[] = requirements === null ? []
+    : acceptanceValidationEffects(requirements as AcceptanceCompletionRequirements, todo)
+      .map(row => ({...requireJsonObject(row.effect, "acceptance validation effect"), criterion_id: row.criterion_id}));
+  if (todo.completion_validation_required === true) {
+    if (params.declaration === null) return unavailable("completion_validation_declaration_unavailable");
+    const declaration = requireJsonObject(params.declaration, "private validation declaration");
+    const normalized = normalizeTodoCompletionValidationDeclaration(declaration, {
+      strict_fields: true, require_command: true, require_canonical_input: true,
+    });
+    if (!normalized.ok || canonicalAuthoritySha256(declaration) !== todo.completion_validation_sha256)
+      return unavailable("completion_validation_declaration_mismatch");
+    effects.push({kind: "caller_validation", validation_command: normalized.value.validation_command,
+      validation_argv: normalized.value.validation_command_argv,
+      validation_label: normalized.value.validation_label,
+      validation_timeout_seconds: normalized.value.validation_timeout_seconds,
+      validation_declaration_sha256: todo.completion_validation_sha256,
+      task_repository: todo.task_repository ?? null});
+  } else {
+    requireThat(params.declaration === null && todo.completion_validation_sha256 == null,
+      "Todo without canonical validation authority cannot supply a declaration");
+    if (requirements === null) return unavailable("independent_delegation_validation_required");
+  }
+  return {todo_id: todo.todo_id, state: "ready",
+    source: requirements === null ? "todo_validation" : "goal_acceptance",
+    effects, canonical_done: todo.done === true && todo.status === "done"};
+}
 export function selectDelegationBinding(params: JsonObject): JsonObject {
   const config = requireJsonObject(params.config, "delegation configuration");
   requireThat(config.schema_version === "loopx_local_delegation_v0", "unsupported delegation configuration");

loopx/control_plane/collaboration/delegation_validation.py

@@ -0,0 +1,58 @@
+"""Host IO for delegation's single typed acceptance/validation plan.
+
+The canonical reader resolves Goal scope; TypeScript selects validation effects.
+Python resolves private declarations and executes only those authorized effects.
+No validator output or successful declaration read completes a canonical Todo.
+"""
+
+from ...agent_registry import load_goal_from_registry
+from ...materials import goal_state_path
+from ..effect_runtime import effect_runtime_result
+from ..goals.acceptance import (
+    inspect_goal_acceptance,
+    run_goal_acceptance_validation_effect,
+    validation_effect_files_current,
+)
+from ..todos.completion_validation import resolve_private_completion_validation_declaration
+
+
+def capture(service, binding: dict) -> dict:
+    basis = inspect_goal_acceptance(
+        registry_path=service.registry, runtime_root=str(service.root),
+        goal_id=service.goal_id, agent_id=binding["agent_id"], todo_id=binding["todo_id"],
+    )
+    todo = basis.get("todo")
+    if not isinstance(todo, dict):
+        raise ValueError("delegation canonical Todo unavailable")
+    goal = load_goal_from_registry(service.registry, service.goal_id)
+    state_file = goal_state_path(goal) if goal is not None else None
+    if state_file is None:
+        raise ValueError("delegation validation workspace unavailable")
+    declaration = resolve_private_completion_validation_declaration(
+        canonical_todo=todo, state_file=state_file, runtime_root=service.root,
+        registry_path=service.registry, goal_id=service.goal_id,
+        todo_id=binding["todo_id"], role=todo.get("role"), persist_if_resolved=False,
+    )
+    plan = effect_runtime_result("collaboration.delegation.validation_plan", {
+        "binding": {key: binding[key] for key in ("id", "agent_id", "todo_id")},
+        "basis": basis, "declaration": declaration,
+    })
+    files_current = plan["state"] == "ready" and validation_effect_files_current(
+        effects=plan["effects"], registry_path=service.registry, goal_id=service.goal_id,
+    )
+    return {"basis": basis, "plan": plan, "files_current": files_current}
+
+
+def validate(service, binding: dict) -> dict:
+    before = capture(service, binding)
+    if before["plan"]["state"] != "ready" or not before["files_current"]:
+        raise ValueError("delegation task acceptance rejected")
+    results = [run_goal_acceptance_validation_effect(
+        effect=effect, registry_path=service.registry, goal_id=service.goal_id,
+    ) for effect in before["plan"]["effects"]]
+    after = capture(service, binding)
+    if after != before:
+        raise ValueError("delegation task acceptance changed during validation")
+    if not all(result["passed"] for result in results):
+        raise ValueError("delegation task acceptance rejected")
+    return after