refactor(control-plane): remove unused legacy Todo, write-preview and trace helpers

docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-10-27-primary-action-instruction-selection.md

@@ -51,6 +51,9 @@ scalar/label codecs and facts carrier only after the last Python text/label
 consumer exits and a complete typed compatibility codec proves Python coercion,
 Unicode, truncation and trace parity. This slice does not retire those owners.
 
+Correction (2026-11-02): `protocol_monitor_action` had no remaining caller
+and is removed. The resolution trace remains live.
+
 ## Vocabulary budget
 
 The registry's `meaning` says "module_budget counts modules under loopx/ whose text carries the standalone field token" and "the token count stays until Q11 decides whether the syntactic metric replaces it"; this deliberately conservative lexical proxy protects against unexplained growth in potential field readers.

docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-10-27-primary-action-instruction-selection.zh-CN.md

@@ -43,6 +43,9 @@ facade。仅当最后一个 Python text/label 消费者退出，且完整 typed
 Python coercion、Unicode、截断及 trace 等价时，才删除 scalar/label codec 和 facts
 carrier。本切片不宣称这些 owner 已退役。
 
+更正（2026-11-02）：`protocol_monitor_action` 已无剩余调用方，现已删除。
+resolution trace 仍在使用。
+
 ## 词汇预算
 
 registry 的 `meaning` 写明 "module_budget counts modules under loopx/ whose text carries the standalone field token" 和 "the token count stays until Q11 decides whether the syntactic metric replaces it"；这一刻意保守的词法代理用于防止潜在字段 reader 数量出现无解释增长。

docs/architecture/rfcs/typescript-control-plane-migration-v0.md

@@ -1118,8 +1118,9 @@ closure, not completion of all Todo fields or Goal promotion.
 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
+Monitor wait baseline. Correction (2026-11-02): `update_source.py` had no
+remaining caller and is removed; planning still requires 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

docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md

@@ -844,8 +844,9 @@ promotion 的 Goal 继续使用旧 writer。这是有边界的 T1 闭合，不
 
 已闭合的前置项：`todos/public_update.ts` 在同一锁内快照上组合 authoring scope、
 external-wait 拓扑和 Monitor/field 规划。公开 Python writer 不再逐个调用这些
-leaf RPC，也不推导 Monitor 等待基线。`update_source.py` 只输送完整、紧凑的
-active/archive 事实，不使用受展示条数限制的 inventory。局部拓扑修改必须验证
+leaf RPC，也不推导 Monitor 等待基线。更正（2026-11-02）：`update_source.py`
+已无剩余调用方，现已删除；规划仍要求完整、紧凑的 active/archive 事实，
+不使用受展示条数限制的 inventory。局部拓扑修改必须验证
 保留的等待条件；纯文案修改保留原 fence，不重新设置等待。显式清除条件后，仍可
 修改原来的拓扑。锁内 completion proof 先于纯规划检查，因此 proof 已过期时，
 优先返回该失败而非其他非法字段诊断；两种失败均不写入。

loopx/control_plane/runtime/local_state_write_correctness.py

@@ -100,79 +100,6 @@ def build_local_state_write_correctness_dry_run_packet(
     }
 
 
-def shadow_validate_local_state_write_correctness_packet(
-    packet: dict[str, Any],
-    *,
-    current_state_text: str,
-    observed_lease_ref: dict[str, Any] | None = None,
-) -> dict[str, Any]:
-    """Preview revision and lease conflicts without changing write behavior."""
-
-    validated = json.loads(json.dumps(packet, ensure_ascii=False))
-    intent = validated.get("write_intent") if isinstance(validated, dict) else {}
-    if not isinstance(intent, dict):
-        return validated
-    apply_result = validated.setdefault("apply_result", {})
-    if not isinstance(apply_result, dict):
-        apply_result = {}
-        validated["apply_result"] = apply_result
-    apply_result.setdefault("status", "preview_only")
-    apply_result.setdefault("applied_revision", None)
-    apply_result.setdefault("duplicate_of", None)
-    apply_result.setdefault("conflict", None)
-
-    expected_revision = intent.get("expected_revision")
-    current_revision = active_state_revision(current_state_text)
-    if (
-        isinstance(expected_revision, dict)
-        and expected_revision.get("value")
-        and expected_revision.get("value") != current_revision["value"]
-    ):
-        apply_result.update(
-            {
-                "status": "revision_conflict",
-                "applied_revision": None,
-                "duplicate_of": None,
-                "conflict": {
-                    "kind": "revision_conflict",
-                    "expected_revision": expected_revision,
-                    "current_revision": current_revision,
-                },
-            }
-        )
-        return validated
-
-    expected_lease_ref = intent.get("lease_ref")
-    if _lease_refs_conflict(expected_lease_ref, observed_lease_ref):
-        apply_result.update(
-            {
-                "status": "lease_conflict",
-                "applied_revision": None,
-                "duplicate_of": None,
-                "conflict": {
-                    "kind": "lease_conflict",
-                    "expected_lease_ref": expected_lease_ref,
-                    "observed_lease_ref": observed_lease_ref,
-                },
-            }
-        )
-    return validated
-
-
-def _lease_refs_conflict(expected: Any, observed: Any) -> bool:
-    if not isinstance(expected, dict) or not isinstance(observed, dict):
-        return False
-    if str(expected.get("kind") or "") != str(observed.get("kind") or ""):
-        return False
-    if str(expected.get("goal_id") or "") != str(observed.get("goal_id") or ""):
-        return False
-    if str(expected.get("todo_id") or "") != str(observed.get("todo_id") or ""):
-        return False
-    expected_holder = str(expected.get("claimed_by") or "")
-    observed_holder = str(observed.get("claimed_by") or "")
-    return bool(expected_holder and observed_holder and expected_holder != observed_holder)
-
-
 def _todo_lease_ref(
     *,
     goal_id: str,

loopx/control_plane/todos/active_state_editing.py

@@ -12,8 +12,6 @@
     todo_done_for_status,
     todo_marker_for_status,
 )
-from .text import normalize_new_todo
-from .todo_summary import normalize_todo_text
 from .machine_region import find_todo_source_regions, visible_markdown_lines
 from .todo_block_codec import decode_todo_blocks
 
@@ -86,35 +84,6 @@ def set_todo_marker(lines: list[str], block: dict[str, Any], status: str) -> boo
     return True
 
 
-def set_todo_text(
-    lines: list[str],
-    block: dict[str, Any],
-    text: str,
-    *,
-    status: str,
-) -> bool:
-    normalized = normalize_new_todo(text)
-    if normalize_todo_text(str(block.get("text") or "")) == normalized:
-        return False
-
-    start = int(block["start"])
-    end = int(block["end"])
-    marker = todo_marker_for_status(status)
-    replacement = f"- [{marker}] {normalized}"
-    continuation_indexes = [
-        index
-        for index in range(start + 1, end)
-        if lines[index].startswith((" ", "\t"))
-        and parse_todo_metadata_line(lines[index]) is None
-    ]
-    lines[start] = replacement
-    for index in reversed(continuation_indexes):
-        del lines[index]
-    block["text"] = normalized
-    block["end"] = end - len(continuation_indexes)
-    return True
-
-
 def heading_index(lines: list[str], heading: str) -> int | None:
     needle = f"## {heading}"
     for index in sorted(visible_markdown_lines(lines)):

loopx/control_plane/todos/completion_validation.py

@@ -34,7 +34,6 @@
     persist_completion_validation_declaration,
     read_completion_validation_declaration,
 )
-from .contract import TODO_STATUS_DONE, normalize_todo_status
 from ..content_digest import BARE_SHA256_PATTERN
 
 # Kept safely under the 30s outer CLI/MCP subprocess budget so a timed-out
@@ -639,62 +638,6 @@ def completion_policy_source_from_state(
     )
 
 
-def prepare_user_todo_update_completion(
-    *,
-    status: str | None,
-    state_file: Path,
-    todo_id: str,
-    role: str | None,
-    registry_path: Path,
-    goal_id: str,
-    dry_run: bool,
-    no_followup: bool,
-    requested_has_successor: bool,
-) -> dict[str, Any] | None:
-    """Prepare the coarse transaction for a direct user-Todo completion."""
-
-    if normalize_todo_status(status) != TODO_STATUS_DONE:
-        return None
-    match = find_todo_block(
-        state_file.read_text(encoding="utf-8").splitlines(),
-        todo_id=todo_id,
-        role=role,
-    )
-    if match is None or (role or match[0]) == "agent":
-        return None
-    return run_completion_validation_gate_with_source(
-        state_file=state_file,
-        todo_id=todo_id,
-        role=role,
-        registry_path=registry_path,
-        goal_id=goal_id,
-        dry_run=dry_run,
-        no_followup=no_followup,
-        requested_has_successor=requested_has_successor,
-    )
-
-
-def locked_todo_completion_source(
-    *,
-    lines: list[str],
-    state_file: Path,
-    project: Path | None,
-    registry_path: Path,
-    goal_id: str,
-    todo_id: str,
-    role: str | None,
-) -> tuple[Any, dict[str, Any] | None]:
-    """Resolve the materialized Todo under the write lock."""
-
-    match = find_todo_block(lines, todo_id=todo_id, role=role)
-    if match:
-        item_role, _section, _start, _end, block = match
-        todo = dict(block)
-        todo["role"] = item_role
-        return match, todo
-    return None, None
-
-
 def execute_completion_validation_effects(
     result: Mapping[str, Any], *, registry_path: Path, goal_id: str,
     delivery_workspace: Mapping[str, Any] | None = None,

loopx/control_plane/todos/mutation_response.py

@@ -5,20 +5,6 @@
 from pathlib import Path
 from typing import Any
 
-from .contract import (
-    normalize_explore_result_node_refs, normalize_required_capabilities,
-    normalize_target_capabilities, normalize_todo_blocks_agent,
-    normalize_todo_bound_agent, normalize_todo_claimed_by,
-    normalize_todo_continuation_policy, normalize_todo_decision_scope,
-    normalize_todo_decision_scope_outcomes, normalize_todo_excluded_agents,
-    normalize_todo_global_gate, normalize_todo_goal_bound, normalize_todo_id,
-    normalize_todo_id_list, normalize_todo_no_followup,
-    normalize_todo_required_decision_scopes, normalize_todo_resume_when,
-    normalize_todo_task_domain, normalize_todo_task_repository,
-)
-from .completion_state import (
-    normalize_todo_completion_continuation, normalize_todo_completion_recovery,
-)
 from .todo_semantics import todo_priority_label
 
 def serialize_added_todo_payload(
@@ -78,104 +64,3 @@ def serialize_added_todo_payload(
         "updated_at": updated_at if changed else None,
         **handoff_gate,
     }
-
-def serialize_todo_update_result(
-    *, role: str, section: str, todo: str | None, todo_id: str,
-    status: str, priority: str | None, status_changed: bool,
-    text_changed: bool, metadata_updated: bool, metadata: Mapping[str, Any],
-    monitor_poll_transition: Mapping[str, Any] | None = None,
-    external_wait_transition: Mapping[str, Any] | None = None,
-) -> dict[str, Any]:
-    """Project the public update result from already decided Todo facts."""
-    effective_metadata = metadata
-    return {
-        **({"monitor_poll_transition": dict(monitor_poll_transition)}
-           if monitor_poll_transition is not None else {}),
-        **({"external_wait_transition": dict(external_wait_transition)}
-           if external_wait_transition is not None else {}),
-        "role": role,
-        "section": section,
-        "todo": todo,
-        "todo_id": todo_id,
-        "status": status,
-        "status_changed": status_changed,
-        "text_changed": text_changed,
-        "priority": priority,
-        "metadata_updated": metadata_updated,
-        "changed": status_changed or text_changed or metadata_updated,
-        "claimed_by": normalize_todo_claimed_by(effective_metadata.get("claimed_by")),
-        "bound_agent": normalize_todo_bound_agent(
-            effective_metadata.get("bound_agent")
-        ),
-        "goal_bound": normalize_todo_goal_bound(effective_metadata.get("goal_bound")),
-        "task_class": effective_metadata.get("task_class"),
-        "action_kind": effective_metadata.get("action_kind"),
-        "task_domain": normalize_todo_task_domain(
-            effective_metadata.get("task_domain")
-        ),
-        "capability_binding_ref": effective_metadata.get("capability_binding_ref"),
-        "task_repository": normalize_todo_task_repository(
-            effective_metadata.get("task_repository")
-        ),
-        "continuation_policy": normalize_todo_continuation_policy(
-            effective_metadata.get("continuation_policy")
-        ),
-        "required_capabilities": normalize_required_capabilities(
-            effective_metadata.get("required_capabilities")
-        ),
-        "target_capabilities": normalize_target_capabilities(
-            effective_metadata.get("target_capabilities")
-        ),
-        "explore_result_node_refs": normalize_explore_result_node_refs(
-            effective_metadata.get("explore_result_node_refs")
-        ),
-        "decision_scope": normalize_todo_decision_scope(
-            effective_metadata.get("decision_scope")
-        ),
-        "required_decision_scopes": normalize_todo_required_decision_scopes(
-            effective_metadata.get("required_decision_scopes")
-        ),
-        "decision_outcome": effective_metadata.get("decision_outcome"),
-        "decision_scope_outcomes": normalize_todo_decision_scope_outcomes(
-            effective_metadata.get("decision_scope_outcomes")
-        ),
-        "blocks_agent": normalize_todo_blocks_agent(
-            effective_metadata.get("blocks_agent")
-        ),
-        "excluded_agents": normalize_todo_excluded_agents(
-            effective_metadata.get("excluded_agents")
-        ),
-        "global_gate": normalize_todo_global_gate(
-            effective_metadata.get("global_gate")
-        ),
-        "unblocks_todo_id": normalize_todo_id(
-            effective_metadata.get("unblocks_todo_id")
-        ),
-        "successor_todo_ids": normalize_todo_id_list(
-            effective_metadata.get("successor_todo_ids")
-        ),
-        "completion_continuation": normalize_todo_completion_continuation(
-            effective_metadata.get("completion_continuation")
-        ),
-        "completion_recovery": normalize_todo_completion_recovery(
-            effective_metadata.get("completion_recovery")
-        ),
-        "completion_receipt_id": effective_metadata.get("completion_receipt_id"),
-        "resume_when": normalize_todo_resume_when(
-            effective_metadata.get("resume_when")
-        ),
-        "resume_monitor_generation": effective_metadata.get(
-            "resume_monitor_generation"
-        ),
-        "no_followup": normalize_todo_no_followup(
-            effective_metadata.get("no_followup")
-        ),
-        "target_key": effective_metadata.get("target_key"),
-        "cadence": effective_metadata.get("cadence"),
-        "next_due_at": effective_metadata.get("next_due_at"),
-        "expires_at": effective_metadata.get("expires_at"),
-        "watch_only": effective_metadata.get("watch_only"),
-        "material_change_generation": effective_metadata.get(
-            "material_change_generation"
-        ),
-    }