perf(goals): resolve checkpoint context in one TypeScript request

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

@@ -1220,10 +1220,19 @@ authority. Missing, changed, non-private or over-64-MiB files fail closed.
 After a handler may have committed, an unverifiable response stays ambiguous
 and requires exact receipt readback, never automatic mutation retry.
 
-This removes the immediate transport ceiling, not the cost of projecting a
-complete multi-megabyte basis. The next measured T3 cut should combine the
-canonical source read and checkpoint reduction inside one TypeScript call, then
-offer a versioned manifest with bounded pages for human/Agent inspection.
+Source read and read/check reduction now compose inside one TypeScript request.
+Both context reading and commit preflight use one checkpoint request instead of
+two; the complete authoritative facts no longer return to Python only to be sent
+back for reduction. The existing reducer, full-basis receipt, source locks and
+final provider-fenced commit remain the owners. Reduction follows release of
+the optimistic read fence; the final commit still rereads under its own fence.
+Python retains local source IO and receipt persistence. The unused source-only
+and standalone evaluation effects are retired with their adapter calls; commit
+and replay retain their existing single-request boundaries.
+
+This removes the intermediate full-fact round trip, not the cost of projecting
+and returning a complete multi-megabyte basis. The remaining measured T3 cut is
+a versioned manifest with bounded pages for human/Agent inspection.
 Every page must bind to the same source head and disclose omitted components;
 the receipt must still hash the complete relevant Todo/dependency, User Todo,
 Goal prose, acceptance and vision basis. A display limit must never become a

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

@@ -851,6 +851,25 @@ codec；较大的完整事实快照通过私有临时文件和摘要绑定的引
 内存，也不证明分布式执行。游标/checkpoint 归约保留为以测量驱动、完整源语义一致
 为前提的后续工作，不再造 Python 规则。见[历史决策证据](ledger/typescript-control-plane-migration-v0/2026-09-22-replan-history-policy.zh-CN.md)。
 
+**Checkpoint 读取上下文的传输边界。** 完整 Goal prose 和已归档 Todo 事实可能
+同时超过 2 MiB 请求与响应边界。Checkpoint 来源读取、归约、重放检查和提交显式
+使用同 UID 私有文件及字节数、SHA-256 摘要绑定；其他 effect 的默认传输边界不变。
+文件缺失、改变、权限不符或超过 64 MiB 时拒绝。Handler 可能已经提交后，无法验证
+的响应仍按不明确结果处理，要求精确回执回读，不能自动重试写入。
+
+来源读取和 read/check 归约现在在一次 TypeScript 请求内完成。上下文读取与提交前
+检查各从两次 checkpoint 请求降为一次，完整权威事实不再先返回 Python 再传回 TS。
+复用原 reducer、完整 basis 回执、来源锁及最终 provider 围栏提交；乐观读取释放
+provider 围栏后再归约，最终提交仍在自己的围栏内重读。Python 保留本地来源 IO 和
+回执持久化。来源专用、独立 evaluate effect 及其适配调用一并退役；commit 与 replay
+仍各使用一次请求。这消除中间完整事实往返，不代表多 MiB basis 的组装和返回成本
+已经消失，也不改变 File/SQLite、legacy Markdown 的回执、权限或 provider 默认值。
+
+剩余以测量驱动的 T3 工作是提供有版本的 manifest 与有界展示分页。每页绑定同一
+来源 head，并说明省略部分；回执仍须哈希完整相关 Todo/依赖、User Todo、Goal prose、
+acceptance 和 vision。展示限制不能变成结算限制。在 legacy、File、SQLite 上完成
+语义一致性与 stale-head 恢复验证前，保留当前完整读取路径。
+
 **恢复边界（2026-09-22）。** [authority archive 命令](../../reference/authority-archive.md)
 由现有 TS coordination owner 负责历史校验、状态 delta 重建和可重入恢复；Python
 只解析 CLI 路径、传递请求并展示紧凑结果。复用 state-log codec，避免各 provider

loopx/control_plane/effect_runtime.py

@@ -38,8 +38,7 @@
 MAX_LOCAL_SNAPSHOT_BYTES = 64 * 1024 * 1024
 LOCAL_SNAPSHOT_METHODS = frozenset({
     "todo.context.page",
-    "goal.checkpoint_read_context.source",
-    "goal.checkpoint_read_context.evaluate",
+    "goal.checkpoint_read_context.resolve",
     "goal.checkpoint_read_context.commit",
     "goal.checkpoint_read_context.inspect_replay",
     "performance_diagnosis.inspect",

loopx/control_plane/effect_runtime_handlers.ts

@@ -350,8 +350,7 @@ export function createEffectRuntimeHandlers(
     ["work_item.delivery_response.project", lazyHandler(() => import("./work_items/delivery_history.ts"), ({projectDeliveryResponse}) => projectDeliveryResponse)],
     ["work_item.delivery_claim.validate", lazyHandler(() => import("./work_items/delivery_outcome.ts"), ({validateDeliveryClaim}) => validateDeliveryClaim)],
     ["goal.vision_checkpoint.evaluate", lazyHandler(() => import("./goals/vision_checkpoint.ts"), ({buildVisionCheckpoint}) => buildVisionCheckpoint)],
-    ["goal.checkpoint_read_context.evaluate", lazyHandler(() => import("./goals/checkpoint_read_context.ts"), ({evaluateCheckpointReadContext}) => evaluateCheckpointReadContext)],
-    ["goal.checkpoint_read_context.source", lazyHandler(() => import("./goals/checkpoint_authority.ts"), ({readCheckpointAuthority}) => readCheckpointAuthority)],
+    ["goal.checkpoint_read_context.resolve", lazyHandler(() => import("./goals/checkpoint_authority.ts"), ({resolveCheckpointReadContext}) => resolveCheckpointReadContext)],
     ["goal.checkpoint_read_context.commit", lazyHandler(() => import("./goals/checkpoint_commit.ts"), ({commitCheckpoint}) => commitCheckpoint)],
     ["goal.checkpoint_read_context.inspect_replay", lazyHandler(() => import("./goals/checkpoint_commit.ts"), ({inspectCheckpointReplay}) => inspectCheckpointReplay)],
     ["goal.vision_wait.coverage", lazyHandler(() => import("./goals/vision_wait_coverage.ts"), ({projectVisionWaitCoverage}) => projectVisionWaitCoverage)],

loopx/control_plane/effect_runtime_server.ts

@@ -29,8 +29,7 @@ const MAX_INLINE_RESPONSE_BYTES = 2 * 1024 * 1024;
 // Explicit opt-in: ordinary effects retain the 2 MiB request/response wire.
 const LOCAL_SNAPSHOT_METHODS = new Set([
   "todo.context.page",
-  "goal.checkpoint_read_context.source",
-  "goal.checkpoint_read_context.evaluate",
+  "goal.checkpoint_read_context.resolve",
   "goal.checkpoint_read_context.commit",
   "goal.checkpoint_read_context.inspect_replay",
   "performance_diagnosis.inspect",

loopx/control_plane/goals/checkpoint_authority.ts

@@ -10,6 +10,7 @@ import {openRuntimeAuthorityStore, requireLocalAuthorityRuntimeRoot} from "../co
 import {loadLegacyCoordinationWriterFence} from "../coordination/legacy_writer_fence.ts";
 import {indexCoordinationProjectionTodos, validateCoordinationTodoReadModel} from "../coordination/coordination_projection.ts";
 import {readGoalAcceptance} from "./acceptance_contract.ts";
+import {evaluateCheckpointReadContext} from "./checkpoint_read_context.ts";
 
 export async function withCheckpointAuthority(
   root: string, goalId: string, facts: JsonObject, save: (facts: JsonObject) => JsonObject,
@@ -40,11 +41,15 @@ export async function withCheckpointAuthority(
 
 /** Called while the Python adapter holds the local source locks. Optimistic
  * receipts are allowed to go stale after this operation returns. */
-export async function readCheckpointAuthority(value: unknown): Promise<JsonObject> {
-  const request = requireJsonObject(value, "checkpoint source request");
+export async function resolveCheckpointReadContext(value: unknown): Promise<JsonObject> {
+  const request = requireJsonObject(value, "checkpoint read context request");
   const root = requireLocalAuthorityRuntimeRoot(request.runtime_root);
-  const goalId = goalPathSegment(request.goal_id);
+  const identity = requireJsonObject(request.identity, "identity");
+  const goalId = goalPathSegment(identity.goal_id);
   const facts = requireJsonObject(request.facts, "checkpoint facts");
   requireNonEmptyString(requireJsonObject(facts.source, "checkpoint source").state_file, "state_file");
-  return await withCheckpointAuthority(root, goalId, facts, current => current);
+  const current = await withCheckpointAuthority(root, goalId, facts, current => current);
+  // Reduce the captured snapshot after releasing the read fence, as before.
+  // Only commitCheckpoint holds the provider fence through its final check/save.
+  return evaluateCheckpointReadContext({...request, facts: current});
 }

loopx/control_plane/goals/checkpoint_context_io.py

@@ -43,8 +43,10 @@ def _checkpoint_effect(method: str, request: dict[str, Any]) -> Any:
             "error_code": error.diagnostic_code, "reread_required": False}) from error
 
 
-def _evaluate(**request: Any) -> dict[str, Any]:
-    result = _checkpoint_effect("goal.checkpoint_read_context.evaluate", request)
+def _resolve(runtime_root: Path, **request: Any) -> dict[str, Any]:
+    result = _checkpoint_effect("goal.checkpoint_read_context.resolve", {
+        "runtime_root": str(runtime_root.resolve()), **request,
+    })
     if not isinstance(result, dict) or not isinstance(result.get("ok"), bool):
         raise RuntimeError("invalid typed checkpoint read context result")
     if not result["ok"]:
@@ -121,27 +123,6 @@ def _local_source_facts(
     }
 
 
-def _source_facts(
-    root: Path,
-    registry_path: Path,
-    state_file: Path,
-    identity: SettlementIdentity,
-    goal_ref: Mapping[str, Any] | None = None,
-) -> dict[str, Any]:
-    # The typed owner derives Todo and complete acceptance from one head and
-    # fails closed after cutover. Local parsed Markdown cannot override it.
-    return _checkpoint_effect("goal.checkpoint_read_context.source", {
-        "runtime_root": str(root.resolve()), "goal_id": identity.goal_id,
-        "facts": _local_source_facts(
-            root,
-            registry_path,
-            state_file,
-            identity,
-            goal_ref,
-        ),
-    })
-
-
 def read_checkpoint_context(
     *, registry_path: Path, runtime_root_override: str | None, goal_id: str,
     agent_id: str, todo_id: str | None, turn_instance_id: str,
@@ -178,9 +159,9 @@ def read_checkpoint_context(
             raise ValueError("checkpoint-context requires the original committed Turn writeback")
         identity = readback.identity.value
         with _source_guard(root, goal_id, path):
-            result = _evaluate(phase="read", identity=identity.as_dict(), prior=readback.writeback_run,
+            result = _resolve(root, phase="read", identity=identity.as_dict(), prior=readback.writeback_run,
                 read_context_id=uuid4().hex, dependency_todo_ids=dependency_todo_ids or [],
-                facts=_source_facts(root, registry_path, path, identity, goal_ref))
+                facts=_local_source_facts(root, registry_path, path, identity, goal_ref))
             receipt = result.pop("receipt")
             atomic_write_json(_receipt_path(root, identity), receipt)
     return {**result, "read_context_id": receipt["read_context_id"], "settlement_identity": identity.as_dict(),
@@ -203,8 +184,8 @@ def checkpoint_commit_guard(
             receipt = json.loads(_receipt_path(runtime_root, identity).read_text(encoding="utf-8"))
         except FileNotFoundError:
             receipt = None
-        result = _evaluate(phase="check", identity=identity.as_dict(), read_context_id=read_context_id,
-            receipt=receipt, facts=_source_facts(
+        result = _resolve(runtime_root, phase="check", identity=identity.as_dict(), read_context_id=read_context_id,
+            receipt=receipt, facts=_local_source_facts(
                 runtime_root,
                 registry_path,
                 state_file,

loopx/semantics/project_registry_io_manifest_v1.json

@@ -1143,7 +1143,7 @@
     },
     {
       "site": "loopx/control_plane/goals/checkpoint_context_io.py::<module>.read_checkpoint_context::codec_read:load_registry#1",
-      "line": 156,
+      "line": 137,
       "column": 16,
       "kind": "codec_read",
       "api": "load_registry",

tests/control_plane/test_checkpoint_read_context.py

@@ -132,15 +132,14 @@ def test_context_reads_real_canonical_todo_and_owner_acceptance(tmp_path, monkey
     from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
     from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
     from loopx.control_plane.coordination.local_authority_shadow_projection import canonical_bytes
-    from loopx.control_plane.goals.checkpoint_context_io import _source_facts, _source_guard
-    from loopx.control_plane.quota.settlement import SettlementIdentity
+    from loopx.control_plane.goals.checkpoint_context_io import read_checkpoint_context
     import hashlib
 
     if provider == "sqlite":
         isolate_sqlite_runtime(tmp_path, monkeypatch)
         import tempfile
         monkeypatch.setattr(tempfile, "tempdir", str(tmp_path))
-    project, runtime, registry = _write_fixture(tmp_path)
+    project, runtime, registry, _, _, _ = _missing(tmp_path)
     state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
     todo = {"schema_version": "todo_item_v0", "todo_id": TODO_ID, "index": 1,
             "role": "agent", "status": "done", "done": True, "text": "Canonical delivered result",
@@ -153,12 +152,42 @@ def test_context_reads_real_canonical_todo_and_owner_acceptance(tmp_path, monkey
         "revision": 1, "digest": hashlib.sha256(canonical_bytes(document)).hexdigest(),
         "document": document, "bindings": [], "verification": None}
     initialize_canonical_authority(runtime, GOAL_ID, projection, state_path=state, provider=provider)
-    identity = SettlementIdentity(GOAL_ID, AGENT_ID, TODO_ID, TURN_ID)
-    with _source_guard(runtime, GOAL_ID, state):
-        facts = _source_facts(runtime, registry, state, identity)
-    assert facts["todos"][0]["text"] == "Canonical delivered result"
-    assert facts["acceptance"]["contract"]["objective"] == "Canonical owner objective"
-    assert facts["source"]["authority"] == f"{provider}_v0"
+    context = read_checkpoint_context(registry_path=registry, runtime_root_override=str(runtime),
+        goal_id=GOAL_ID, agent_id=AGENT_ID, todo_id=TODO_ID, turn_instance_id=TURN_ID)
+    assert context["basis"]["todo"]["text"] == "Canonical delivered result"
+    assert context["basis"]["goal"]["acceptance"]["contract"]["objective"] == "Canonical owner objective"
+    assert context["basis"]["goal"]["acceptance"]["contract"]["criteria"] == document["criteria"]
+    assert context["basis"]["source"]["authority"] == f"{provider}_v0"
+
+
+def test_read_and_check_resolve_checkpoint_once_without_retired_calls(tmp_path, monkeypatch):
+    from loopx.control_plane.goals import checkpoint_context_io
+    from loopx.control_plane.quota.settlement import SettlementIdentity
+
+    project, runtime, registry, _, _, _ = _missing(tmp_path)
+    state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+    calls = []
+    resolve = checkpoint_context_io._checkpoint_effect
+
+    def observe(method, request):
+        calls.append((method, request.get("phase")))
+        return resolve(method, request)
+
+    monkeypatch.setattr(checkpoint_context_io, "_checkpoint_effect", observe)
+    context = checkpoint_context_io.read_checkpoint_context(
+        registry_path=registry, runtime_root_override=str(runtime), goal_id=GOAL_ID,
+        agent_id=AGENT_ID, todo_id=TODO_ID, turn_instance_id=TURN_ID,
+    )
+    assert calls == [("goal.checkpoint_read_context.resolve", "read")]
+
+    identity = SettlementIdentity.from_runtime_payload(context["settlement_identity"])
+    calls.clear()
+    with checkpoint_context_io.checkpoint_commit_guard(
+        runtime_root=runtime, registry_path=registry, state_file=state,
+        identity=identity, read_context_id=context["read_context_id"],
+    ) as checked:
+        assert checked["ok"]
+    assert calls == [("goal.checkpoint_read_context.resolve", "check")]
 
 
 def test_source_writers_remain_excluded_until_checkpoint_append(tmp_path, monkeypatch):