perf(coordination): read complete canonical snapshots via private transport

docs/reference/canonical-snapshot-pagination.md

@@ -1,15 +1,28 @@
 # Canonical Todo snapshot pagination
 
 After whole-Goal promotion, collection consumers use
-`coordination.local_authority.todo_snapshot_page`. Python assembles its pages
+`coordination.local_authority.todo_snapshot`. Python assembles its result
 into the existing Todo-list result before returning to CLI, status, Turn, Chat,
 team planning or projection delivery. Before promotion, the durable writer
 fence is absent and the existing legacy path remains unchanged.
 
 ## Observable contract
 
-A large, valid canonical collection no longer has to fit in a single RPC
-response. Each page contains complete records, up to 4,096 Todo/lease records and
+A large, valid canonical collection no longer has to fit in a single inline RPC
+response. The complete-read entrypoint uses the existing opt-in same-UID private
+file transport when needed. It validates the provider head once and returns a
+complete snapshot when its UTF-8 JSON fits within 62 MiB, reserving the remaining
+2 MiB of the existing 64 MiB private-file capacity for the RPC envelope. Smaller
+responses still travel inline. File ownership, permissions, byte count and SHA-256
+are verified before Python admits the result; the temporary files are reclaimed
+after the call. No provider head is cached between calls.
+
+When a complete snapshot exceeds that capacity, the same owner returns the
+original bounded first page instead. Subsequent requests retain that page's
+continuation, so a large valid population remains readable without raising a
+transport limit or retrying a lost oversized response. The standalone wire-only
+`coordination.local_authority.todo_snapshot_page` endpoint remains compatible.
+Its pages, and capacity fallback pages, contain up to 4,096 Todo/lease records and
 1,792 KiB of UTF-8 JSON. The generic RPC response ceiling remains 2 MiB, leaving 256 KiB for its
 envelope. Normal collections can still complete in one page. A record
 and its acceptance guard travel together; records are never shortened to meet a
@@ -43,8 +56,9 @@ A page carries the same snapshot, collection metadata, complete `todos`, optiona
 `leases`, page-local `goal_acceptance_work_guards`, and `next` (null at the end).
 Metadata includes the existing read-model declaration, acceptance projection,
 handoff mode and requested projection readback. The old direct TS `todo_list`
-endpoint remains compatible; the shipped Python collection adapter now uses
-pages, and its public result shape stays unchanged.
+endpoint remains compatible. The complete-read and wire-only entrypoints share
+the same request, snapshot, metadata and page schemas, including a null `next`
+for a complete answer. The Python adapter's public result shape stays unchanged.
 
 ## Consistency and recovery
 
@@ -68,17 +82,21 @@ an unreadable identity remains an identity failure. Selected SQLite and
 PostgreSQL profiles retain their existing identity checks and provider failures.
 No failure path falls back to a display file.
 
-A single record or metadata envelope that cannot fit is rejected with
+A single record or metadata envelope that cannot fit a fallback page is rejected with
 `canonical_snapshot_record_too_large` or
 `canonical_snapshot_metadata_too_large`. These explicit limits do not qualify
-arbitrarily large individual records. Projection confirmation remains tied to
+arbitrarily large individual records. A complete snapshot within private-file
+capacity can carry records or metadata beyond the inline page limit; the
+standalone wire-only endpoint retains its previous rejection behavior.
+Projection confirmation remains tied to
 the snapshot read; pending delivery must not trigger a second business mutation.
 
 ## Cost and qualification boundary
 
 This implementation bounds **transport responses**, not the whole collection's
-memory or provider IO. Each page loads and validates the current provider head;
-Python retains the assembled collection. A busy writer may force a caller to
+memory or provider IO. A complete private-file read loads and validates the head
+once. Capacity fallback and standalone wire pages load and validate the current
+head on every page; Python retains the assembled collection. A busy writer may force a caller to
 restart. Database cursor streaming or retained read transactions would need a
 separate provider lifecycle/cleanup contract and are not implied here.
 
@@ -89,7 +107,8 @@ The remaining prefix search retains a continuation and is monotone. Raising the
 RPC budget or dropping metadata is not the recovery strategy.
 
 Qualification covers native/imported complex populations, archives, leases,
-acceptance guards, overlapping commits, query/incarnation mismatch, malformed
+acceptance guards, complete/wire result parity, capacity fallback, store replacement
+during complete admission, overlapping commits, query/incarnation mismatch, malformed
 continuations, Unicode ordering and real File/SQLite RPC-to-CLI readback.
 The shared authority-store conformance registers the same snapshot tests for
 PostgreSQL and other real providers. The real PostgreSQL suite requires an

loopx/control_plane/coordination/canonical_snapshot.py

@@ -14,7 +14,7 @@
     LOCAL_COORDINATION_TODO_SNAPSHOT_PAGE_REQUEST_SCHEMA as REQUEST_SCHEMA,
     LOCAL_COORDINATION_TODO_SNAPSHOT_PAGE_RESULT_SCHEMA as RESULT_SCHEMA,
 )
-METHOD = "coordination.local_authority.todo_snapshot_page"
+METHOD = "coordination.local_authority.todo_snapshot"
 
 
 def read_canonical_snapshot(
@@ -55,6 +55,7 @@ def require(condition: bool, message: str) -> None:
                     "after": after,
                 },
                 timeout=timeout,
+                large_local_snapshot=True,
             )
             require(isinstance(page, dict), "snapshot page is not an object")
             if page.get("status") != "page":

loopx/control_plane/coordination/canonical_snapshot_page.ts

@@ -2,6 +2,7 @@
  * provider revision, never permission to combine independently current heads. */
 import type {JsonObject} from "../effect_program.ts";
 import {requireJsonObject} from "../runtime_decode.ts";
+import {MAX_LOCAL_SNAPSHOT_BYTES} from "../effect_runtime_snapshot.ts";
 import {projectGoalAcceptanceWorkGuards} from "../goals/acceptance_contract.ts";
 import {decodeProjectionReadback, confirmProjectionReadback} from "../todos/projection_delivery.ts";
 import {authorityStoreSourceAuthority, type AuthorityStore} from "./authority_store.ts";
@@ -16,6 +17,9 @@ export {CANONICAL_SNAPSHOT_PAGE_REQUEST, CANONICAL_SNAPSHOT_PAGE_RESULT};
 // Leave room for the RPC envelope below the unchanged 2 MiB transport limit.
 export const CANONICAL_SNAPSHOT_PAGE_BYTES = 1792 * 1024;
 export const CANONICAL_SNAPSHOT_PAGE_ITEMS = 4096;
+// The private file already has a 64 MiB transport ceiling. Reserve the entire
+// 2 MiB request ceiling for its response envelope (including the request id).
+export const CANONICAL_SNAPSHOT_LOCAL_BYTES = MAX_LOCAL_SNAPSHOT_BYTES - 2 * 1024 * 1024;
 interface Snapshot extends JsonObject {
   goal_id: string; store_identity: string; provider_revision: string; cursor: string;
   query_sha256: string; todo_count: number; lease_count: number;
@@ -58,7 +62,8 @@ function decodePosition(value: unknown): Position | null {
 
 /** Exported for provider conformance, using the same store contract as the
  * shipped local entrypoint. No test-only pagination implementation is used. */
-export async function readCanonicalSnapshotFromStore(value: unknown, store: AuthorityStore): Promise<JsonObject> {
+export async function readCanonicalSnapshotFromStore(value: unknown, store: AuthorityStore,
+  completeBytes = 0): Promise<JsonObject> {
   const input = requireJsonObject(value, "canonical snapshot request");
   check(hasExactAuthorityKeys(input, ["schema_version", "runtime_root", "goal_id", "include_leases", "projection_readback", "after"]) &&
     input.schema_version === CANONICAL_SNAPSHOT_PAGE_REQUEST && typeof input.include_leases === "boolean",
@@ -100,15 +105,11 @@ export async function readCanonicalSnapshotFromStore(value: unknown, store: Auth
   check(observedIdentity.status === "available" && observedIdentity.store_identity === identity.store_identity,
     "canonical_snapshot_changed", "store incarnation changed while loading the snapshot");
   const todoStart = after?.todo_offset ?? 0, leaseStart = after?.lease_offset ?? 0;
-  const todoIds = projection.todo_ids.slice(todoStart, todoStart + CANONICAL_SNAPSHOT_PAGE_ITEMS);
-  const leaseIds = leases?.lease_todo_ids.slice(leaseStart, leaseStart + CANONICAL_SNAPSHOT_PAGE_ITEMS - todoIds.length) ?? [];
-  const guards = acceptance.enabled === true ? projectGoalAcceptanceWorkGuards(head.head, goal, todoIds) : {};
   const metadata = {todo_read_model: readModel,
     ...(acceptance.enabled === true ? {goal_acceptance_contract: acceptance} : {}),
     ...(leases === null ? {} : {handoff_mode: head.head.handoff_mode ?? "legacy"}),
     ...(readback === null ? {} : {projection_readback: confirmProjectionReadback(readback, head.provider_revision)})};
-  function page(size: number): JsonObject {
-    const chosenTodos = todoIds.slice(0, size), chosenLeases = leaseIds.slice(0, Math.max(0, size - todoIds.length));
+  function result(chosenTodos: readonly string[], chosenLeases: readonly string[], guards: JsonObject): JsonObject {
     const todoEnd = todoStart + chosenTodos.length, leaseEnd = leaseStart + chosenLeases.length;
     return {...base, status: "page", snapshot, metadata,
       todos: chosenTodos.map(id => projection.todos.get(id)!),
@@ -119,6 +120,24 @@ export async function readCanonicalSnapshotFromStore(value: unknown, store: Auth
         {snapshot, todo_offset: todoEnd, lease_offset: leaseEnd}};
   }
   const bytes = (result: JsonObject) => Buffer.byteLength(JSON.stringify(result), "utf8");
+  // Only a new complete read may use the explicit private-file entrypoint.
+  // Reuse this same admission, metadata and identity; never cache a provider
+  // head between RPCs or substitute the older unguarded todo_list endpoint.
+  let completeGuards: JsonObject | null = null;
+  if (after === null && completeBytes > 0) {
+    completeGuards = acceptance.enabled === true
+      ? projectGoalAcceptanceWorkGuards(head.head, goal, projection.todo_ids) : {};
+    const complete = result(projection.todo_ids, leases?.lease_todo_ids ?? [], completeGuards);
+    if (bytes(complete) <= Math.min(completeBytes, CANONICAL_SNAPSHOT_LOCAL_BYTES)) return complete;
+  }
+  // A snapshot beyond the private-file capacity still uses the original page
+  // bounds and continuation checks. No lost oversized response or retry occurs.
+  const todoIds = projection.todo_ids.slice(todoStart, todoStart + CANONICAL_SNAPSHOT_PAGE_ITEMS);
+  const leaseIds = leases?.lease_todo_ids.slice(leaseStart, leaseStart + CANONICAL_SNAPSHOT_PAGE_ITEMS - todoIds.length) ?? [];
+  const guards = completeGuards ?? (acceptance.enabled === true
+    ? projectGoalAcceptanceWorkGuards(head.head, goal, todoIds) : {});
+  const page = (size: number) => result(todoIds.slice(0, size),
+    leaseIds.slice(0, Math.max(0, size - todoIds.length)), guards);
   const available = todoIds.length + leaseIds.length;
   const complete = page(available);
   if (bytes(complete) <= CANONICAL_SNAPSHOT_PAGE_BYTES) return complete;
@@ -136,17 +155,24 @@ export async function readCanonicalSnapshotFromStore(value: unknown, store: Auth
 /** Read-only RPC owner. A failed page never supplies a partial success or a
  * legacy fallback; the client must discard any earlier pages of that read. */
 export async function readCanonicalSnapshotPage(value: unknown,
-  dependencies: LocalAuthorityProviderDependencies = {}): Promise<JsonObject> {
+  dependencies: LocalAuthorityProviderDependencies = {}, completeBytes = 0): Promise<JsonObject> {
   try {
     const input = requireJsonObject(value, "canonical snapshot request");
     const root = requireLocalAuthorityRuntimeRoot(input.runtime_root);
     const goal = requireAuthorityStoreId(input.goal_id, "goal id");
     const store = await openRuntimeAuthorityStore(root, goal, dependencies, {existingOnly: true});
-    return await readCanonicalSnapshotFromStore(input, store);
+    return await readCanonicalSnapshotFromStore(input, store, completeBytes);
   } catch (error) {
     return {schema_version: CANONICAL_SNAPSHOT_PAGE_RESULT, status: "failed",
       reason_code: error instanceof SnapshotReadError ? error.code : "canonical_snapshot_request_invalid",
       reason: error instanceof Error ? error.message : "canonical snapshot read failed",
       decision_read_from_provider: true, legacy_fallback_used: false, ...localAuthorityOpenFailure(error)};
   }
 }
+
+/** Same canonical contract, opting into the existing same-UID file transport.
+ * Too-large collections keep bounded pages; failures never supply partial rows. */
+export async function readCanonicalSnapshot(value: unknown,
+  dependencies: LocalAuthorityProviderDependencies = {}): Promise<JsonObject> {
+  return readCanonicalSnapshotPage(value, dependencies, CANONICAL_SNAPSHOT_LOCAL_BYTES);
+}

loopx/control_plane/effect_runtime.py

@@ -45,6 +45,7 @@
     "collaboration.delegation.inventory_page",
     "benchmark.trajectory.codex-usage",
     "coordination.local_authority.todo_source",
+    "coordination.local_authority.todo_snapshot",
     "todo.context.page",
     "work_item.context.project",
     "work_item.interaction_reads.project",

loopx/control_plane/effect_runtime_handlers.ts

@@ -450,6 +450,7 @@ export function createEffectRuntimeHandlers(
     ["coordination.ownership_observation", lazyHandler(() => import("./coordination/ownership_observation.ts"), ({projectOwnershipObservation}) => projectOwnershipObservation)],
     ["coordination.local_authority.ownership_observation", lazyHandler(() => import("./coordination/local_authority_runtime.ts"), ({observeLocalCoordinationOwnership}) => observeLocalCoordinationOwnership)],
     ["coordination.local_authority.todo_snapshot_page", lazyHandler(() => import("./coordination/canonical_snapshot_page.ts"), ({readCanonicalSnapshotPage}) => readCanonicalSnapshotPage)],
+    ["coordination.local_authority.todo_snapshot", lazyHandler(() => import("./coordination/canonical_snapshot_page.ts"), ({readCanonicalSnapshot}) => readCanonicalSnapshot)],
     ["coordination.local_authority.todo_list", lazyHandler(() => import("./coordination/local_authority_read.ts"), ({listLocalCoordinationTodos}) => listLocalCoordinationTodos)],
     [
       "coordination.local_authority.legacy_writer_fence.engage",

loopx/control_plane/effect_runtime_server.ts

@@ -31,6 +31,7 @@ const LOCAL_SNAPSHOT_METHODS = new Set([
   "collaboration.delegation.inventory_page",
   "benchmark.trajectory.codex-usage",
   "coordination.local_authority.todo_source",
+  "coordination.local_authority.todo_snapshot",
   "todo.context.page",
   "work_item.context.project",
   "work_item.interaction_reads.project",

tests/control_plane/test_canonical_snapshot.py

@@ -61,10 +61,11 @@ def pages():
 def read(responses, *, include_leases=True):
     calls = []
 
-    def rpc(method, request, *, timeout):
+    def rpc(method, request, *, timeout, large_local_snapshot):
         assert method == METHOD
         assert request["include_leases"] is include_leases
         assert timeout == 15
+        assert large_local_snapshot is True
         assert request["after"] == (None if not calls else calls[-1]["next"])
         response = responses[len(calls)]
         calls.append(deepcopy(response))

tests/control_plane/test_canonical_snapshot_integration.py

@@ -12,6 +12,7 @@
 )
 
 from loopx.control_plane.coordination import local_authority
+from loopx.control_plane import effect_runtime
 from loopx.control_plane.coordination.local_authority import (
     LocalCoordinationAuthorityUnavailable,
 )
@@ -154,8 +155,16 @@ def test_real_rpc_keeps_budget_and_cli_recovers_complete_display(
     # caller cannot classify that lost response as a safe rejection or retry.
     assert oversized.value.diagnostic_code == "runtime_response_ambiguous"
     measured = []
+    private_sizes = []
+    read_private = effect_runtime._read_local_snapshot_response
+
+    def capture_private(envelope, *args, **kwargs):
+        private_sizes.append(envelope["result_ref"]["byte_count"])
+        return read_private(envelope, *args, **kwargs)
 
     def capture(method, payload, **kwargs):
+        assert method == "coordination.local_authority.todo_snapshot"
+        assert kwargs["large_local_snapshot"] is True
         result = effect_runtime_result(method, payload, **kwargs)
         measured.append(
             len(json.dumps(result, ensure_ascii=False, separators=(",", ":")).encode())
@@ -164,9 +173,28 @@ def capture(method, payload, **kwargs):
 
     with monkeypatch.context() as patch:
         patch.setattr(local_authority, "effect_runtime_result", capture)
+        patch.setattr(effect_runtime, "_read_local_snapshot_response", capture_private)
         result = native_read(runtime)
-    assert len(measured) > 1 and max(measured) <= 1792 * 1024
+    assert len(measured) == 1 and measured[0] > MAX_RESPONSE_BYTES
+    assert len(private_sizes) == 1
+    assert MAX_RESPONSE_BYTES < private_sizes[0] <= effect_runtime.MAX_LOCAL_SNAPSHOT_BYTES
     assert result["todos"] == projection["todos"]
+    # The original wire-only endpoint keeps its exact bounds and complete
+    # assembly contract. Both paths return the same immutable canonical facts.
+    measured.clear()
+
+    def paged(_method, payload, **kwargs):
+        page = effect_runtime_result(
+            "coordination.local_authority.todo_snapshot_page", payload,
+            timeout=kwargs["timeout"],
+        )
+        measured.append(len(json.dumps(page, ensure_ascii=False, separators=(",", ":")).encode()))
+        return page
+
+    with monkeypatch.context() as patch:
+        patch.setattr(local_authority, "effect_runtime_result", paged)
+        assert native_read(runtime) == result
+    assert len(measured) > 1 and max(measured) <= 1792 * 1024
     # A rejected oversized read may close a socket with buffered response bytes.
     # It must not take down the shared runtime used by the paged successor.
     assert effect_runtime_result("runtime.ping", {})["pid"] == runtime_pid
@@ -205,7 +233,13 @@ def test_real_overlapping_commit_has_no_partial_or_legacy_success(
     calls = []
 
     def overlap(method, payload, **kwargs):
-        result = effect_runtime_result(method, payload, **kwargs)
+        assert method == "coordination.local_authority.todo_snapshot"
+        # Force the real bounded owner to exercise its continuation recovery,
+        # rather than a complete result which needs no second provider read.
+        result = effect_runtime_result(
+            "coordination.local_authority.todo_snapshot_page", payload,
+            timeout=kwargs["timeout"],
+        )
         calls.append(result)
         if len(calls) == 1:
             # A real writer wins between pages, with unchanged Todo data. Revision still matters.