From 9a1b7f8a59588bd76ba6421d8d3b595a4179980a Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 18 Oct 2026 23:54:26 +0800 Subject: [PATCH] perf(coordination): read complete canonical snapshots via private transport Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../canonical-snapshot-pagination.md | 39 ++++++++++---- .../coordination/canonical_snapshot.py | 3 +- .../coordination/canonical_snapshot_page.ts | 42 ++++++++++++--- loopx/control_plane/effect_runtime.py | 1 + .../control_plane/effect_runtime_handlers.ts | 1 + loopx/control_plane/effect_runtime_server.ts | 1 + .../control_plane/test_canonical_snapshot.py | 3 +- .../test_canonical_snapshot_integration.py | 38 +++++++++++++- .../test_local_coordination_authority.py | 40 ++++++++------- .../canonical_snapshot_conformance.ts | 51 ++++++++++++++++++- .../canonical_snapshot_page.test.ts | 10 +++- 11 files changed, 186 insertions(+), 43 deletions(-) diff --git a/docs/reference/canonical-snapshot-pagination.md b/docs/reference/canonical-snapshot-pagination.md index cf2f64a50b..c65d692c3e 100644 --- a/docs/reference/canonical-snapshot-pagination.md +++ b/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 diff --git a/loopx/control_plane/coordination/canonical_snapshot.py b/loopx/control_plane/coordination/canonical_snapshot.py index 95a9bc9699..d4353118bc 100644 --- a/loopx/control_plane/coordination/canonical_snapshot.py +++ b/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": diff --git a/loopx/control_plane/coordination/canonical_snapshot_page.ts b/loopx/control_plane/coordination/canonical_snapshot_page.ts index 336fb3b2f9..5814fa1887 100644 --- a/loopx/control_plane/coordination/canonical_snapshot_page.ts +++ b/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 { +export async function readCanonicalSnapshotFromStore(value: unknown, store: AuthorityStore, + completeBytes = 0): Promise { 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,13 +155,13 @@ 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 { + dependencies: LocalAuthorityProviderDependencies = {}, completeBytes = 0): Promise { 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", @@ -150,3 +169,10 @@ export async function readCanonicalSnapshotPage(value: unknown, 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 { + return readCanonicalSnapshotPage(value, dependencies, CANONICAL_SNAPSHOT_LOCAL_BYTES); +} diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index afe7fde75c..7e18939128 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/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", diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 5f25e1c69f..ab48e0118a 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/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", diff --git a/loopx/control_plane/effect_runtime_server.ts b/loopx/control_plane/effect_runtime_server.ts index 81eb661525..57852c25d1 100644 --- a/loopx/control_plane/effect_runtime_server.ts +++ b/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", diff --git a/tests/control_plane/test_canonical_snapshot.py b/tests/control_plane/test_canonical_snapshot.py index 183a04075c..a5f4d82f58 100644 --- a/tests/control_plane/test_canonical_snapshot.py +++ b/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)) diff --git a/tests/control_plane/test_canonical_snapshot_integration.py b/tests/control_plane/test_canonical_snapshot_integration.py index 421aa5496b..786491906a 100644 --- a/tests/control_plane/test_canonical_snapshot_integration.py +++ b/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. diff --git a/tests/control_plane/test_local_coordination_authority.py b/tests/control_plane/test_local_coordination_authority.py index 960b7dd1de..e73e677587 100644 --- a/tests/control_plane/test_local_coordination_authority.py +++ b/tests/control_plane/test_local_coordination_authority.py @@ -150,7 +150,9 @@ def test_engaged_fence_reads_typescript_provider_result( _engage_fence(tmp_path) calls: list[tuple[str, float]] = [] - def _read(method: str, _params: object, *, timeout: float) -> dict[str, object]: + def _read(method: str, _params: object, *, timeout: float, + large_local_snapshot: bool) -> dict[str, object]: + assert large_local_snapshot is True calls.append((method, timeout)) return single_snapshot_page({ "status": "loaded", @@ -173,7 +175,7 @@ def _read(method: str, _params: object, *, timeout: float) -> dict[str, object]: ) assert result is not None assert result["todos"][0]["todo_id"] == "todo_a" - assert calls == [("coordination.local_authority.todo_snapshot_page", 15.0)] + assert calls == [("coordination.local_authority.todo_snapshot", 15.0)] def test_promoted_claim_adapter_invokes_typescript_without_markdown_fallback( @@ -1485,12 +1487,12 @@ def count_authority_runtime_call( # The extra bounded crossing admits/replays before resolving private argv. assert terminal_phases == ["resolve_validation", "execute_validation", "applied"] assert runtime_calls == [ - "coordination.local_authority.todo_snapshot_page", + "coordination.local_authority.todo_snapshot", "coordination.local_authority.todo_terminal", "coordination.local_authority.todo_terminal", "coordination.local_authority.todo_terminal", - "coordination.local_authority.todo_snapshot_page", - "coordination.local_authority.todo_snapshot_page", + "coordination.local_authority.todo_snapshot", + "coordination.local_authority.todo_snapshot", ] successor_id = completed["generated_successor_todo_ids"][0] @@ -1511,10 +1513,10 @@ def count_authority_runtime_call( assert superseded["superseded"] is True assert superseded["projection_delivery"] == "delivered" assert runtime_calls == [ - "coordination.local_authority.todo_snapshot_page", + "coordination.local_authority.todo_snapshot", "coordination.local_authority.todo_terminal", - "coordination.local_authority.todo_snapshot_page", - "coordination.local_authority.todo_snapshot_page", + "coordination.local_authority.todo_snapshot", + "coordination.local_authority.todo_snapshot", ] canonical = read_canonical_todos_if_promoted( @@ -1542,10 +1544,10 @@ def count_authority_runtime_call( assert archived["moved_count"] == 2 assert archived["projection_delivery"] == "delivered" assert runtime_calls == [ - "coordination.local_authority.todo_snapshot_page", + "coordination.local_authority.todo_snapshot", "coordination.local_authority.todo_archive", - "coordination.local_authority.todo_snapshot_page", - "coordination.local_authority.todo_snapshot_page", + "coordination.local_authority.todo_snapshot", + "coordination.local_authority.todo_snapshot", "coordination.local_authority.todo_archive_ack", ] canonical_after_archive = read_canonical_todos_if_promoted( @@ -1575,10 +1577,10 @@ def count_authority_runtime_call( assert no_change["moved_count"] == 0 assert no_change["provider_revision"] == archive_revision assert runtime_calls == [ - "coordination.local_authority.todo_snapshot_page", + "coordination.local_authority.todo_snapshot", "coordination.local_authority.todo_archive", - "coordination.local_authority.todo_snapshot_page", - "coordination.local_authority.todo_snapshot_page", + "coordination.local_authority.todo_snapshot", + "coordination.local_authority.todo_snapshot", ] unchanged = read_canonical_todos_if_promoted( runtime_root=runtime_root, @@ -2009,7 +2011,7 @@ def _crash_projection(*_args: object, **_kwargs: object) -> dict[str, object]: with pytest.raises(OSError, match="projection delivery crash"): complete_goal_todo(**request) assert runtime_calls == [ - "coordination.local_authority.todo_snapshot_page", + "coordination.local_authority.todo_snapshot", "coordination.local_authority.todo_terminal", ] @@ -2022,12 +2024,12 @@ def _crash_projection(*_args: object, **_kwargs: object) -> dict[str, object]: assert replay["provider_status"] == "replayed" assert replay["idempotent_replay"] is True assert runtime_calls == [ - "coordination.local_authority.todo_snapshot_page", + "coordination.local_authority.todo_snapshot", "coordination.local_authority.todo_terminal", - "coordination.local_authority.todo_snapshot_page", + "coordination.local_authority.todo_snapshot", "coordination.local_authority.todo_terminal", - "coordination.local_authority.todo_snapshot_page", - "coordination.local_authority.todo_snapshot_page", + "coordination.local_authority.todo_snapshot", + "coordination.local_authority.todo_snapshot", ] canonical = read_canonical_todos_if_promoted( runtime_root=runtime_root, goal_id="goal-a" diff --git a/tests/control_plane_ts/canonical_snapshot_conformance.ts b/tests/control_plane_ts/canonical_snapshot_conformance.ts index 4ce9f35e2d..5db5738981 100644 --- a/tests/control_plane_ts/canonical_snapshot_conformance.ts +++ b/tests/control_plane_ts/canonical_snapshot_conformance.ts @@ -4,7 +4,7 @@ import test from "node:test"; import type {JsonObject} from "../../loopx/control_plane/effect_program.ts"; import type {AuthorityStore} from "../../loopx/control_plane/coordination/authority_store.ts"; import {coordinationTodoReadModel} from "../../loopx/control_plane/coordination/coordination_projection.ts"; -import {CANONICAL_SNAPSHOT_PAGE_REQUEST, CANONICAL_SNAPSHOT_PAGE_BYTES, +import {CANONICAL_SNAPSHOT_PAGE_REQUEST, CANONICAL_SNAPSHOT_PAGE_BYTES, CANONICAL_SNAPSHOT_LOCAL_BYTES, readCanonicalSnapshotFromStore} from "../../loopx/control_plane/coordination/canonical_snapshot_page.ts"; import {canonicalAuthoritySha256} from "../../loopx/control_plane/coordination/authority_store_codec.ts"; import {normalizeGoalAcceptanceDocument} from "../../loopx/control_plane/goals/acceptance_contract.ts"; @@ -66,11 +66,26 @@ export function registerCanonicalSnapshotConformance(name: string, factory: Auth const byId = (rows: JsonObject[]) => new Map(rows.map(row => [row.todo_id, row])); assert.deepEqual(byId(result.todos), byId(fixture.projection.todos as JsonObject[])); assert.deepEqual(byId(result.leases), byId(fixture.projection.leases as JsonObject[])); + const complete = await readCanonicalSnapshotFromStore(snapshotRequest(), store, CANONICAL_SNAPSHOT_LOCAL_BYTES); + assert.equal(complete.next, null); + assert.deepEqual(complete.snapshot, result.pages[0].snapshot); + assert.deepEqual(complete.metadata, result.pages[0].metadata); + assert.deepEqual(complete.todos, result.todos); + assert.deepEqual(complete.leases, result.leases); + assert.ok(Buffer.byteLength(JSON.stringify(complete)) > CANONICAL_SNAPSHOT_PAGE_BYTES); + // Exercise capacity fallback on this real provider with a smaller private + // response allowance. No caller/protocol constant or wire budget changes. + const bounded = await readCanonicalSnapshotFromStore(snapshotRequest(), store, CANONICAL_SNAPSHOT_PAGE_BYTES); + assert.deepEqual(bounded, result.pages[0]); assert.deepEqual(await store.loadAuthority(), before, "pagination is read-only"); const todoOnly = await collectSnapshot(store, snapshotRequest({include_leases: false})); assert.deepEqual(todoOnly.todos, result.todos); assert.deepEqual(todoOnly.leases, []); assert.ok(todoOnly.pages.every(page => !("leases" in page))); + const completeTodoOnly = await readCanonicalSnapshotFromStore(snapshotRequest({include_leases: false}), + store, CANONICAL_SNAPSHOT_LOCAL_BYTES); + assert.deepEqual(completeTodoOnly.todos, result.todos); + assert.equal("leases" in completeTodoOnly, false); }); } @@ -87,6 +102,8 @@ export function registerCanonicalSnapshotConformance(name: string, factory: Auth operation_id: "concurrent-commit", next_projection: head.head, events: [], receipts: []})).status, "applied"); await assert.rejects(readCanonicalSnapshotFromStore(snapshotRequest({after: first.next}), store), {code: "canonical_snapshot_changed"}); + await assert.rejects(readCanonicalSnapshotFromStore(snapshotRequest({after: first.next}), + store, CANONICAL_SNAPSHOT_LOCAL_BYTES), {code: "canonical_snapshot_changed"}); const restarted = await collectSnapshot(store); assert.notDeepEqual(restarted.pages[0].snapshot, first.snapshot); }); @@ -101,6 +118,8 @@ export function registerCanonicalSnapshotConformance(name: string, factory: Auth identity[key] = key.endsWith("count") ? Number(identity[key]) + 1 : "foreign"; await assert.rejects(readCanonicalSnapshotFromStore(snapshotRequest({after: next}), store), {code: "canonical_snapshot_changed"}, key); + await assert.rejects(readCanonicalSnapshotFromStore(snapshotRequest({after: next}), + store, CANONICAL_SNAPSHOT_LOCAL_BYTES), {code: "canonical_snapshot_changed"}, key); } for (const changed of [{include_leases: false}, {projection_readback: {provider_revision: "other", changed: true, attempt: 1, target: "latest" as const}}]) { @@ -120,6 +139,8 @@ export function registerCanonicalSnapshotConformance(name: string, factory: Auth assert.equal(result.pages[0].next, null); assert.deepEqual(result.todos, []); assert.deepEqual(result.leases, []); + assert.deepEqual(await readCanonicalSnapshotFromStore(snapshotRequest(), store, + CANONICAL_SNAPSHOT_LOCAL_BYTES), result.pages[0]); }); test(`${name}: projection confirmation belongs to the same snapshot as every page`, async t => { @@ -132,6 +153,9 @@ export function registerCanonicalSnapshotConformance(name: string, factory: Auth const result = await collectSnapshot(store, snapshotRequest({projection_readback: {provider_revision: revision, changed, attempt: 1, target: "latest" as const}})); assert.ok(result.pages.every(page => ((page.metadata as JsonObject).projection_readback as JsonObject).status === expected)); + const complete = await readCanonicalSnapshotFromStore(snapshotRequest({projection_readback: + {provider_revision: revision, changed, attempt: 1, target: "latest" as const}}), store, CANONICAL_SNAPSHOT_LOCAL_BYTES); + assert.deepEqual(complete.metadata, result.pages[0].metadata); } }); test(`${name}: enabled acceptance guards stay attached to their own page`, async t => { @@ -161,6 +185,31 @@ export function registerCanonicalSnapshotConformance(name: string, factory: Auth assert.equal(((page.metadata as JsonObject).goal_acceptance_contract as JsonObject).enabled, true); } assert.equal(seen.size, result.todos.length); + const complete = await readCanonicalSnapshotFromStore(snapshotRequest(), store, CANONICAL_SNAPSHOT_LOCAL_BYTES); + const allGuards = Object.assign({}, ...result.pages.map(page => page.goal_acceptance_work_guards)); + assert.deepEqual(complete.goal_acceptance_work_guards, allGuards); + assert.deepEqual(complete.todos, result.todos); + }); + + test(`${name}: a complete read rejects store replacement during admission`, async t => { + const {store} = await factory(t); + await seedSnapshot(store, snapshotFixture().projection); + const identity = store.storeIdentity.bind(store); + let calls = 0; + // Inject replacement at the snapshot admission boundary. Providers may + // check identity internally during load; keep those calls bound to the + // real store rather than changing the provider's own verification path. + const observedStore = new Proxy(store, {get(target, key) { + if (key === "storeIdentity") return async () => { + const observed = await identity(); + return ++calls > 1 && observed.status === "available" + ? {...observed, store_identity: `${observed.store_identity}:replaced`} : observed; + }; + const value = Reflect.get(target, key, target); + return typeof value === "function" ? value.bind(target) : value; + }}); + await assert.rejects(readCanonicalSnapshotFromStore(snapshotRequest(), observedStore, + CANONICAL_SNAPSHOT_LOCAL_BYTES), {code: "canonical_snapshot_changed"}); }); } diff --git a/tests/control_plane_ts/canonical_snapshot_page.test.ts b/tests/control_plane_ts/canonical_snapshot_page.test.ts index 222d95e9fd..e7bcdd9354 100644 --- a/tests/control_plane_ts/canonical_snapshot_page.test.ts +++ b/tests/control_plane_ts/canonical_snapshot_page.test.ts @@ -8,7 +8,8 @@ import {FileAuthorityStore} from "../../loopx/control_plane/coordination/file_au import {SqliteAuthorityStore} from "../../loopx/control_plane/coordination/sqlite_authority_store.ts"; import {sqliteRuntimeIdentity} from "../../loopx/control_plane/coordination/sqlite_runtime.ts"; import {coordinationTodoReadModel} from "../../loopx/control_plane/coordination/coordination_projection.ts"; -import {readCanonicalSnapshotFromStore, readCanonicalSnapshotPage, CANONICAL_SNAPSHOT_PAGE_BYTES} from "../../loopx/control_plane/coordination/canonical_snapshot_page.ts"; +import {readCanonicalSnapshotFromStore, readCanonicalSnapshotPage, readCanonicalSnapshot, + CANONICAL_SNAPSHOT_PAGE_BYTES, CANONICAL_SNAPSHOT_LOCAL_BYTES} from "../../loopx/control_plane/coordination/canonical_snapshot_page.ts"; import {productionScaleCoordinationFixture} from "./production_scale_coordination_fixture.ts"; import {collectSnapshot, registerCanonicalSnapshotConformance, seedSnapshot, snapshotRequest, snapshotFixture} from "./canonical_snapshot_conformance.ts"; @@ -52,6 +53,9 @@ test("oversized single records fail explicitly instead of truncating or looping" await seedSnapshot(store, projection); const before = await store.loadAuthority(); await assert.rejects(collectSnapshot(store), {code: "canonical_snapshot_record_too_large"}); + const complete = await readCanonicalSnapshotFromStore(snapshotRequest(), store, CANONICAL_SNAPSHOT_LOCAL_BYTES); + assert.equal(complete.next, null); + assert.deepEqual(complete.todos, todos, "the private transport keeps the whole oversized inline record"); assert.deepEqual(await store.loadAuthority(), before); }); @@ -76,6 +80,8 @@ test("a missing runtime provider read does not initialize a replacement authorit assert.equal(result.reason_code, undefined); assert.deepEqual(await readdir(root), before); assert.equal("todos" in result, false); + assert.deepEqual(await readCanonicalSnapshot(snapshotRequest({runtime_root: root})), result); + assert.deepEqual(await readdir(root), before); }); test("ordering uses Unicode code points rather than JavaScript default UTF-16 sorting", async t => { @@ -125,6 +131,8 @@ for (const kind of (qualifiedSqlite ? ["file", "sqlite"] : ["file"]) as ("file" const ownershipReason = variant === "duplicate_todo" ? /duplicate todo ids/ : variant === "bad_digest" ? /digest mismatch/ : /lease references an unknown todo/; await assert.rejects(readCanonicalSnapshotFromStore(snapshotRequest(), store), collectionReason); + await assert.rejects(readCanonicalSnapshotFromStore(snapshotRequest(), store, + CANONICAL_SNAPSHOT_LOCAL_BYTES), collectionReason); await assert.rejects(readCoordinationOwnership(store, "goal-a", "2026-09-29T00:00:00Z"), ownershipReason); // Todo-only callers never acquired a lease-graph dependency. Keep that // acceptance boundary even while complete reads reuse the Todo index.