From fe5ef4f831170f9b6f734c3fd806c9ba8a806109 Mon Sep 17 00:00:00 2001
From: huangruiteng <huangrt01@163.com>
Date: Thu, 10 Sep 2026 16:27:52 +0800
Subject: [PATCH] refactor(todos): compose public update planning over one
 snapshot

Signed-off-by: huangruiteng <huangrt01@163.com>
---
 ...shared-goal-authority-state-provider-v0.md |   8 +-
 ...-goal-authority-state-provider-v0.zh-CN.md |   5 +-
 .../typescript-control-plane-migration-v0.md  |  16 +++
 ...script-control-plane-migration-v0.zh-CN.md |  13 ++
 docs/project-agent-todo-contract.md           |  10 ++
 .../control_plane/effect_runtime_handlers.ts  |   2 +
 loopx/control_plane/todos/authoring_scope.py  |  14 ++-
 .../todos/external_wait_writeback.py          | 112 ------------------
 loopx/control_plane/todos/line_update.py      |  39 +++++-
 loopx/control_plane/todos/public_update.ts    |  70 +++++++++++
 loopx/control_plane/todos/resume_condition.py |   9 +-
 loopx/control_plane/todos/update_source.py    |  43 +++++++
 loopx/todos.py                                |  63 ++--------
 .../test_public_todo_update_plan.py           |  86 ++++++++++++++
 .../public_todo_update.test.ts                |  84 +++++++++++++
 15 files changed, 390 insertions(+), 184 deletions(-)
 delete mode 100644 loopx/control_plane/todos/external_wait_writeback.py
 create mode 100644 loopx/control_plane/todos/public_update.ts
 create mode 100644 loopx/control_plane/todos/update_source.py
 create mode 100644 tests/control_plane/test_public_todo_update_plan.py
 create mode 100644 tests/control_plane_ts/public_todo_update.test.ts

diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
index be3a160202..5029d9066c 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
@@ -2524,9 +2524,11 @@ commands fail closed; they do not fall back to the old writer.
 
 #### Refactoring roadmap overview
 
-The Monitor state owner now lives in TS and is composed by the legacy update
-field plan. This removes Python poll/generation and metadata rules, but the
-legacy writer still holds the lock and commits the result. The typed plan is
+The Monitor state owner now lives in TS and is composed with authoring scope,
+external-wait validation and field updates by one public update plan. Python
+transports the locked compact snapshot instead of sequencing those leaf RPCs;
+partial topology edits cannot invalidate retained waits. The legacy writer
+still owns admission, the lock and persistence. The typed plan is
 not an authority receipt; monitor/successor atomicity, native metadata update,
 provider defaults and D1–D3 remain separate, unfinished gates. Permanent
 Markdown projection remains part of the target architecture.
diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
index 5a44fad64f..e4e3dea5dc 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
@@ -2002,8 +2002,9 @@ backend、实时双向同步或按命令拆开的权威；晋升后不支持的
 
 #### 重构主线总览
 
-Monitor 状态 owner 现位于 TS，并由 legacy update field plan 组合调用；删除 Python
-poll/generation 与 metadata 规则，但持锁及结果提交仍由 legacy writer 负责。
+Monitor 状态 owner 现位于 TS，并与 authoring scope、external-wait 校验及字段更新
+组合为一次公开 update 规划。Python 输送锁内完整紧凑快照，不再逐个编排 leaf RPC；
+局部拓扑修改不能破坏保留的等待条件。准入、持锁及持久化仍由 legacy writer 负责。
 Typed plan 不是 authority receipt；Monitor/successor 原子性、原生 metadata update、
 provider 默认值及 D1–D3 仍是独立、未完成的门禁。永久 Markdown 投影仍属于终态架构。
 
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
index eeb2465cbd..8c8d87cee6 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
@@ -379,6 +379,22 @@ the shared plan, not another per-agent checklist database.
 
 **T1 — close the public Todo update transaction.**
 
+Bounded prerequisite: `todos/public_update.ts` now composes authoring scope,
+external-wait topology and Monitor/field planning over one locked source.
+The public Python writer no longer sequences their leaf RPCs or derives the
+Monitor wait baseline. `update_source.py` supplies complete compact active/archive
+facts, never a display-limited inventory. A partial topology edit validates its
+retained wait; copy-only edits preserve the original fence without re-arming it.
+Explicitly clearing the condition still permits changing its former topology.
+Locked completion proof is checked before this pure plan, so a stale proof wins
+over unrelated invalid field diagnostics; no write occurs in either case.
+This deletes orchestration, not persistence: lifecycle/lease admission, completion
+effects, writer lock, capture and provider CAS/replay remain with their existing
+owners. The internal terminal/import field codec still has actual callers and
+does not acquire the public update policy. Native metadata expansion and T2
+atomic follow-up remain held. Reconcile the separate lease-edit PR #4152 before
+changing the provider transaction; do not infer it is merged from this checkpoint.
+
 - Reuse the current provider text/note transaction, lifecycle admission,
   field-plan and completion rules. Enumerate actual public metadata edits and
   explicit-clear behavior before implementation; this is not permission to
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
index 98d5cd7080..68c4273124 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
@@ -294,6 +294,19 @@ commit。#4121（SQLite 候选）和 #4101（投影 receipt 保留）是独立
 
 **T1 — 闭合公开 Todo update 事务。**
 
+已闭合的前置项：`todos/public_update.ts` 在同一锁内快照上组合 authoring scope、
+external-wait 拓扑和 Monitor/field 规划。公开 Python writer 不再逐个调用这些
+leaf RPC，也不推导 Monitor 等待基线。`update_source.py` 只输送完整、紧凑的
+active/archive 事实，不使用受展示条数限制的 inventory。局部拓扑修改必须验证
+保留的等待条件；纯文案修改保留原 fence，不重新设置等待。显式清除条件后，仍可
+修改原来的拓扑。锁内 completion proof 先于纯规划检查，因此 proof 已过期时，
+优先返回该失败而非其他非法字段诊断；两种失败均不写入。
+这里删除的是编排而非持久化：lifecycle/lease 准入、completion effect、writer
+lock、capture、provider CAS/replay 仍由既有 owner 负责。内部 terminal/import
+field codec 仍有真实 caller，不引入公开 update 限制。Native metadata 扩展和
+T2 原子后续动作仍未闭合。修改 provider 事务前先核对独立 lease-edit PR #4152，
+不能从本检查点推断它已经合入。
+
 - 复用现有 provider text/note 事务、lifecycle 准入、field-plan 和 completion
   规则。先枚举公开 metadata 编辑与显式 clear，不把 `UPDATE_FIELDS` 扩成所有存储
   字段，也不让 generic patch 获得 terminal transition 权限。
diff --git a/docs/project-agent-todo-contract.md b/docs/project-agent-todo-contract.md
index 9fd3435d54..0afe909469 100644
--- a/docs/project-agent-todo-contract.md
+++ b/docs/project-agent-todo-contract.md
@@ -102,6 +102,16 @@ eligible time, `--cadence` is the retry interval, `--monitor-target-key` is the
 stable idempotency key, and optional `--expires-at` is the hard stop after
 which the monitor must not catch up.
 
+Public Todo updates validate the effective waiting state, not just newly supplied
+`resume_when`. Changing a Monitor-waiting Todo's status/task class or successor
+list must preserve the open advancement-task/independent-successor contract.
+To leave that contract, explicitly clear `resume_when` in the same update; this
+also clears its Monitor generation fence. Ordinary text/note corrections do not
+re-arm a wait, reset its baseline, or demand a new successor after its condition
+becomes satisfied. Explicitly re-submitting a satisfied Monitor condition still
+requires clearing it before re-arming. These checks are planning constraints,
+not permission to claim work, commit to a provider, or execute a successor.
+
 Monitor observations are reduced against the Todo under its existing writer lock.
 Callers report a result hash and material-change fact; they must not independently
 increment counters. A material observation with a different result hash increments
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index ac03f744cc..025a19384b 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -66,6 +66,7 @@ import {
 import { reduceTodoCompletionTransaction } from "./todos/completion_transaction.ts";
 import { transitionTodoNextAction } from "./todos/next_action.ts";
 import { planTodoFieldUpdate } from "./todos/field_update.ts";
+import { planPublicTodoUpdate } from "./todos/public_update.ts";
 import { planMonitorMetadata } from "./todos/monitor_metadata.ts";
 import { planTodoAuthoringScope } from "./todos/authoring_scope.ts";
 import {
@@ -371,6 +372,7 @@ export function createEffectRuntimeHandlers(
     ["todo.completion_state.require_metadata", requireTodoCompletionMetadataValue],
     ["todo.completion_state.continuation_for_write", selectTodoCompletionContinuation],
     ["todo.field_update.plan", planTodoFieldUpdate],
+    ["todo.public_update.plan", planPublicTodoUpdate],
     ["todo.monitor_metadata.plan", planMonitorMetadata],
     ["todo.authoring_scope.plan", planTodoAuthoringScope],
     [
diff --git a/loopx/control_plane/todos/authoring_scope.py b/loopx/control_plane/todos/authoring_scope.py
index 6333ae7a22..3faa08dc0b 100644
--- a/loopx/control_plane/todos/authoring_scope.py
+++ b/loopx/control_plane/todos/authoring_scope.py
@@ -8,16 +8,20 @@
 )
 
 
-def plan_todo_authoring_scope(
-    *, command: str, role: str, intent: dict[str, Any],
-    registered_agents: list[str], goal_id: str, todo: dict[str, Any] | None = None,
-) -> dict[str, Any]:
-    source = todo or {}
+def todo_authoring_facts(source: dict[str, Any]) -> dict[str, Any]:
     facts = {key: source.get(key) for key in ("status", "task_class", "resume_when", "excluded_agents")}
     facts.update({"blocks_agent": normalize_todo_blocks_agent(source.get("blocks_agent")),
         "bound_agent": normalize_todo_bound_agent(source.get("bound_agent")),
         "global_gate": normalize_todo_global_gate(source.get("global_gate")),
         "goal_bound": normalize_todo_goal_bound(source.get("goal_bound"))})
+    return facts
+
+
+def plan_todo_authoring_scope(
+    *, command: str, role: str, intent: dict[str, Any],
+    registered_agents: list[str], goal_id: str, todo: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    facts = todo_authoring_facts(todo or {})
     try:
         result = effect_runtime_result("todo.authoring_scope.plan", {
             "schema_version": "todo_authoring_scope_request_v0", "command": command,
diff --git a/loopx/control_plane/todos/external_wait_writeback.py b/loopx/control_plane/todos/external_wait_writeback.py
deleted file mode 100644
index 40a5f64aa5..0000000000
--- a/loopx/control_plane/todos/external_wait_writeback.py
+++ /dev/null
@@ -1,112 +0,0 @@
-from __future__ import annotations
-
-from collections.abc import Mapping
-from typing import Any
-
-from .active_state_editing import archive_section_bounds, section_bounds, todo_blocks
-from .contract import (
-    TODO_STATUS_OPEN,
-    TODO_TASK_CLASS_ADVANCEMENT,
-    normalize_todo_id,
-    normalize_todo_id_list,
-)
-from .external_wait_contract import build_monitor_advancement_authoring_contract
-from .resume_condition import plan_todo_external_wait_transition
-
-
-def _transition_items(lines: list[str]) -> list[dict[str, Any]]:
-    """Collect complete active/archive Todo metadata under the caller's lock."""
-
-    collected: list[dict[str, Any]] = []
-    archive_bounds = archive_section_bounds(lines)
-    if archive_bounds:
-        collected.extend(
-            {
-                **item,
-                "role": "agent",
-            }
-            for item in todo_blocks(
-                lines,
-                archive_bounds[0],
-                archive_bounds[1],
-                role="agent",
-                source_section="Completed Work Archive",
-            )
-        )
-    for role in ("user", "agent"):
-        bounds = section_bounds(lines, role)
-        if bounds:
-            collected.extend(
-                {
-                    **item,
-                    "role": role,
-                }
-                for item in todo_blocks(
-                    lines,
-                    bounds[0],
-                    bounds[1],
-                    role=role,
-                    source_section=bounds[2],
-                )
-            )
-    return collected
-
-
-def plan_todo_external_wait_update(
-    *,
-    lines: list[str],
-    todo_id: str,
-    resume_when: str | None,
-    successor_todo_ids: list[str] | None,
-    existing_successor_todo_ids: Any,
-    role: str,
-    status: str,
-    task_class: str,
-) -> tuple[dict[str, Any] | None, int | None]:
-    """Plan one typed open-Todo wait and return its persisted monitor fence."""
-
-    if not resume_when or not resume_when.startswith(
-        ("todo_done:", "monitor_changed:")
-    ):
-        return None, None
-    uses_open_external_wait = (
-        role == "agent"
-        and status == TODO_STATUS_OPEN
-        and task_class == TODO_TASK_CLASS_ADVANCEMENT
-    )
-    # todo_done is also valid for ordinary deferred authoring. monitor_changed,
-    # however, always uses the typed open-wait protocol so invalid role/status/
-    # task-class combinations receive the exact TS-owned diagnostic.
-    if resume_when.startswith("todo_done:") and not uses_open_external_wait:
-        return None, None
-
-    items = _transition_items(lines)
-    normalized_id = normalize_todo_id(todo_id)
-    for index, item in enumerate(items):
-        if normalize_todo_id(item.get("todo_id")) == normalized_id:
-            items[index] = {
-                **item,
-                "role": role,
-                "status": status,
-                "task_class": task_class,
-            }
-    transition = plan_todo_external_wait_transition(
-        todo_id=todo_id,
-        resume_when=resume_when,
-        successor_todo_ids=(
-            successor_todo_ids
-            if successor_todo_ids is not None
-            else normalize_todo_id_list(existing_successor_todo_ids)
-        ),
-        items=items,
-    )
-    if resume_when.startswith("monitor_changed:"):
-        transition["authoring_contract"] = (
-            build_monitor_advancement_authoring_contract(
-                monitor_todo_id=resume_when.partition(":")[2],
-                successor_todo_ids=list(transition.get("successor_todo_ids") or []),
-            )
-        )
-    updates = transition.get("metadata_updates")
-    baseline = updates.get("resume_monitor_generation") if isinstance(updates, Mapping) else None
-    return transition, int(baseline) if baseline is not None else None
diff --git a/loopx/control_plane/todos/line_update.py b/loopx/control_plane/todos/line_update.py
index 876a56ed1c..4327cdb967 100644
--- a/loopx/control_plane/todos/line_update.py
+++ b/loopx/control_plane/todos/line_update.py
@@ -4,6 +4,9 @@
 from typing import Any
 
 from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
+from .authoring_scope import todo_authoring_facts
+from .external_wait_contract import TodoExternalWaitAuthoringError, build_monitor_advancement_authoring_contract
+from .update_source import todo_update_snapshot
 
 from .active_state_editing import (
     TODO_SECTION_HEADINGS,
@@ -153,15 +156,18 @@ def link_superseding_todo_id(
 def _field_update_plan(
     block: Mapping[str, Any], intent: dict[str, Any], updated_at: str,
     monitor_context: dict[str, Any] | None = None,
+    public_context: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Adapt source facts only; the TS planner owns omission/clear/state rules."""
     try:
         result = effect_runtime_result(
-            "todo.field_update.plan",
+            "todo.public_update.plan" if public_context is not None else "todo.field_update.plan",
             {
-                "schema_version": "loopx_todo_field_update_request_v0",
+                "schema_version": "todo_public_update_request_v0" if public_context is not None else "loopx_todo_field_update_request_v0",
                 "todo": {
-                    key: block.get(key)
+                    **todo_authoring_facts(dict(block)),
+                    "role": block.get("role"),
+                    **{key: block.get(key)
                     for key in (
                         "todo_id",
                         "status",
@@ -171,16 +177,25 @@ def _field_update_plan(
                         "no_followup",
                         "completion_continuation",
                         "successor_todo_ids",
+                        "resume_monitor_generation",
                         "task_class",
                         *TODO_MONITOR_METADATA_FIELDS,
-                    )
+                    )},
                 },
                 "intent": intent,
                 "updated_at": updated_at,
                 "monitor_context": monitor_context,
+                "context": public_context,
             },
         )
     except EffectRuntimeRejected as exc:
+        if public_context is not None and exc.diagnostic_code.startswith("external_wait_"):
+            condition = str(intent.get("resume_when") or block.get("resume_when") or "").strip().lower()
+            kind, _, target = condition.partition(":")
+            raise TodoExternalWaitAuthoringError(str(exc), code=exc.diagnostic_code,
+                monitor_todo_id=target if kind == "monitor_changed" else None,
+                successor_todo_ids=intent.get("successor_todo_ids")
+                    if intent.get("successor_todo_ids") is not None else block.get("successor_todo_ids")) from None
         raise ValueError(str(exc)) from None
     if (
         not isinstance(result, dict)
@@ -191,6 +206,11 @@ def _field_update_plan(
         or not isinstance(result.get("metadata_updates"), dict)
     ):
         raise RuntimeError("TypeScript Todo field update result shape mismatch")
+    transition = result.get("external_wait_transition")
+    if isinstance(transition, dict) and transition.get("resume_kind") == "monitor_changed":
+        transition["authoring_contract"] = build_monitor_advancement_authoring_contract(
+            monitor_todo_id=transition["dependency_todo_id"],
+            successor_todo_ids=transition["successor_todo_ids"])
     return result
 
 
@@ -238,11 +258,12 @@ def apply_todo_update_to_lines(
     no_followup: bool | None = None,
     monitor_metadata: dict[str, Any] | None = None,
     monitor_context: dict[str, Any] | None = None,
+    public_context: dict[str, Any] | None = None,
     clear_claim: bool = False,
     claim_only: bool = False,
     updated_at: str,
 ) -> dict[str, Any]:
-    normalized_resume_when = require_supported_todo_resume_when(resume_when)
+    normalized_resume_when = resume_when if public_context is not None else require_supported_todo_resume_when(resume_when)
     if normalized_resume_when and clear_resume_when:
         raise ValueError(
             "todo update accepts either resume_when or clear_resume_when, not both"
@@ -261,8 +282,11 @@ def apply_todo_update_to_lines(
             f"todo_id {normalized_todo_id!r} was not found in active user or agent todos"
         )
     resolved_role, section, _start, _end, block = block_match
+    if public_context is not None:
+        public_context = {**public_context, "items": todo_update_snapshot(lines)
+                          if resume_when or block.get("resume_when") else []}
     plan = _field_update_plan(
-        block,
+        {**block, "role": resolved_role},
         {
             "status": status,
             "note": note,
@@ -306,6 +330,7 @@ def apply_todo_update_to_lines(
         },
         updated_at,
         monitor_context,
+        public_context,
     )
     normalized_status = plan["normalized_status"]
     target_status = plan["target_status"]
@@ -328,6 +353,8 @@ def apply_todo_update_to_lines(
     return {
         **({"monitor_poll_transition": plan["monitor_poll_transition"]}
            if "monitor_poll_transition" in plan else {}),
+        **({"external_wait_transition": plan["external_wait_transition"]}
+           if "external_wait_transition" in plan else {}),
         "role": resolved_role,
         "section": section,
         "todo": block.get("text"),
diff --git a/loopx/control_plane/todos/public_update.ts b/loopx/control_plane/todos/public_update.ts
new file mode 100644
index 0000000000..1c4bf4789e
--- /dev/null
+++ b/loopx/control_plane/todos/public_update.ts
@@ -0,0 +1,70 @@
+/** One public edit plan over one locked snapshot. This is not admission, a
+ * provider transaction, or a receipt: callers retain permission/lock/commit. */
+import type { JsonObject } from "../effect_program.ts";
+import { requireJsonObject } from "../runtime_decode.ts";
+import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
+import { planTodoAuthoringScope, TODO_AUTHORING_SCOPE_REQUEST_SCHEMA } from "./authoring_scope.ts";
+import { planTodoFieldUpdate, TODO_FIELD_UPDATE_REQUEST_SCHEMA } from "./field_update.ts";
+import { planTodoExternalWaitTransition, TODO_EXTERNAL_WAIT_REQUEST_SCHEMA_VERSION } from "./resume_condition.ts";
+
+export const TODO_PUBLIC_UPDATE_REQUEST_SCHEMA = "todo_public_update_request_v0";
+
+const SCOPE_INTENT_FIELDS = ["task_class", "status", "claimed_by", "bound_agent", "goal_bound",
+  "blocks_agent", "clear_blocks_agent", "global_gate", "clear_global_gate", "excluded_agents",
+  "task_repository", "task_domain", "resume_when", "clear_resume_when"] as const;
+
+function externalWait(todo: JsonObject, intent: JsonObject, scope: JsonObject,
+  context: JsonObject): JsonObject | null {
+  // Copy-only edits do not re-arm a satisfied wait or demand a fresh successor.
+  // Changes to its shape, however, must validate the retained condition too.
+  const shapeChanged = ["status", "task_class"].some(key => scope[key] !== todo[key]) ||
+    context.role !== todo.role || intent.successor_todo_ids != null;
+  const resume = scope.normalized_resume_when || (shapeChanged ? scope.effective_resume_when : null);
+  if (typeof resume !== "string") return null;
+  const kind = resume.split(":", 1)[0];
+  if (kind !== "todo_done" && kind !== "monitor_changed") return null;
+  if (kind === "todo_done" && !(context.role === "agent" && scope.status === "open" &&
+      scope.task_class === "advancement_task")) return null;
+  if (!Array.isArray(context.items)) throw new EffectRuntimeRequestError("public update requires its locked Todo snapshot");
+  const items = context.items.map(value => {
+    const item = requireJsonObject(value, "public update Todo item");
+    return item.todo_id === todo.todo_id ? {...item, role: context.role,
+      status: scope.status, task_class: scope.task_class} : item;
+  });
+  return planTodoExternalWaitTransition({schema_version: TODO_EXTERNAL_WAIT_REQUEST_SCHEMA_VERSION,
+    todo_id: todo.todo_id, resume_when: resume,
+    successor_todo_ids: intent.successor_todo_ids ?? todo.successor_todo_ids ?? [], items});
+}
+
+export function planPublicTodoUpdate(value: unknown): JsonObject {
+  const request = requireJsonObject(value, "public Todo update request");
+  if (request.schema_version !== TODO_PUBLIC_UPDATE_REQUEST_SCHEMA) {
+    throw new EffectRuntimeRequestError("public Todo update schema mismatch");
+  }
+  const todo = requireJsonObject(request.todo, "public Todo update source");
+  const intent = requireJsonObject(request.intent, "public Todo update intent");
+  const context = requireJsonObject(request.context, "public Todo update context");
+  const scope = planTodoAuthoringScope({schema_version: TODO_AUTHORING_SCOPE_REQUEST_SCHEMA,
+    command: "update", role: context.role, todo,
+    intent: {...Object.fromEntries(SCOPE_INTENT_FIELDS.map(key => [key, intent[key] ?? null])),
+      actor_agent_id: context.actor_agent_id ?? null},
+    goal_id: context.goal_id, registered_agents: context.registered_agents});
+  const transition = externalWait(todo, intent, scope, context);
+  const metadata = transition ? requireJsonObject(transition.metadata_updates, "external wait updates") : null;
+  const role = context.role;
+  const effectiveIntent = {...intent,
+    bound_agent: role === "user" ? scope.bound_agent : null,
+    goal_bound: role === "user" && scope.goal_bound ? true : null,
+    clear_user_binding: scope.clear_user_binding,
+    resume_when: scope.normalized_resume_when,
+    resume_monitor_generation: metadata?.resume_monitor_generation ?? null};
+  // A retained condition revalidated by a topology edit keeps its original
+  // generation; it must never silently capture a newer Monitor observation.
+  if (transition) effectiveIntent.resume_when = transition.resume_when;
+  const plan = planTodoFieldUpdate({schema_version: TODO_FIELD_UPDATE_REQUEST_SCHEMA,
+    todo, intent: effectiveIntent, updated_at: request.updated_at,
+    monitor_context: {metadata: intent.monitor_metadata ?? null,
+      observation: context.monitor_observation ?? null, role, task_class: scope.task_class,
+      resume_when: scope.effective_resume_when, enforce_boundedness: context.enforce_monitor_boundedness}});
+  return {...plan, ...(transition ? {external_wait_transition: transition} : {})};
+}
diff --git a/loopx/control_plane/todos/resume_condition.py b/loopx/control_plane/todos/resume_condition.py
index 3e2dd1e1a4..f76015f0c0 100644
--- a/loopx/control_plane/todos/resume_condition.py
+++ b/loopx/control_plane/todos/resume_condition.py
@@ -136,6 +136,11 @@ def _compact_item(value: Mapping[str, Any]) -> dict[str, Any]:
     }
 
 
+def compact_todo_resume_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Lossless-for-resume facts, independent of display limits and prose size."""
+    return [_compact_item(item) for item in items if item.get("todo_id")]
+
+
 def _pr_ref_number(value: Any) -> int | None:
     if not isinstance(value, str):
         return None
@@ -358,9 +363,7 @@ def plan_todo_external_wait_transition(
                 "todo_id": todo_id,
                 "resume_when": resume_when,
                 "successor_todo_ids": successor_todo_ids,
-                "items": [
-                    _compact_item(item) for item in items if item.get("todo_id")
-                ],
+                "items": compact_todo_resume_items(items),
             },
         )
     except EffectRuntimeRejected as exc:
diff --git a/loopx/control_plane/todos/update_source.py b/loopx/control_plane/todos/update_source.py
new file mode 100644
index 0000000000..2bafe1f034
--- /dev/null
+++ b/loopx/control_plane/todos/update_source.py
@@ -0,0 +1,43 @@
+"""Complete locked Todo facts for public update planning; no state decisions."""
+from typing import Any
+
+from .active_state_editing import archive_section_bounds, section_bounds, todo_blocks
+from .resume_condition import compact_todo_resume_items
+
+
+def todo_update_snapshot(lines: list[str]) -> list[dict[str, Any]]:
+    """Collect complete active/archive Todo metadata under the caller's lock."""
+
+    collected: list[dict[str, Any]] = []
+    archive_bounds = archive_section_bounds(lines)
+    if archive_bounds:
+        collected.extend(
+            {
+                **item,
+                "role": "agent",
+            }
+            for item in todo_blocks(
+                lines,
+                archive_bounds[0],
+                archive_bounds[1],
+                role="agent",
+                source_section="Completed Work Archive",
+            )
+        )
+    for role in ("user", "agent"):
+        bounds = section_bounds(lines, role)
+        if bounds:
+            collected.extend(
+                {
+                    **item,
+                    "role": role,
+                }
+                for item in todo_blocks(
+                    lines,
+                    bounds[0],
+                    bounds[1],
+                    role=role,
+                    source_section=bounds[2],
+                )
+            )
+    return compact_todo_resume_items(collected)
diff --git a/loopx/todos.py b/loopx/todos.py
index e353174dfe..c296817898 100644
--- a/loopx/todos.py
+++ b/loopx/todos.py
@@ -92,7 +92,6 @@
     todo_summaries_from_fields,
 )
 from .control_plane.todos import monitor_metadata as todo_monitor_metadata
-from .control_plane.todos.external_wait_writeback import plan_todo_external_wait_update
 from .control_plane.todos.mutation_authority import authorize_todo_lifecycle_mutation, todo_update_authority_action
 from .control_plane.todos.succession_warning import build_open_parent_successor_advisory
 from .control_plane.todos.successor_derivation import (
@@ -1160,8 +1159,6 @@ def update_goal_todo(
         validation_failure = completion_validation_gate.get("failure")
         if validation_failure is not None:
             return validation_failure
-    external_wait_transition: dict[str, Any] | None = None
-    resume_monitor_generation: int | None = None
     with legacy_todo_write_transaction(
         registry_path, goal_id, resolved_state_file, agent_id or claimed_by, "todo_update", dry_run,
         runtime_root=shadow_runtime_root,
@@ -1267,21 +1264,6 @@ def update_goal_todo(
                 registry_path=registry_path, goal_id=goal_id, excluded_agents=excluded_agents,
             ) if excluded_agents is not None else None
         )
-        authoring_scope = plan_todo_authoring_scope(
-            command="update", role=target_role, goal_id=goal_id, todo=existing_block,
-            registered_agents=registered_agent_ids_from_registry(registry_path, goal_id),
-            intent={
-                "task_class": task_class, "status": status, "actor_agent_id": effective_agent_id,
-                "claimed_by": effective_claimed_by, "bound_agent": effective_bound_agent,
-                "goal_bound": goal_bound, "blocks_agent": effective_blocks_agent,
-                "clear_blocks_agent": clear_blocks_agent, "global_gate": global_gate,
-                "clear_global_gate": clear_global_gate, "excluded_agents": effective_excluded_agents,
-                "task_repository": task_repository, "task_domain": task_domain,
-                "resume_when": resume_when, "clear_resume_when": clear_resume_when,
-            },
-        )
-        target_task_class = authoring_scope["task_class"]
-        target_status = authoring_scope["status"]
         completion_metadata_updates_override = None
         if completion_validation_gate is not None:
             locked_completion = locked_todo_completion_transaction(
@@ -1304,30 +1286,10 @@ def update_goal_todo(
                     ),
                 )
             )
-        target_bound_agent = authoring_scope["bound_agent"]
-        target_goal_bound = authoring_scope["goal_bound"]
         normalized_unblocks_todo_id = normalize_todo_id(unblocks_todo_id) if unblocks_todo_id else None
         if unblocks_todo_id and not normalized_unblocks_todo_id:
             raise ValueError("unblocks_todo_id must use the public token shape todo_<letters-digits-underscore-hyphen>")
         normalized_successor_todo_ids = requested_successor_todo_ids
-        normalized_resume_when = authoring_scope["normalized_resume_when"]
-        effective_resume_when = authoring_scope["effective_resume_when"]
-        external_wait_transition, resume_monitor_generation = (
-            plan_todo_external_wait_update(
-                lines=lines,
-                todo_id=todo_id,
-                resume_when=normalized_resume_when,
-                successor_todo_ids=(
-                    normalized_successor_todo_ids
-                    if successor_todo_ids is not None
-                    else None
-                ),
-                existing_successor_todo_ids=existing_block.get("successor_todo_ids"),
-                role=target_role,
-                status=target_status,
-                task_class=target_task_class,
-            )
-        )
         update_result = apply_todo_update_to_lines(
             lines,
             todo_id=todo_id,
@@ -1349,13 +1311,8 @@ def update_goal_todo(
             decision_scope=decision_scope,
             required_decision_scopes=required_decision_scopes,
             claimed_by=effective_claimed_by,
-            bound_agent=target_bound_agent if target_role == "user" else None,
-            goal_bound=(
-                True
-                if target_role == "user" and target_goal_bound
-                else None
-            ),
-            clear_user_binding=authoring_scope["clear_user_binding"],
+            bound_agent=effective_bound_agent,
+            goal_bound=goal_bound,
             blocks_agent=effective_blocks_agent,
             clear_blocks_agent=clear_blocks_agent,
             excluded_agents=effective_excluded_agents,
@@ -1363,17 +1320,19 @@ def update_goal_todo(
             clear_global_gate=clear_global_gate,
             unblocks_todo_id=normalized_unblocks_todo_id,
             successor_todo_ids=normalized_successor_todo_ids if successor_todo_ids is not None else None,
-            resume_when=normalized_resume_when,
-            resume_monitor_generation=resume_monitor_generation,
+            resume_when=resume_when,
             clear_resume_when=clear_resume_when,
             no_followup=no_followup,
             completion_metadata_updates_override=(
                 completion_metadata_updates_override
             ),
-            monitor_context={
-                **monitor_intent, "role": target_role, "task_class": target_task_class,
-                "resume_when": effective_resume_when,
-                "enforce_boundedness": enforce_monitor_boundedness,
+            monitor_metadata=monitor_intent["metadata"],
+            public_context={
+                "goal_id": goal_id, "role": target_role,
+                "actor_agent_id": effective_agent_id,
+                "registered_agents": registered_agent_ids_from_registry(registry_path, goal_id),
+                "monitor_observation": monitor_intent["observation"],
+                "enforce_monitor_boundedness": enforce_monitor_boundedness,
             },
             clear_claim=clear_claim,
             claim_only=claim_only,
@@ -1408,8 +1367,6 @@ def update_goal_todo(
         )
         if parent_successor_advisory:
             payload["parent_successor_advisory"] = parent_successor_advisory
-    if external_wait_transition is not None:
-        payload["external_wait_transition"] = external_wait_transition
     payload = _attach_todo_write_correctness_dry_run_packet(
         payload,
         goal_id=goal_id,
diff --git a/tests/control_plane/test_public_todo_update_plan.py b/tests/control_plane/test_public_todo_update_plan.py
new file mode 100644
index 0000000000..1e443c0df3
--- /dev/null
+++ b/tests/control_plane/test_public_todo_update_plan.py
@@ -0,0 +1,86 @@
+"""Public update invariants: a partial edit must not invalidate an armed wait."""
+import pytest
+
+from loopx.todos import add_goal_todo, update_goal_todo
+from loopx.control_plane.testing.canary_harness import run_json_cli_result
+from tests.control_plane.test_monitor_followthrough_contract import (
+    AGENT_ID, GOAL_ID, _add_monitor, _write_fixture,
+)
+
+
+def waiting_goal(tmp_path):
+    registry, runtime, state = _write_fixture(tmp_path)
+    monitor = _add_monitor(registry, text="Observe fixture", target_key="fixture")
+    update_goal_todo(registry_path=registry, goal_id=GOAL_ID,
+        todo_id=monitor["todo_id"], agent_id=AGENT_ID,
+        monitor_metadata={"material_change_generation": "3"})
+    def add(text):
+        return add_goal_todo(registry_path=registry, goal_id=GOAL_ID,
+            role="agent", task_class="advancement_task", text=text, claimed_by=AGENT_ID)
+    waiting, successor = add("Await observation"), add("Independent work")
+    update_goal_todo(registry_path=registry, goal_id=GOAL_ID,
+        todo_id=waiting["todo_id"], agent_id=AGENT_ID,
+        resume_when=f"monitor_changed:{monitor['todo_id']}",
+        successor_todo_ids=[successor["todo_id"]])
+    return registry, runtime, state, waiting, successor
+
+
+@pytest.mark.parametrize("edit", [{"successor_todo_ids": []}, {"status": "blocked"},
+                                    {"task_class": "continuous_monitor"}])
+def test_partial_edit_cannot_invalidate_retained_monitor_wait(tmp_path, edit):
+    registry, _, state, waiting, _ = waiting_goal(tmp_path)
+    before = state.read_bytes()
+    with pytest.raises(ValueError, match="external-wait"):
+        update_goal_todo(registry_path=registry, goal_id=GOAL_ID,
+            todo_id=waiting["todo_id"], agent_id=AGENT_ID, **edit)
+    assert state.read_bytes() == before
+
+
+def test_clear_condition_and_successors_together_is_valid(tmp_path):
+    registry, _, state, waiting, _ = waiting_goal(tmp_path)
+    result = update_goal_todo(registry_path=registry, goal_id=GOAL_ID,
+        todo_id=waiting["todo_id"], agent_id=AGENT_ID,
+        clear_resume_when=True, successor_todo_ids=[])
+    assert result["resume_when"] is None
+    assert result["resume_monitor_generation"] is None
+    assert result["successor_todo_ids"] == []
+
+
+def test_copy_edit_preserves_existing_wait_fence(tmp_path):
+    registry, _, _, waiting, _ = waiting_goal(tmp_path)
+    result = update_goal_todo(registry_path=registry, goal_id=GOAL_ID,
+        todo_id=waiting["todo_id"], agent_id=AGENT_ID, note="Clarify context")
+    assert result["resume_when"].startswith("monitor_changed:")
+    assert result["resume_monitor_generation"] == 3
+    assert "external_wait_transition" not in result
+
+
+def test_real_cli_rejects_partial_wait_edit_and_supports_explicit_repair(tmp_path):
+    registry, runtime, state, waiting, _ = waiting_goal(tmp_path)
+    common = ("todo", "update", "--goal-id", GOAL_ID, "--todo-id", waiting["todo_id"],
+              "--agent-id", AGENT_ID, "--status", "blocked")
+    before = state.read_bytes()
+    code, rejected = run_json_cli_result(*common, registry_path=registry, runtime_root=runtime)
+    assert code != 0, rejected
+    assert state.read_bytes() == before
+    code, accepted = run_json_cli_result(*common, "--clear-resume-when", "--dry-run",
+        registry_path=registry, runtime_root=runtime)
+    assert code == 0, accepted
+    assert state.read_bytes() == before
+
+
+def test_public_update_uses_one_scope_wait_field_crossing(tmp_path, monkeypatch):
+    from loopx.control_plane.todos import line_update, authoring_scope, resume_condition
+    registry, _, _, waiting, successor = waiting_goal(tmp_path)
+    actual = line_update.effect_runtime_result
+    calls = []
+    def traced(method, params):
+        calls.append(method)
+        return actual(method, params)
+    for module in (line_update, authoring_scope, resume_condition):
+        monkeypatch.setattr(module, "effect_runtime_result", traced)
+    result = update_goal_todo(registry_path=registry, goal_id=GOAL_ID,
+        todo_id=waiting["todo_id"], agent_id=AGENT_ID,
+        successor_todo_ids=[successor["todo_id"]])
+    assert result["resume_monitor_generation"] == 3
+    assert calls == ["todo.public_update.plan"]
diff --git a/tests/control_plane_ts/public_todo_update.test.ts b/tests/control_plane_ts/public_todo_update.test.ts
new file mode 100644
index 0000000000..b82033125a
--- /dev/null
+++ b/tests/control_plane_ts/public_todo_update.test.ts
@@ -0,0 +1,84 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import type { JsonObject } from "../../loopx/control_plane/effect_program.ts";
+import { planPublicTodoUpdate, TODO_PUBLIC_UPDATE_REQUEST_SCHEMA } from "../../loopx/control_plane/todos/public_update.ts";
+import { productionScaleCoordinationFixture } from "./production_scale_coordination_fixture.ts";
+
+const todo = {todo_id: "todo_waiting", role: "agent", status: "open", task_class: "advancement_task",
+  claimed_by: "agent-a", resume_when: "monitor_changed:todo_monitor", resume_monitor_generation: 3,
+  successor_todo_ids: ["todo_successor"]};
+const monitor = {todo_id: "todo_monitor", role: "agent", status: "open", task_class: "continuous_monitor",
+  material_change_generation: 3};
+const successor = {todo_id: "todo_successor", role: "agent", status: "open", task_class: "advancement_task"};
+function plan(intent: JsonObject = {}, source: JsonObject = todo, extra: JsonObject = {}) {
+  return planPublicTodoUpdate({schema_version: TODO_PUBLIC_UPDATE_REQUEST_SCHEMA, todo: source,
+    intent, updated_at: "2030-01-01T00:00:00Z", context: {goal_id: "goal-a", role: source.role,
+      actor_agent_id: "agent-a", registered_agents: ["agent-a", "agent-b"],
+      enforce_monitor_boundedness: true, items: [source, monitor, successor], ...extra}});
+}
+
+test("partial topology edits validate the effective retained wait", () => {
+  for (const edit of [{successor_todo_ids: []}, {status: "blocked"}, {task_class: "continuous_monitor"}]) {
+    assert.throws(() => plan(edit), /external-wait/);
+  }
+  const clear = plan({clear_resume_when: true, successor_todo_ids: []});
+  const updates = clear.metadata_updates as JsonObject;
+  assert.equal(updates.resume_when, null);
+  assert.equal(updates.resume_monitor_generation, null);
+  assert.deepEqual(updates.successor_todo_ids, []);
+});
+
+test("same-condition retry keeps the baseline; ordinary copy does not rearm", () => {
+  const copy = plan({note: "Clarified"}, todo, {items: [todo, {...monitor, material_change_generation: 4}, successor]});
+  assert.equal(copy.external_wait_transition, undefined);
+  assert.equal((copy.metadata_updates as JsonObject).resume_monitor_generation, undefined);
+  const retry = plan({resume_when: todo.resume_when});
+  assert.equal((retry.metadata_updates as JsonObject).resume_monitor_generation, 3);
+  assert.equal((retry.external_wait_transition as JsonObject).state, "already_waiting");
+  assert.throws(() => plan({resume_when: todo.resume_when}, todo,
+    {items: [todo, {...monitor, material_change_generation: 4}, successor]}), /clear the satisfied/);
+  assert.equal(todo.resume_monitor_generation, 3);
+});
+
+test("deferred Todo and ordinary capacity/PR conditions retain their distinct contracts", () => {
+  const deferred = plan({status: "deferred", resume_when: "todo_done:todo_successor", successor_todo_ids: []});
+  assert.equal(deferred.external_wait_transition, undefined);
+  assert.equal((deferred.metadata_updates as JsonObject).resume_monitor_generation, null);
+  for (const condition of ["capacity_available:network", "pr_merged:owner/repo#12"]) {
+    assert.equal(plan({resume_when: condition}).external_wait_transition, undefined);
+  }
+  assert.throws(() => plan({resume_when: "monitor_changed:todo_waiting"}), /itself/);
+  assert.throws(() => plan({resume_when: "monitor_changed:todo_absent"}), /absent/);
+});
+
+test("public scope and fields agree without inventing a global gate", () => {
+  const gate = {todo_id: "todo_gate", role: "user", task_class: "user_gate", status: "open",
+    bound_agent: "agent-a", blocks_agent: "agent-a", goal_bound: false, global_gate: false};
+  const updates = plan({blocks_agent: "agent-b"}, gate).metadata_updates as JsonObject;
+  assert.equal(updates.bound_agent, "agent-b");
+  assert.equal(updates.blocks_agent, "agent-b");
+  assert.equal(updates.global_gate, undefined);
+  assert.throws(() => plan({goal_bound: true}, gate), /same agent/);
+  assert.throws(() => plan({status: "done"}), /complete_goal_todo/);
+});
+
+test("monitor observations use the same effective task scope and cannot be raw-state overrides", () => {
+  const observation = {generated_at: "2030-01-01T00:00:00Z", material_change: true,
+    result_hash: "changed", monitor_effect_id: "effect-a"};
+  const result = plan({}, {...monitor, watch_only: true}, {monitor_observation: observation});
+  assert.equal((result.metadata_updates as JsonObject).material_change_generation, "4");
+  assert.throws(() => plan({task_class: "advancement_task"}, {...monitor, watch_only: true},
+    {monitor_observation: observation}), /monitor schedule metadata/);
+  assert.throws(() => plan({monitor_metadata: {result_hash: "forged"}}, {...monitor, watch_only: true},
+    {monitor_observation: observation}), /cannot be combined/);
+});
+
+test("complete production-scale snapshot stays immutable and is not display-capped", () => {
+  const fixture = productionScaleCoordinationFixture("goal-a");
+  const snapshot = structuredClone(fixture);
+  const records = fixture.projection.todos as JsonObject[];
+  const result = plan({resume_when: todo.resume_when}, todo,
+    {items: [...records, todo, monitor, successor]});
+  assert.equal((result.metadata_updates as JsonObject).resume_monitor_generation, 3);
+  assert.deepEqual(fixture, snapshot);
+});