perf(todos): reduce read startup and lineage costs

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

@@ -175,9 +175,16 @@ reader; supplied values still use the existing validators. Objective-only
 acceptance observations remain visible without configured checks.
 Canonical summaries prepare recovery history only for role groups that contain
 a resume condition. Succession still uses complete active and archived lineage.
+Role summaries reuse request-local lineage facts and the succession owner's
+existing context fields. That owner validates and indexes the complete graph
+before materializing requested evaluations. Source fingerprinting still reads
+every admitted byte and rechecks freshness; sequential reads avoid starting a
+thread pool in each process. Healthy read commands use canonical registry/path
+owners and defer mutation error imports until needed.
 The output measurements above retain their original controlled-transport scope.
-Current normal-IPC, package and complete-command evidence belongs to the exact
-revision reported in [#6874](https://github.com/loopx-project/loopx/pull/6874).
+[#6874](https://github.com/loopx-project/loopx/pull/6874) records the original
+normal-IPC, package and complete-command evidence. Follow-up measurements apply
+only to their reported revisions.
 Historical performance failures remain failures; neither fewer RPCs nor codec
 CPU savings waive the full-CLI gate or establish R3/D2/D3 adoption.
 The current warm-request miss remains an owner review hold. A passing full-CLI

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

@@ -142,8 +142,13 @@ Markdown 不变。仅三个已测 Todo 查询视图的首次信封 schema 迁移
 仍保留当前目标观察。
 Canonical 摘要仅在某角色组包含恢复条件时准备恢复历史。
 继任计算仍使用完整的活跃与归档血缘。
-上述输出测量保留原受控传输范围。当前正常 IPC、安装包及完整命令证据以
-[#6874](https://github.com/loopx-project/loopx/pull/6874) 记录的精确版本为准。
+角色摘要复用一次请求内的血缘事实和继任 owner 的既有上下文字段。
+该 owner 先校验并索引完整图，再生成需要返回的评估。
+源码指纹仍读取每个已准入字节并复查新鲜度；顺序读取省去每个进程的线程池创建。
+正常读取直接使用注册表和路径的 canonical owner，发生错误时才导入相关错误类型。
+上述输出测量保留原受控传输范围。
+[#6874](https://github.com/loopx-project/loopx/pull/6874) 记录原始正常 IPC、安装包及完整命令证据。
+后续测量仅适用于各自记录的精确版本。
 历史性能失败仍是失败；RPC 减少或 codec CPU 改善不能免除完整 CLI 门禁，
 也不证明 R3／D2／D3 采用。
 当前 warm 请求未达标仍需 owner 评审。完整 CLI 配对通过不免除 2 ms 目标。

loopx/cli_commands/todo_event.py

@@ -4,19 +4,8 @@
 from collections.abc import Callable, Mapping
 from pathlib import Path
 
-from ..control_plane.coordination.local_authority import (
-    LocalCoordinationAuthorityRejection,
-)
-from ..control_plane.todos.external_wait_contract import TodoExternalWaitAuthoringError
-from ..control_plane.todos.handoff_mode import HandoffModeError
 from ..control_plane.todos.contract import decision_scope_metadata_value
-from ..control_plane.work_items.task_lease import TaskLeaseError
 from ..file_lock import lock_timeout_error_fields
-from ..control_plane.coordination.legacy_writer_fence import LegacyCoordinationWriterFenced
-from ..control_plane.coordination.shadow_management import ShadowManagementError
-from ..control_plane.coordination.legacy_writer_fence import ActiveStateAuthorityMutationError
-from ..control_plane.coordination.local_authority import LocalCoordinationAuthorityUnavailable
-from .todo_argument_validation import TodoClaimArgumentError
 
 
 RolloutEventAppender = Callable[..., dict[str, object]]
@@ -35,6 +24,18 @@ def todo_error_payload(
     args: argparse.Namespace, exc: Exception, *, registry_path: Path,
     runtime_root_arg: str | None,
 ) -> dict[str, object]:
+    from ..control_plane.coordination.local_authority import (
+        LocalCoordinationAuthorityRejection, LocalCoordinationAuthorityUnavailable,
+    )
+    from ..control_plane.coordination.legacy_writer_fence import (
+        ActiveStateAuthorityMutationError, LegacyCoordinationWriterFenced,
+    )
+    from ..control_plane.coordination.shadow_management import ShadowManagementError
+    from ..control_plane.todos.external_wait_contract import TodoExternalWaitAuthoringError
+    from ..control_plane.todos.handoff_mode import HandoffModeError
+    from ..control_plane.work_items.task_lease import TaskLeaseError
+    from .todo_argument_validation import TodoClaimArgumentError
+
     payload: dict[str, object] = {
         "ok": False,
         "dry_run": not bool(args.execute)

loopx/cli_rollout.py

@@ -3,7 +3,7 @@
 from collections.abc import Mapping
 from pathlib import Path
 
-from .history import load_registry
+from .control_plane.projects.registry_codec import load_registry
 from .paths import resolve_runtime_root
 from .rollout_event_log import (
     append_rollout_event,

loopx/control_plane/effect_runtime.py

@@ -337,9 +337,9 @@ def _runtime_fingerprint_for_snapshot(
     digest.update(os.fsencode(os.path.normcase(root)))
     digest.update(b"\0")
     paths = (source_root / relative for relative, *_metadata in snapshot)
-    # Reads may finish out of order; hash the same relative names and original
-    # bytes in snapshot order. No disk cache or skipped freshness check.
-    with closing(iter_binary_file_reads(paths)) as reads:
+    # A fresh CLI hashes these small package files; starting a pool costs more
+    # than sequential local reads. Keep every byte and the freshness recheck.
+    with closing(iter_binary_file_reads(paths, max_workers=1)) as reads:
         for (relative, *_metadata), read in zip(snapshot, reads, strict=True):
             if read.error is not None:
                 raise read.error

loopx/control_plane/todos/list_readback.py

@@ -8,7 +8,7 @@
 from pathlib import Path
 from typing import Any, Literal
 
-from ...history import load_registry
+from ..projects.registry_codec import load_registry
 from ...paths import resolve_runtime_root
 from ...rollout_event_log import load_rollout_events, rollout_event_log_path
 from ..goals.state_resolution import resolve_goal_state

loopx/control_plane/todos/read_projection.ts

@@ -9,7 +9,7 @@ import {schedulerTimestampMilliseconds} from "../scheduler/state_store.ts";
 import {evaluateTodoResumeConditionsFromSource} from "./resume_condition.ts";
 import {readTodoPriority, todoPriorityRank} from "./priority.ts";
 import {evaluateTodoSuccession, validateTodoSuccession, SUCCESSION_FACT_COLUMNS,
-  SUCCESSION_EVALUATION_COLUMNS} from "./succession.ts";
+  SUCCESSION_EVALUATION_COLUMNS, SUCCESSION_CONTEXT_FIELDS} from "./succession.ts";
 import {projectTodoSummary, TODO_SUMMARY_PROJECTION_COLUMNS} from "./summary_projection.ts";
 import {projectTodoHandoffContext} from "./handoff_projection.ts";
 import {projectStandingDecisions} from "./standing_decision.ts";
@@ -32,7 +32,7 @@ function successionFacts(item: JsonObject): JsonObject {
     route_flag: typeof item.route_continuation_replan_required === "boolean" ? item.route_continuation_replan_required : null,
     legacy_route_label: handoff && typeof item.route_continuation_replan_required !== "boolean"
       ? [item.action_kind, item.title, item.text].filter(v => text(v).trim()).map(text).join(" ") : "",
-    context_fields: Object.keys(withoutNull(item)).filter(key => key !== "succession_evaluation").sort(authorityUnicodeCompare)};
+    context_fields: SUCCESSION_CONTEXT_FIELDS.filter(key => Object.prototype.propertyIsEnumerable.call(item, key) && item[key] != null)};
 }
 function compactRecord(item: JsonObject): JsonObject {
   const result: JsonObject = {index: item.index ?? null, done: Boolean(item.done), text: item.text ?? null};
@@ -150,7 +150,7 @@ export async function projectTodoReadGroup(input: JsonObject): Promise<JsonObjec
       else selected.set(key(row), count - 1);
     });
     try {
-      decisions = evaluateTodoSuccession([...source, ...facts]).slice(source.length);
+      decisions = evaluateTodoSuccession([...source, ...facts], source.length);
     } catch (error) {
       if (!(error instanceof EffectRuntimeRequestError)) throw error;
       return {schema_version: TODO_READ_RESULT, rejection: {phase: "succession", message: error.message}};
@@ -204,15 +204,16 @@ export async function projectCanonicalTodoFields(input: JsonObject): Promise<Jso
   const archived = new Set(all.filter(row => row.archive_state === "archive").map(row => row.todo_id));
   const acceptance = object(input.goal_acceptance_contract), guards = object(input.goal_acceptance_work_guards);
   const checksConfigured = configuredGoalChecks(acceptance) !== null;
+  const lineage = Array.isArray(input.lineage) ? input.lineage.map(value => requireJsonObject(value, "Todo lineage")) : all;
+  const lineageFacts = input.lineage_facts ?? lineage.map(successionFacts);
   const fields: JsonObject = {}, evaluations: JsonObject = {};
   for (const role of ["user", "agent"]) {
     const section = role === "agent" ? "Agent Todo" : "User Todo / Owner Review Reading Queue";
     const items = all.filter(row => (row.role === "user" ? "user" : "agent") === role && (input.retained === true || !archived.has(row.todo_id)))
       .map(row => structureCanonicalTodo({...row,
         ...(input.retained !== true && checksConfigured && guards[text(row.todo_id)] ? {goal_acceptance_guard: guards[text(row.todo_id)]} : {})}, role, section,
         input.retained === true ? null : 500, input.retained !== true));
-    const lineage = Array.isArray(input.lineage) ? input.lineage.map(value => requireJsonObject(value, "Todo lineage")) : all;
-    const grouped = await projectTodoReadGroup({...input, items, lineage, evaluated: false, role, source_section: section,
+    const grouped = await projectTodoReadGroup({...input, items, lineage, lineage_facts: lineageFacts, evaluated: false, role, source_section: section,
       item_limit: null, full_selection: true, resume_lineage: items.some(row => readResume(row.resume_when))
         ? all.map(row => structureCanonicalTodo(row, row.role ?? null, row.source_section ?? section)) : []});
     if (grouped.rejection) return grouped;

loopx/control_plane/todos/succession.ts

@@ -13,6 +13,7 @@ const CONTEXT_FIELDS = ["action_kind", "task_repository", "continuation_policy",
   "completed_at", "updated_at", "required_write_scopes", "required_capabilities", "target_capabilities",
   "explore_result_node_refs", "decision_scope", "required_decision_scopes", "unblocks_todo_id",
   "resume_when", "blocks_agent", "excluded_agents", "global_gate"] as const;
+export {CONTEXT_FIELDS as SUCCESSION_CONTEXT_FIELDS};
 const EVALUATION_SCHEMA = "todo_succession_evaluation_v0";
 interface Row {
   facts: JsonObject; id: string | null; status: "open" | "blocked" | "done" | "deferred";
@@ -92,7 +93,7 @@ export function indexInferredSuccessors(rows: readonly Pick<Row, "id" | "advance
 }
 
 /** Index inferred edges once; both completion and handoff use the same resolver. */
-export function evaluateTodoSuccession(values: readonly unknown[]): JsonObject[] {
+export function evaluateTodoSuccession(values: readonly unknown[], resultStart = 0): JsonObject[] {
   const rows = values.map(decode), byId = new Map<string, Row>();
   const generations = new Set<string>();
   for (const row of rows) {
@@ -105,7 +106,7 @@ export function evaluateTodoSuccession(values: readonly unknown[]): JsonObject[]
     if (row.active || !byId.has(row.id)) byId.set(row.id, row);
   }
   const inferred = indexInferredSuccessors([...byId.values()]);
-  return rows.map(row => {
+  return rows.slice(resultStart).map(row => {
     const declared = [...new Set([...row.successors, ...(row.supersededBy ? [row.supersededBy] : [])])];
     // A retained archived target is still evidence. A missing/self target is not.
     const resolved = declared.filter(target => target !== row.id && byId.has(target));

loopx/rollout_event_log.py

@@ -252,7 +252,7 @@ def rollout_event_log_path(runtime_root: Path, goal_id: str) -> Path:
     # Goal ids are used as directory names throughout the control plane. Keep
     # this shared path helper fail-closed so every reader and writer inherits
     # the same single-segment boundary.
-    from .history import validate_goal_id_path_segment
+    from .paths import validate_goal_id_path_segment
 
     safe_goal_id = validate_goal_id_path_segment(goal_id)
     return (

tests/control_plane/test_todo_list_read_transport.py

@@ -58,6 +58,7 @@ def run(*tail, format_name="json", expected_exit=0, isolated=False):
                 "import sys;from loopx.entrypoint import main;code=main(sys.argv[1:]);"
                 "assert code==0;"
                 "assert 'loopx.control_plane.todos.mutation_api' not in sys.modules;"
+                "assert 'loopx.history' not in sys.modules;"
                 "assert 'loopx.state_refresh' not in sys.modules;"
                 "assert 'loopx.cli_commands.todo' not in sys.modules;"
                 "assert 'loopx.capabilities.periodic_report.post_writeback_hook' not in sys.modules"

tests/control_plane/test_turn_journal_runtime_readiness.py

@@ -123,25 +123,25 @@ def test_runtime_fingerprint_rescans_when_a_snapshotted_file_disappears_while_re
     later.write_text("export const later = true;\n", encoding="utf-8")
     original_read_bytes = Path.read_bytes
     reads: list[str] = []
-    later_prefetched = Event()
+    first_read = Event()
 
-    def remove_later_after_prefetch(path: Path) -> bytes:
+    def remove_first_after_reading_last(path: Path) -> bytes:
         content = original_read_bytes(path)
-        if path == later:
+        if path == first:
             reads.append(path.name)
-            later_prefetched.set()
+            first_read.set()
         else:
-            if later.exists():
-                assert later_prefetched.wait(timeout=5)
-                later.unlink()
+            if first.exists():
+                assert first_read.wait(timeout=5)
+                first.unlink()
             reads.append(path.name)
         return content
 
     monkeypatch.setattr(effect_runtime, "_control_plane_root", lambda: tmp_path)
-    monkeypatch.setattr(Path, "read_bytes", remove_later_after_prefetch)
+    monkeypatch.setattr(Path, "read_bytes", remove_first_after_reading_last)
 
     assert len(effect_runtime._runtime_fingerprint()) == 64
-    assert reads == ["later.ts", "first.ts", "first.ts"]
+    assert reads == ["first.ts", "later.ts", "later.ts"]
 
 
 def _install_persistent_stat_read_churn(
@@ -155,31 +155,31 @@ def _install_persistent_stat_read_churn(
     original_scan = effect_runtime._scan_runtime_source_files
     original_read_bytes = Path.read_bytes
     scans: list[tuple[str, ...]] = []
-    later_prefetched = Event()
-    first_reads = 0
+    first_read = Event()
+    later_reads = 0
 
     def record_scan(root: Path) -> tuple[str, ...]:
         files = original_scan(root)
         scans.append(files)
         return files
 
-    def churn_after_each_first_read(path: Path) -> bytes:
-        nonlocal first_reads
+    def churn_after_each_later_read(path: Path) -> bytes:
+        nonlocal later_reads
         content = original_read_bytes(path)
-        if path == later:
-            later_prefetched.set()
-        else:
-            first_reads += 1
-        if first_reads == 1 and path == first:
-            assert later_prefetched.wait(timeout=5)
-            later.unlink()
-        elif first_reads == 2 and path == first:
-            later.write_text("export const later = true;\n", encoding="utf-8")
+        if path == first:
+            first_read.set()
+        elif path == later:
+            later_reads += 1
+            if later_reads == 1:
+                assert first_read.wait(timeout=5)
+                first.unlink()
+            elif later_reads == 2:
+                first.write_text("export const first = true;\n", encoding="utf-8")
         return content
 
     monkeypatch.setattr(effect_runtime, "_control_plane_root", lambda: tmp_path)
     monkeypatch.setattr(effect_runtime, "_scan_runtime_source_files", record_scan)
-    monkeypatch.setattr(Path, "read_bytes", churn_after_each_first_read)
+    monkeypatch.setattr(Path, "read_bytes", churn_after_each_later_read)
     return scans
 
 
@@ -204,7 +204,7 @@ def test_runtime_source_churn_has_a_stable_readiness_diagnostic(
     )
     assert scans == [
         ("first.ts", "later.ts"),
-        ("first.ts",),
+        ("later.ts",),
         ("first.ts", "later.ts"),
     ]
 
@@ -221,7 +221,7 @@ def test_runtime_request_source_churn_raises_a_stable_startup_diagnostic(
     assert error.value.diagnostic_code == "packaged_runtime_source_unstable"
     assert scans == [
         ("first.ts", "later.ts"),
-        ("first.ts",),
+        ("later.ts",),
         ("first.ts", "later.ts"),
     ]
 

tests/control_plane_ts/todo_succession.test.ts

@@ -81,6 +81,22 @@ test("filtering preserves full-source evidence; editing relevant facts invalidat
   assert.throws(() => evaluateTodoSuccession([source, source]), /duplicate succession identity/);
 });
 
+test("selected evaluations retain omitted dependencies and validate the complete source", () => {
+  const rows = [row("todo_archive", {active: false}), row("todo_inferred", {resumes: "todo_source"}),
+    row("todo_source", {successors: ["todo_archive"]})];
+  const complete = evaluateTodoSuccession(rows);
+  for (const start of [0, 2, rows.length]) {
+    assert.deepEqual(evaluateTodoSuccession(rows, start), complete.slice(start));
+  }
+  const [selected] = evaluateTodoSuccession(rows, 2);
+  assert.deepEqual(selected.successor_todo_ids, ["todo_archive", "todo_inferred"]);
+  assert.equal(selected.successor_gap, false);
+  for (const start of [2, rows.length]) {
+    assert.throws(() => evaluateTodoSuccession([{...rows[0], no_followup: "invalid"}, ...rows.slice(1)], start), /no_followup/);
+    assert.throws(() => evaluateTodoSuccession([rows[0], rows[0], rows[2]], start), /duplicate succession identity/);
+  }
+});
+
 const closed = {status: "done", watch_only: false, no_followup: true, successor_gap: false, replan: false, handoff_state: null};
 const closure = (rows: unknown[], full_selection = true) => projectTodoClosure({schema_version: "todo_closure_request_v0",
   role: "agent", source_section: "Agent Todo", full_selection, rows});