refactor(authority): unify retained journals and enforce scan snapshots

docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md

@@ -758,6 +758,31 @@ separate schemas and roles, and relate them only through opaque identities or
 digests. Provider-specific payloads do not enter the provider-neutral LoopX
 schema.
 
+#### Retained-journal scan contract
+
+`scanCommitted(after_cursor, limit)` binds the head, rows and lookahead to one
+read snapshot. The current providers retain a contiguous journal from cursor 1;
+`null` is the sole start checkpoint, and every supplied cursor is a positive
+canonical decimal string. A positive checkpoint beyond the head, including an
+empty store, fails with `scan_cursor_out_of_range` rather than acknowledging
+successful exhaustion. Malformed runtime values fail before storage access.
+
+The shared TS scan owner checks the exact requested interval, including the
+lookahead row that proves `has_more`. Missing, repeated or reordered rows and a
+last transaction inconsistent with the snapshot head fail as protocol violations.
+PostgreSQL metadata/head/row reads use repeatable read; File/NoKV validate one
+retained envelope and SQLite keeps its existing read transaction. This does not
+introduce a snapshot token across pages: later calls may observe later commits.
+A page does not certify rows before its checkpoint, arbitrary payload integrity,
+or a future compacted/segmented history format.
+
+File and NoKV share journal decoding and append construction in the existing
+transaction module, while retaining their own revision digest inputs, identities,
+CAS and durability effects. Validation covers all four adapters with the shared
+complex fixture, real PostgreSQL concurrent commits and disposable corrupted
+rows, plus isolated real-source File/PostgreSQL pagination. No active Goal
+migration, default-provider change or D1–D3 qualification is implied.
+
 #### 6.2.2 Target store contract after the reference CAS slice
 
 The current Stage 2/3 reference implementation deliberately uses the smaller

docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md

@@ -638,6 +638,26 @@ commit-marker protocol，并通过新的合同 review。
 LoopX 控制面记录与应用领域记录，只通过 opaque identity 或 digest 建立关联。
 provider-specific payload 不进入 provider-neutral 的 LoopX schema。
 
+#### 保留 journal 的扫描合同
+
+`scanCommitted(after_cursor, limit)` 将 head、记录和 lookahead 绑定到同一个读取
+snapshot。当前 provider 保留从 cursor 1 起的连续 journal；`null` 是唯一起点，
+其他游标必须是规范的正十进制字符串。超过 head 的正数 checkpoint（包括空存储）
+返回 `scan_cursor_out_of_range`，不能确认“已成功读完”；非法运行时类型在访问
+存储前拒绝。
+
+共享 TS scan owner 验证请求区间及用于证明 `has_more` 的 lookahead 行。
+缺行、重复、乱序及末条 transaction 与 snapshot head 不一致均为协议错误。
+PostgreSQL metadata/head/row 使用 repeatable read；File/NoKV 验证同一个保留
+历史的 envelope，SQLite 保留原读事务。此合同不增加跨页 snapshot token，后续
+调用可以看到后续提交；也不证明 checkpoint 之前全部历史、任意 payload 的完整性，
+或未来压缩/分段历史格式。
+
+File 与 NoKV 在既有 transaction 模块共用 journal 解码及 append 构造，各自
+保留版本摘要输入、identity、CAS 与持久化副作用。验证包含四个 adapter 的复杂
+fixture、真实 PostgreSQL 并发提交及一次性损坏行，以及真实来源隔离副本上的
+File/PostgreSQL 分页。不迁移活跃 Goal，不切换默认 provider，不宣布 D1–D3 合格。
+
 #### 6.2.2 参考 CAS 切片之后的目标 store contract
 
 当前 Stage 2/3 参考实现刻意使用上面的较小 `load` / `compare_and_put` document

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

@@ -771,6 +771,25 @@ import, provider qualification, soak, or D3 cutover requirements.
   authority separate. Unknown observations cannot settle Todo/replan work.
   Explicitly disclose any semantic correction; do not label it full parity.
 
+The retained-journal read boundary now shares one TS owner for scan admission,
+checkpoint range, contiguous page coverage, lookahead and final-head agreement.
+File and NoKV also share retained-history validation and append construction;
+provider revision hashes, physical locks/CAS and backend headers remain local.
+This retires duplicated storage-protocol knowledge without a new RPC, Python
+bridge, capability or provider. The existing coordination internal owner is
+sufficient; built-in File and optional NoKV/SQLite/PostgreSQL implementations
+retain their deployment boundaries.
+
+Intentional corrections: a positive checkpoint against an empty store is
+`scan_cursor_out_of_range`; non-string cursors are `invalid_scan_request`;
+a missing/reordered retained row or contradictory final head cannot produce a
+successful page. PostgreSQL read operations use one repeatable-read snapshot,
+so a concurrent commit appears on the next call instead of mixing newer rows
+with an older head. The scan proves its requested interval, not an audit of
+history before that checkpoint. Successful schemas, File/NoKV persisted bytes,
+request identity and revision algorithms remain compatible. This supports T3/D1
+readers but does not finish Todo writers, retention/compaction or promotion.
+
 **T4 — collect full-writer retirement after durability cutover.**
 
 - Depends on T1–T3 and the shared RFC's [D1–D3](shared-goal-authority-state-provider-v0.md#durability-execution-cards), including owner approval

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

@@ -594,6 +594,19 @@ user role。历史节点不会进入活动工作或 lease lane。无法识别的
 - 区分历史监督、canonical 义务与 settlement 权威；unknown 不能结清 Todo/replan。
   有意语义修正单独披露，不标成全量 parity。
 
+保留 journal 的读取边界现由同一个 TS owner 负责扫描参数、checkpoint 范围、
+分页连续性、lookahead 和末行/head 一致性。File 与 NoKV 同时共用历史校验及
+append 构造，版本哈希、物理锁/CAS 和后端头字段仍归各 provider。这删除了重复
+存储协议知识，没有新增 RPC、Python bridge、capability 或 provider；既有
+coordination 内部 owner 足够，File 内置及 NoKV/SQLite/PostgreSQL 可选部署边界不变。
+
+明确修正：空存储上的正数 checkpoint 返回 `scan_cursor_out_of_range`，非字符串
+游标返回 `invalid_scan_request`；历史缺行、乱序或末行/head 矛盾不能返回成功分页。
+PostgreSQL 读取使用同一个 repeatable-read snapshot，并发提交在下一次调用可见，
+不会将较新的行混入较旧 head。扫描只证明请求区间，不审计 checkpoint 之前的全部
+历史。合法结果 schema、File/NoKV 持久字节、请求身份及版本算法保持兼容。这支持
+T3/D1 reader，未完成全部 Todo writer、retention/compaction 或 promotion。
+
 **T4 — durable cutover 后兑现完整 writer 删除。**
 
 - 前提是 T1–T3 和 shared RFC 的 [D1–D3](shared-goal-authority-state-provider-v0.zh-CN.md#持久化执行卡)，包括 owner 批准及明确的 legacy 迁移窗口。

examples/control_plane/authority-three-arm-rehearsal.py

@@ -135,6 +135,20 @@
     assert.equal(loaded.status, 'loaded', `${name} readback failed`);
     const receipt = await store.readReceipt(`${name}-three-arm-archive`);
     assert.equal(receipt.status, 'found', `${name} receipt missing`);
+    const firstPage = await store.scanCommitted(null, 1);
+    assert.equal(firstPage.status, 'page', `${name} first journal page failed`);
+    assert.equal(firstPage.has_more, true);
+    assert.deepEqual(firstPage.transactions[0].projection, request.initial);
+    const finalPage = await store.scanCommitted(firstPage.next_cursor, 1);
+    assert.equal(finalPage.status, 'page', `${name} final journal page failed`);
+    assert.equal(finalPage.has_more, false);
+    assert.deepEqual(finalPage.transactions[0].projection, loaded.head);
+    assert.equal(finalPage.transactions[0].provider_revision, loaded.provider_revision);
+    assert.deepEqual(finalPage.transactions[0].receipts, receipt.receipts);
+    const end = await store.scanCommitted(finalPage.next_cursor, 1);
+    assert.deepEqual(end, {status: 'page', transactions: [],
+      next_cursor: finalPage.next_cursor, has_more: false});
+    assert.deepEqual(await store.loadAuthority(), loaded, 'journal reads changed authority');
     results[name] = {archived, head: loaded.head};
   }
 
@@ -231,6 +245,7 @@
     active_lease_count_after: activeLeases.length,
     moved_ids_sha256_prefix: movedDigest.slice(0, 16),
     provider_heads_exact: true,
+    journal_pages_exact: true,
     legacy_active_semantics_exact: true,
     relative_order_exact: true,
     non_target_semantics_unchanged: true,

loopx/control_plane/coordination/authority_journal_scan.ts

@@ -0,0 +1,64 @@
+/** A page proves a contiguous segment of one retained journal snapshot.
+ * Storage effects and snapshot acquisition remain with each provider. */
+import type {AuthorityStoreCommittedTransaction, AuthorityStoreHead,
+  AuthorityStoreReadFailure, AuthorityStoreScanResult} from "./authority_store.ts";
+import {AuthorityStoreProtocolError, canonicalAuthorityBytes, parseAuthorityCursor} from "./authority_store_codec.ts";
+
+export class AuthorityJournalScan {
+  readonly after: string | null;
+  readonly offset: bigint;
+  readonly limit: number;
+
+  private constructor(after: string | null, offset: bigint, limit: number) {
+    this.after = after; this.offset = offset; this.limit = limit;
+  }
+
+  static prepare(after: string | null, limit: number): AuthorityJournalScan | AuthorityStoreReadFailure {
+    try {
+      const offset = parseAuthorityCursor(after);
+      if (!Number.isSafeInteger(limit) || limit < 1) {
+        throw new AuthorityStoreProtocolError("scan limit must be a positive safe integer");
+      }
+      return new AuthorityJournalScan(after, offset, limit);
+    } catch (error) {
+      if (!(error instanceof AuthorityStoreProtocolError)) throw error;
+      return {status: "failed", reason_code: "invalid_scan_request", reason: error.message};
+    }
+  }
+
+  rangeFailure(headCursor: string | null): AuthorityStoreReadFailure | null {
+    return this.offset > parseAuthorityCursor(headCursor)
+      ? {status: "failed", reason_code: "scan_cursor_out_of_range",
+        reason: "scan cursor is ahead of the provider head"} : null;
+  }
+
+  /** Providers fetch up to limit + 1 rows inside the same snapshot as head.
+   * The extra row proves has_more and must pass the same validation. */
+  page(rows: readonly AuthorityStoreCommittedTransaction[], head: AuthorityStoreHead | null): AuthorityStoreScanResult {
+    const range = this.rangeFailure(head?.cursor ?? null);
+    if (range) return range;
+    const remaining = parseAuthorityCursor(head?.cursor ?? null) - this.offset;
+    const requested = BigInt(this.limit) + 1n;
+    const expected = remaining < requested ? remaining : requested;
+    if (BigInt(rows.length) !== expected) {
+      throw new AuthorityStoreProtocolError("committed scan does not cover its retained snapshot interval");
+    }
+    const operations = new Set<string>();
+    for (const [index, row] of rows.entries()) {
+      if (parseAuthorityCursor(row.cursor) !== this.offset + BigInt(index) + 1n) {
+        throw new AuthorityStoreProtocolError("committed scan cursor lineage is invalid");
+      }
+      if (operations.has(row.operation_id)) {
+        throw new AuthorityStoreProtocolError("committed scan operation identity is duplicated");
+      }
+      operations.add(row.operation_id);
+      if (head && row.cursor === head.cursor && (row.provider_revision !== head.provider_revision ||
+          !canonicalAuthorityBytes(row.projection).equals(canonicalAuthorityBytes(head.head)))) {
+        throw new AuthorityStoreProtocolError("committed scan head lineage is invalid");
+      }
+    }
+    const transactions = structuredClone(rows.slice(0, this.limit));
+    return {status: "page", transactions,
+      next_cursor: transactions.at(-1)?.cursor ?? this.after, has_more: rows.length > this.limit};
+  }
+}

loopx/control_plane/coordination/authority_store_codec.ts

@@ -110,7 +110,7 @@ export function canonicalAuthoritySha256(value: unknown): string {
 
 export function parseAuthorityCursor(value: string | null): bigint {
   if (value === null) return 0n;
-  if (!/^[1-9]\d*$/.test(value)) {
+  if (typeof value !== "string" || !/^[1-9]\d*$/.test(value)) {
     throw new AuthorityStoreProtocolError("provider cursor is invalid");
   }
   return BigInt(value);

loopx/control_plane/coordination/authority_store_transactions.ts

@@ -1,21 +1,25 @@
 import type { JsonObject } from "../effect_program.ts";
-import type { AuthorityStoreCommittedTransaction } from "./authority_store.ts";
+import type { AuthorityStoreCommit, AuthorityStoreCommittedTransaction } from "./authority_store.ts";
 import {
   AuthorityStoreProtocolError,
   canonicalAuthorityObject,
   canonicalAuthorityObjectList,
   hasExactAuthorityKeys,
   isAuthorityJsonObject,
   requireAuthorityStoreId,
+  parseAuthorityCursor,
+  canonicalAuthorityBytes,
 } from "./authority_store_codec.ts";
 
 /** Shared wire decoder used by every authority provider. */
 export function decodeAuthorityTransaction(value: unknown): AuthorityStoreCommittedTransaction {
   if (!isAuthorityJsonObject(value) || !hasExactAuthorityKeys(value, [
     "cursor", "provider_revision", "operation_id", "events", "projection", "receipts",
   ])) throw new AuthorityStoreProtocolError("committed transaction is invalid");
+  const cursor = requireAuthorityStoreId(value.cursor, "transaction cursor");
+  parseAuthorityCursor(cursor);
   return {
-    cursor: requireAuthorityStoreId(value.cursor, "transaction cursor"),
+    cursor,
     provider_revision: requireAuthorityStoreId(value.provider_revision, "transaction provider revision"),
     operation_id: requireAuthorityStoreId(value.operation_id, "operation id"),
     events: canonicalAuthorityObjectList(value.events, "transaction events"),
@@ -24,10 +28,6 @@ export function decodeAuthorityTransaction(value: unknown): AuthorityStoreCommit
   };
 }
 
-export function cloneAuthorityTransaction(value: AuthorityStoreCommittedTransaction): AuthorityStoreCommittedTransaction {
-  return structuredClone(value);
-}
-
 export function transactionForRevision(value: AuthorityStoreCommittedTransaction): JsonObject {
   return {
     cursor: value.cursor,
@@ -37,3 +37,62 @@ export function transactionForRevision(value: AuthorityStoreCommittedTransaction
     receipts: value.receipts,
   };
 }
+
+/** File and NoKV retain the whole journal in one envelope. Their wire headers,
+ * physical CAS, and revision algorithms remain provider-owned. */
+export interface RetainedAuthorityJournal {
+  provider_revision: string;
+  cursor: string;
+  head: JsonObject;
+  committed: AuthorityStoreCommittedTransaction[];
+}
+
+type RevisionInput = ReturnType<typeof transactionForRevision>;
+export type JournalRevision = (previous: string | null, transaction: RevisionInput) => string;
+
+export function decodeRetainedAuthorityJournal(value: JsonObject, label: string,
+  revisionFor: JournalRevision): RetainedAuthorityJournal {
+  const revision = requireAuthorityStoreId(value.provider_revision, "provider revision");
+  const cursor = requireAuthorityStoreId(value.cursor, "provider cursor");
+  const head = canonicalAuthorityObject(value.head, `${label} head`);
+  if (!Array.isArray(value.committed)) {
+    throw new AuthorityStoreProtocolError(`${label} history is invalid`);
+  }
+  const committed = value.committed.map(decodeAuthorityTransaction);
+  if (committed.length === 0 || parseAuthorityCursor(cursor) !== BigInt(committed.length)) {
+    throw new AuthorityStoreProtocolError(`${label} lineage is invalid`);
+  }
+  let previous: string | null = null;
+  const operations = new Set<string>();
+  for (const [index, entry] of committed.entries()) {
+    if (parseAuthorityCursor(entry.cursor) !== BigInt(index + 1)) {
+      throw new AuthorityStoreProtocolError(`${label} cursor lineage is invalid`);
+    }
+    if (operations.has(entry.operation_id)) {
+      throw new AuthorityStoreProtocolError(`${label} operation identity is duplicated`);
+    }
+    operations.add(entry.operation_id);
+    if (entry.provider_revision !== revisionFor(previous, transactionForRevision(entry))) {
+      throw new AuthorityStoreProtocolError(`${label} revision lineage is invalid`);
+    }
+    previous = entry.provider_revision;
+  }
+  const last = committed.at(-1)!;
+  if (last.cursor !== cursor || last.provider_revision !== revision ||
+      !canonicalAuthorityBytes(last.projection).equals(canonicalAuthorityBytes(head))) {
+    throw new AuthorityStoreProtocolError(`${label} head lineage is invalid`);
+  }
+  return {provider_revision: revision, cursor, head, committed};
+}
+
+/** Build only after the provider checked revision and operation uniqueness in
+ * its write boundary. This function neither grants admission nor persists. */
+export function appendRetainedAuthorityJournal(current: RetainedAuthorityJournal | null,
+  commit: AuthorityStoreCommit, revisionFor: JournalRevision): RetainedAuthorityJournal {
+  const cursor = (parseAuthorityCursor(current?.cursor ?? null) + 1n).toString();
+  const base = {cursor, operation_id: commit.operation_id, events: commit.events,
+    projection: commit.next_projection, receipts: commit.receipts};
+  const revision = revisionFor(current?.provider_revision ?? null, base);
+  return {provider_revision: revision, cursor, head: commit.next_projection,
+    committed: [...(current?.committed ?? []), {...base, provider_revision: revision}]};
+}