fix(todos): compose public update planning and preserve retained waits

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.

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 投影仍属于终态架构。
 

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

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 权限。

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

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],
     [

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,

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"),