fix(coordination): distinguish invalid snapshots from invalid requests

docs/architecture/rfcs/monorepo-distribution-split-v0.md

@@ -3391,3 +3391,17 @@ original task and I3/minor-release, M1/#8011, R0/V10/R1, D2/D5/D6/D3,
 context/model/split-install, PostgreSQL host/wheel, installed-host and adoption
 gaps remain open. The next qualification stays with those original owners and
 acceptance boundaries; the minimal-kernel effort is not complete.
+
+### 2026-11-11 — Canonical snapshot failure provenance
+
+Paged and complete snapshot reads distinguish a rejected persisted Todo contract
+(`canonical_snapshot_projection_invalid`) from malformed request arguments. The
+existing typed validator still rejects unknown manifests and invalid record
+digests, preserves the original reason and returns no partial rows or legacy
+fallback. Historical supported manifests remain readable. This is a diagnostic
+correction in the existing source transport owner; it grants no forward
+compatibility to older readers and does not upgrade an installed runtime.
+
+Real File/SQLite RPC and CLI regressions, plus an isolated PostgreSQL read,
+qualify the failure boundary without rewriting authority. The original
+installed-host qualification and full-cost/adoption gaps remain open.

docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md

@@ -2647,3 +2647,14 @@ long-task window。五个中位数为 336.34/313.05/441.09/200.01/194.94 ms，
 D2/D5/D6/D3、context/model/split-install、PostgreSQL host/wheel、
 installed-host 和 adoption 缺口继续开放。后续资格仍由这些原 owner 和验收
 边界承接；最小化内核工作尚未全部完成。
+
+### 2026-11-11 — Canonical snapshot 失败来源
+
+分页和完整 snapshot 读取将持久化 Todo 契约被拒绝
+（`canonical_snapshot_projection_invalid`）与请求参数无效区分开。现有 typed
+validator 仍拒绝未知字段清单和无效记录摘要，保留原始原因，不返回部分记录或
+legacy fallback；已支持的历史字段清单仍可读取。这是既有源传输入口的诊断
+修正，不授予旧 reader 前向兼容能力，也不升级已安装 runtime。
+
+真实 File/SQLite RPC 与 CLI 回归，以及隔离 PostgreSQL 读取，验证该失败边界
+且不重写 authority。原 installed-host 资格、完整成本和 adoption 缺口继续开放。

loopx/control_plane/coordination/canonical_snapshot_page.ts

@@ -6,7 +6,7 @@ import {requireJsonObject} from "../runtime_decode.ts";
 import {MAX_LOCAL_SNAPSHOT_BYTES} from "../effect_runtime_snapshot.ts";
 import {decodeProjectionReadback, confirmProjectionReadback} from "../todos/projection_delivery.ts";
 import {authorityStoreSourceAuthority, readAuthoritySnapshot, type AuthorityStore} from "./authority_store.ts";
-import {canonicalAuthoritySha256, hasExactAuthorityKeys, requireAuthorityStoreId} from "./authority_store_codec.ts";
+import {AuthorityStoreProtocolError, canonicalAuthoritySha256, hasExactAuthorityKeys, requireAuthorityStoreId} from "./authority_store_codec.ts";
 import {canonicalTodoCollection} from "./local_authority_read.ts";
 import {canonicalAuthorityRouteFailure, requireCanonicalAuthority} from "./local_authority_write.ts";
 import {openRuntimeAuthorityStore, requireLocalAuthorityRuntimeRoot, localAuthorityOpenFailure,
@@ -93,8 +93,16 @@ export async function readCanonicalSnapshotFromStore(value: unknown, store: Auth
     after.snapshot.provider_revision === head.provider_revision && after.snapshot.cursor === head.cursor &&
     after.snapshot.query_sha256 === query,
   "canonical_snapshot_changed", "canonical snapshot changed; restart the complete read, never append a newer page");
-  const {projection, todoReadModel: readModel, leaseIndex: leases, acceptance, oracleObservationBasis, projectWorkGuards} =
-    canonicalTodoCollection(head.head, goal, input.include_leases);
+  let collection: ReturnType<typeof canonicalTodoCollection>;
+  try {
+    collection = canonicalTodoCollection(head.head, goal, input.include_leases);
+  } catch (error) {
+    // The request is already decoded. A rejected persisted contract belongs to
+    // the reader/authority boundary, not to the caller's request arguments.
+    if (!(error instanceof AuthorityStoreProtocolError)) throw error;
+    throw new SnapshotReadError("canonical_snapshot_projection_invalid", error.message);
+  }
+  const {projection, todoReadModel: readModel, leaseIndex: leases, acceptance, oracleObservationBasis, projectWorkGuards} = collection;
   const snapshot: Snapshot = {goal_id: goal, store_identity: identity.store_identity,
     provider_revision: head.provider_revision, cursor: head.cursor, query_sha256: query,
     todo_count: projection.todo_ids.length, lease_count: leases?.lease_todo_ids.length ?? 0};

tests/control_plane/test_canonical_snapshot_integration.py

@@ -7,6 +7,7 @@
 
 import pytest
 from canonical_authority_fixture import (
+    advance_canonical_fixture_authority,
     initialize_canonical_authority,
     isolate_sqlite_runtime,
 )
@@ -87,6 +88,29 @@ def native_read(runtime):
     )
 
 
+def test_cli_preserves_projection_failure_and_original_reason(wide_goal):
+    runtime, _state, registry, projection = wide_goal
+    projection["todo_read_model"]["contract_fields"].append("future_field")
+    advance_canonical_fixture_authority(
+        runtime, "goal-a", projection, operation_id="unsupported-manifest"
+    )
+    with pytest.raises(LocalCoordinationAuthorityUnavailable) as failure:
+        native_read(runtime)
+    assert failure.value.code == "canonical_snapshot_projection_invalid"
+    assert "field contract mismatch" in str(failure.value)
+    process = subprocess.run(
+        [sys.executable, "-m", "loopx.entrypoint", "--registry", str(registry),
+         "--format", "json", "todo", "list", "--goal-id", "goal-a"],
+        cwd=REPO, text=True, capture_output=True, timeout=60, check=False,
+    )
+    assert process.returncode != 0
+    result = json.loads(process.stdout)
+    assert result["ok"] is False
+    assert result["error_code"] == "canonical_snapshot_projection_invalid"
+    assert "field contract mismatch" in result["error"]
+    assert "todos" not in result
+
+
 def test_historical_source_uses_private_snapshot_above_inline_budget(wide_goal):
     runtime, _state, _registry, projection = wide_goal
     original = native_read(runtime)

tests/control_plane_ts/canonical_snapshot_page.test.ts

@@ -30,6 +30,38 @@ async function fixture(t: test.TestContext, kind: "file" | "sqlite" = "file") {
   return {store: new Store(root, "goal-a"), contender: new Store(root, "goal-a")};
 }
 const qualifiedSqlite = sqliteRuntimeIdentity().sqlite_authority_qualified;
+for (const provider of (qualifiedSqlite ? ["file", "sqlite"] : ["file"]) as ("file" | "sqlite")[]) {
+  for (const defect of ["manifest", "digest"] as const) {
+    test(`${provider}: invalid projection is distinct from an invalid snapshot request (${defect})`, async t => {
+      const root = await mkdtemp(join(tmpdir(), "snapshot-invalid-projection-"));
+      t.after(() => rm(root, {recursive: true, force: true}));
+      await engageCanonicalRoute(root);
+      const Store = provider === "file" ? FileAuthorityStore : SqliteAuthorityStore;
+      const store = new Store(join(root, "store"), "goal-a");
+      const projection = productionScaleCoordinationFixture("goal-a", "native").projection;
+      const model = projection.todo_read_model as JsonObject;
+      if (defect === "manifest") (model.contract_fields as string[]).push("future_field");
+      else model.records_sha256 = "0".repeat(64);
+      await seedSnapshot(store, projection);
+      const before = await store.loadAuthority();
+      const dependencies = {createStore: () => store};
+      for (const read of [readCanonicalSnapshotPage, readCanonicalSnapshot]) {
+        const request = snapshotRequest({runtime_root: root});
+        const invalidRequest = await read({...request, include_leases: "true"}, dependencies);
+        assert.equal(invalidRequest.reason_code, "canonical_snapshot_request_invalid");
+        const result = await read(request, dependencies);
+        assert.equal(result.status, "failed");
+        assert.equal(result.reason_code, "canonical_snapshot_projection_invalid");
+        assert.match(String(result.reason), defect === "manifest" ? /field contract mismatch/ : /digest mismatch/);
+        assert.equal(result.decision_read_from_provider, true);
+        assert.equal(result.legacy_fallback_used, false);
+        assert.equal("todos" in result, false);
+        assert.equal("next" in result, false);
+      }
+      assert.deepEqual(await store.loadAuthority(), before, "rejected reads must not rewrite authority");
+    });
+  }
+}
 if (qualifiedSqlite) {
   test("selected SQLite snapshot retains the complete independent read contract", async t => {
     const {store} = await fixture(t, "sqlite");

tests/control_plane_ts/postgresql_authority_store.integration.test.ts

@@ -19,7 +19,7 @@ import {
 } from "../../loopx/control_plane/coordination/postgresql_authority_store.ts";
 import { openLocalAuthorityStoreHandle } from "../../loopx/control_plane/coordination/local_authority_provider.ts";
 import {readLocalCoordinationTodoSource} from "../../loopx/control_plane/coordination/local_authority_read.ts";
-import {CANONICAL_SNAPSHOT_PAGE_REQUEST, readCanonicalSnapshotFromStore} from "../../loopx/control_plane/coordination/canonical_snapshot_page.ts";
+import {CANONICAL_SNAPSHOT_PAGE_REQUEST, readCanonicalSnapshotFromStore, readCanonicalSnapshotPage} from "../../loopx/control_plane/coordination/canonical_snapshot_page.ts";
 import {projectTodoContextPage} from "../../loopx/control_plane/todos/context_projection.ts";
 import {readCanonicalTodoProjection, readCanonicalQuotaProjection} from "../../loopx/control_plane/todos/read_transaction.ts";
 import {canonicalReadRoot} from "./local_promotion_fixture.ts";
@@ -270,6 +270,28 @@ if (database && installed) {
     "todo_source_lineage_mismatch");
   });
 
+  test("PostgreSQL snapshot reports a rejected field contract without changing authority", async t => {
+    await installed;
+    const options = {tenant_id: `tenant-${randomUUID()}`, goal_id: `goal-${randomUUID()}`};
+    t.after(() => cleanScope(options.tenant_id, options.goal_id));
+    const store = new PostgreSqlAuthorityStore(database, options);
+    const projection = authorityProjectionFixture(options.goal_id, []);
+    ((projection.todo_read_model as JsonObject).contract_fields as string[]).push("future_field");
+    const committed = await store.commitAuthority({expected_provider_revision: null,
+      operation_id: "seed-unsupported-contract", events: [], receipts: [], next_projection: projection});
+    assert.equal(committed.status, "applied");
+    const before = await store.loadAuthority();
+    const result = await readCanonicalSnapshotPage({schema_version: CANONICAL_SNAPSHOT_PAGE_REQUEST,
+      runtime_root: await canonicalReadRoot(t, options.goal_id), goal_id: options.goal_id,
+      include_leases: true, projection_readback: null, after: null}, {createStore: () => store});
+    assert.equal(result.status, "failed");
+    assert.equal(result.reason_code, "canonical_snapshot_projection_invalid");
+    assert.match(String(result.reason), /field contract mismatch/);
+    assert.equal(result.legacy_fallback_used, false);
+    assert.equal("todos" in result, false);
+    assert.deepEqual(await store.loadAuthority(), before);
+  });
+
   test("PostgreSQL canonical snapshot feeds complete display views without becoming a fresh authority read", async t => {
     await installed;
     const options = {tenant_id: `tenant-${randomUUID()}`, goal_id: `goal-${randomUUID()}`};