feat(sync): pull a stream's journal - verify, stage once, apply (T12343 S5-1)

.changeset/t12343-pull.md

@@ -0,0 +1,34 @@
+---
+id: t12343-pull
+tasks: [T12343, T13306, T13307]
+kind: feat
+summary: Sync pull - stage and apply a stream's verified segments, never a transaction twice (S5-1)
+---
+
+A store can now pull its stream's journal (journal spec §3.1, §2.8; T12343 S5-1).
+
+- **`pullStream(db, …)`** (`store/sync/pull.ts`) refuses while `sync.pull` is off. Otherwise it pages through the stream after the
+  store's persisted cursor. For each page, in one
+  local transaction:
+  - **Decode:** `deflate-raw` canonical JSON of `LedgerTxn[]`, parsed with the ledger schema. A vault count-delta segment
+    (`cleo-vault-delta/v1`) carries no ops and is passed over.
+  - **Verify:** every transaction's signature is checked against the segment's device key, before anything of the page is written.
+    A bad signature or a malformed body refuses the segment. The pull stops there: what precedes it is staged and applied, the
+    cursor stops just before it, and the report's `refused` names its seq, replica and device (T13307).
+  - **Deduplicate:** every transaction id already staged from the stream is skipped (`_sync_seen_txn`), so a re-delivered
+    transaction is never staged or applied twice and a counter delta never doubles.
+  - **Stage and advance:** the rest are staged, and the cursor advances (`_sync_cursor`).
+
+  Then the applier applies, in stream order, own echoes included (they feed the fast path and the scoped rebase).
+- **`_sync_seen_txn` is not pruned yet.** A transaction already seen can come back in a new segment above any checkpoint, for
+  example a retired replica's late segment or a rebind's re-push. If its first-delivery row had been dropped, it would apply
+  twice. `pruneSeenTxns` exists, but the cloud pull calls it only once the receive watermark (T13256) provides a floor the
+  applier refuses to go below.
+- **`pullSyncStream(opts)`** (`cloud/nexus-vault.ts`) wires `pullStream` to the journal client and the trusted device keys. It refuses
+  a store bound to another replica than the link. With `sync.pull` off it refuses before any network call. A store whose last synced checkpoint is a vault snapshot from
+  before the stream's journal genesis is refused, with nothing pulled (T13306). The changes folded into that genesis never travel
+  as segments, so pulling across it would silently diverge. The remedy is to restore the journal checkpoint, then join with
+  `cleo sync enable push`. A store with no pull position starts after the checkpoint it last synced (its
+  genesis, or the snapshot it restored), seeded from that checkpoint's signed replica map.
+- **New sync-journal migration** `t12343-pull-cursor`, local-only: `_sync_cursor` and `_sync_seen_txn`. Both are cleared from vault
+  bundles, so a restored store resumes from its own checkpoint.

packages/contracts/src/nexus-account.ts

@@ -104,7 +104,10 @@ export const NEXUS_ACCOUNT_ERROR_CODES = [
   // Change-journal push (`cleo sync enable push`, T12343 S4-1b).
   /** The store is not fit to reach genesis (a precondition failed); nothing was cut. */
   'E_NEXUS_SYNC_REFUSED',
-  /** Another device already started this stream's change journal: this store joins it by pulling. */
+  /**
+   * Another device already started this stream's change journal: this store restores its journal
+   * checkpoint, then joins the stream (T12999), never writing a second genesis.
+   */
   'E_NEXUS_SYNC_STREAM_JOURNALED',
   // Projects by name (`cleo cloud restore <name>`, T13102).
   /** No project of the account has that name, label or id. */

packages/core/migrations/sync-journal/20261008120000_t12343-pull-cursor/migration.sql

@@ -0,0 +1,28 @@
+-- T12343 S5-1 (journal spec §3.1 "Receive") — where a store's pull stands.
+--
+-- _sync_cursor: per stream, the PullCursor the journal client verifies the
+-- next page against (the last applied stream seq, and per replica its pinned
+-- device and last replicaSeq). It is advanced in the SAME local transaction
+-- that stages the segments it covers (§3.1 step 3), so a crash re-pulls
+-- nothing twice and loses nothing.
+--
+-- _sync_seen_txn: every transaction id this store has staged from a stream.
+-- A transaction re-delivered under another segment (a replayed or re-sealed
+-- segment, a retired replica's late segment carrying a txn already seen) is
+-- skipped, never staged twice: a counter delta ($inc) applied twice would
+-- double (T12343 S5-1).
+--
+-- Local-only.
+--
+-- Do not edit this file once applied: its hash is recorded per store.
+CREATE TABLE IF NOT EXISTS _sync_cursor (
+  stream      TEXT PRIMARY KEY,
+  cursor_json TEXT NOT NULL,
+  updated_at  TEXT NOT NULL
+) WITHOUT ROWID;
+CREATE TABLE IF NOT EXISTS _sync_seen_txn (
+  stream TEXT NOT NULL,
+  txn    TEXT NOT NULL,
+  seq    INTEGER NOT NULL,
+  PRIMARY KEY (stream, txn)
+) WITHOUT ROWID;

packages/core/src/cloud/__tests__/nexus-attach.test.ts

@@ -307,6 +307,7 @@ describe('presence and binding hold their guarantees (review M3)', () => {
         '_sync_capture',
         '_sync_clock',
         '_sync_conflict',
+        '_sync_cursor',
         '_sync_field_leave',
         '_sync_foreign_touch',
         '_sync_frame',
@@ -318,6 +319,7 @@ describe('presence and binding hold their guarantees (review M3)', () => {
         '_sync_replica',
         '_sync_row_meta',
         '_sync_row_undo',
+        '_sync_seen_txn',
         '_sync_segment',
         '_sync_segment_txn',
         '_sync_sequenced',

packages/core/src/cloud/__tests__/nexus-vault.test.ts

@@ -123,6 +123,7 @@ import { listNexusNamedProjects, resolveNexusProjectRef } from '../nexus-project
 import {
   enableSyncPush,
   nexusVaultStatus,
+  pullSyncStream,
   pushNexusVault,
   pushSyncStream,
   releaseNexusVaultLease,
@@ -5507,6 +5508,69 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => {
     });
   });
 
+  it('S5-1: pull starts after the genesis checkpoint, verifies and stages the own echo, and sequences it', async () => {
+    const { m, dbPath } = await journalMachine();
+    await on(m, () => enableSyncPush(vopts(m, { allowUnreleased: true })));
+    await on(m, async () => {
+      (await storeOf(dbPath)).exec(
+        "INSERT INTO tasks_tasks (id, title, type, status, priority, uid, birth_fp) VALUES ('T8', 'title T8', 'task', 'pending', 'medium', 'uid-T8', 'fp-T8')",
+      );
+    });
+    const pushed = await on(m, () => pushSyncStream(vopts(m, { allowUnreleased: true })));
+    expect(pushed.pushed).toBe(1);
+    await on(m, async () => {
+      setSyncFlag(await storeOf(dbPath), 'sync.pull', true, { allowUnreleased: true });
+    });
+    const pulled = await on(m, () => pullSyncStream(vopts(m)));
+    expect(pulled).toMatchObject({ segments: 1, staged: 1, redelivered: 0, vaultDeltas: 0 });
+    expect(pulled.after).toBe(pushed.lastServerSeq);
+    expect(pulled.apply?.applied).toBe(1);
+    await on(m, async () => {
+      const db = await storeOf(dbPath);
+      // The own echo was sequenced (§3.5): its transaction is no longer unsequenced.
+      expect(
+        (db.prepare('SELECT count(*) AS n FROM _sync_sequenced').get() as { n: number }).n,
+      ).toBe(1);
+    });
+    const again = await on(m, () => pullSyncStream(vopts(m)));
+    expect(again).toMatchObject({ segments: 0, staged: 0 });
+  });
+
+  it('S5-1: a store that synced a vault snapshot from before the journal genesis refuses to pull across it (T13306)', async () => {
+    const { m: a, dbPath } = await journalMachine();
+    await on(a, () => pushNexusVault(vopts(a))); // a vault snapshot, before any journal
+    // B restores that vault snapshot and is attached under its own bound replica.
+    const b = await machine('b', DEVICE_B, REPLICA_B);
+    fake.replicas.get(REMOTE_PROJECT)?.set(REPLICA_B, DEVICE_B);
+    await restoreOntoB(b);
+    const bDb = path.join(b.root, '.cleo', 'cleo.db');
+    const bound = await on(b, async () => {
+      const db = await storeOf(bDb);
+      setSyncFlag(db, 'sync.pull', true, { allowUnreleased: true });
+      return ensureProjectReplica(db, { dbPath: bDb, mode: 'live' }).replicaId;
+    });
+    b.replicaId = bound;
+    fake.replicas.get(REMOTE_PROJECT)?.set(bound, DEVICE_B);
+    link(b);
+    // A changes the store, then starts the journal: that change is folded into the genesis, never a segment.
+    await on(a, async () => {
+      (await storeOf(dbPath)).exec(
+        "INSERT INTO tasks_tasks (id, title, type, status, priority, uid, birth_fp) VALUES ('G1', 'title G1', 'task', 'pending', 'medium', 'uid-G1', 'fp-G1')",
+      );
+    });
+    await on(a, () => enableSyncPush(vopts(a, { allowUnreleased: true })));
+    const refused = await failure(on(b, () => pullSyncStream(vopts(b))));
+    expect(refused.code).toBe('E_NEXUS_SYNC_REFUSED');
+    expect(refused.message).toMatch(/started its change journal .* would silently diverge/);
+    expect(refused.fix).toMatch(/restore the journal checkpoint first .*cleo sync enable push/);
+    await on(b, async () => {
+      const db = await storeOf(bDb);
+      expect((db.prepare('SELECT count(*) AS n FROM _sync_inbox').get() as { n: number }).n).toBe(
+        0,
+      );
+    });
+  });
+
   it('S4-2: a device clock ahead of the server pauses push', async () => {
     const { m, dbPath } = await journalMachine();
     await on(m, () => enableSyncPush(vopts(m, { allowUnreleased: true })));

packages/core/src/cloud/nexus-vault.ts

@@ -105,7 +105,7 @@ import {
 } from '../store/portable-bundle-scan.js';
 import { writeRestoreMarker } from '../store/restore-marker.js';
 import { FIRST_OPEN_LOCK_SUFFIX } from '../store/sqlite.js';
-import { UNRELEASED_FLAGS } from '../store/sync/flags.js';
+import { isSyncFlagOn, UNRELEASED_FLAGS } from '../store/sync/flags.js';
 import {
   completeGenesis,
   cutGenesisWithSnapshot,
@@ -114,6 +114,7 @@ import {
   genesisPending,
   readGenesisCut,
 } from '../store/sync/genesis.js';
+import { type PullStreamReport, pullStream, readStreamCursor } from '../store/sync/pull.js';
 import { type PushStreamReport, pushStream } from '../store/sync/push.js';
 import { replayPinOf } from '../store/sync/replay-pin.js';
 import {
@@ -123,7 +124,8 @@ import {
   rebindAfterVaultRestore,
 } from '../store/sync/replica.js';
 import { ReplicaRegistry } from '../store/sync/replica-registry.js';
-import { signTxn } from '../store/sync/txn-signing.js';
+import { sealPending } from '../store/sync/sealer.js';
+import { firstBadTxnSignature, signTxn } from '../store/sync/txn-signing.js';
 import {
   buildVaultManifest,
   type CarriedMachineState,
@@ -152,7 +154,7 @@ import { foreignWriterLeases, storeOpenElsewhere } from '../store/writer-lease.j
 import { deriveKey, uuidv7 } from './crypto.js';
 import { NexusError } from './http.js';
 import { cursorFromCheckpoint, initialPullCursor, Journal, type PullCursor } from './journal.js';
-import type { TrustedSigners } from './keys.js';
+import { signerKeys, type TrustedSigners } from './keys.js';
 import { manifestVersion } from './manifest-check.js';
 import { canonicalGlobalReplicaBinder } from './nexus-attach.js';
 import { NexusAccountError } from './nexus-auth.js';
@@ -805,6 +807,10 @@ const VAULT_CLEARED_JOURNAL_TABLES = [
   '_sync_frame',
   '_sync_quarantine',
   '_sync_apply_intent',
+  // Where this device's pull stands (T12343 S5-1): a restored store resumes
+  // from the checkpoint it restored, never from another device's position.
+  '_sync_cursor',
+  '_sync_seen_txn',
 ] as const;
 
 /**
@@ -1569,7 +1575,7 @@ async function enableSyncPushImpl(
     throw vaultError(
       'E_NEXUS_SYNC_STREAM_JOURNALED',
       `${t.streamId} already carries the change journal: its head snapshot ${parent?.checkpointId} is a journal checkpoint`,
-      'nothing was cut; this store joins the journal by pulling it (`cleo cloud sync`, once pull ships)',
+      'nothing was cut; restore the journal checkpoint (`cleo cloud restore`), then join the stream with `cleo sync enable push` (T12999)',
     );
   }
   const { state: synced } = syncedState(conn, t, parent ? [parent] : [], head.headCheckpointId);
@@ -1821,6 +1827,128 @@ async function pushSyncStreamImpl(
   }
 }
 
+async function pullSyncStreamImpl(opts: NexusVaultCommandOptions = {}): Promise<PullStreamReport> {
+  const conn = await connectNexusVault(opts);
+  const key = await unlockNexusAccountKey(conn);
+  const t = await resolveTarget(conn, key.masterKey, opts, 'push');
+  const replicaId = t.replicaId;
+  if (!replicaId) {
+    throw vaultError(
+      'E_NEXUS_VAULT_NOT_LINKED',
+      'this copy of the project is not attached from this device',
+      'run `cleo project link`',
+    );
+  }
+  const journal = journalFor(conn, t);
+  const { openDualScopeDbAtPath, getDualScopeNativeDb } = await import('../store/dual-scope-db.js');
+  const db = getDualScopeNativeDb(
+    t.scope === 'global'
+      ? await openDualScopeDbAtPath('global', t.dbPath)
+      : await openDualScopeDbAtPath('project', t.dbPath),
+  );
+  // Own echoes are recognised by the bound replica: it must be the stream's (T13304).
+  const bound = activeReplica(db, tableScopeOf(t))?.replicaId ?? null;
+  if (bound !== replicaId) {
+    throw vaultError(
+      'E_NEXUS_SYNC_REFUSED',
+      `this store is bound to replica ${bound ?? '(none)'}, but ${t.streamId} knows it as ${replicaId}`,
+      'nothing was pulled; relink the project (`cleo project link`) so both name one replica',
+    );
+  }
+  // With sync.pull off, pullStream refuses at once: no network round trip.
+  if (!isSyncFlagOn(db, 'sync.pull')) {
+    return pullStream(db, {
+      scope: tableScopeOf(t),
+      stream: t.streamId,
+      replica: replicaId,
+      initialCursor: initialPullCursor(),
+      pull: async () => ({ segments: [], cursor: initialPullCursor(), head: 0 }),
+      verify: () => 0,
+      seal: () => {},
+    });
+  }
+  const head = await streamHead(conn, t.streamId);
+  const checkpoints = head.headCheckpointId ? await listCheckpoints(conn, t.streamId) : [];
+  // A store with no pull position starts after the checkpoint it last synced
+  // (the genesis it cut, or the snapshot it restored), seeded from its signed
+  // replica map.
+  let initialCursor = readStreamCursor(db, t.streamId);
+  if (initialCursor === null) {
+    const { state: synced } = syncedState(conn, t, checkpoints, head.headCheckpointId);
+    const from = checkpoints.find((c) => c.checkpointId === synced?.lastCheckpointId) ?? null;
+    if (from === null) {
+      throw vaultError(
+        'E_NEXUS_SYNC_REFUSED',
+        `this store has synced no checkpoint of ${t.streamId}, so it has no position to pull from`,
+        'restore the stream (`cleo cloud restore`), or enable push (`cleo sync enable push`), first',
+      );
+    }
+    // T13306: a vault snapshot from before the stream's journal genesis. What
+    // the genesis device changed between that snapshot and its cut was folded
+    // into the genesis checkpoint, never sent as segments: pulling across it
+    // would silently diverge.
+    const genesis = isJournalSnapshot(from)
+      ? undefined
+      : checkpoints.find((c) => isJournalSnapshot(c) && c.coversSeq >= from.coversSeq);
+    if (genesis) {
+      throw vaultError(
+        'E_NEXUS_SYNC_REFUSED',
+        `${t.streamId} started its change journal at checkpoint ${genesis.checkpointId}, after the vault snapshot ${from.checkpointId} this store last synced: the changes folded into that journal checkpoint never travel as segments, so pulling from here would silently diverge`,
+        'nothing was pulled; restore the journal checkpoint first (`cleo cloud restore`), then join the stream with `cleo sync enable push` (T12999)',
+      );
+    }
+    journal.verifyCheckpoint(from, key.signers);
+    initialCursor = cursorFromCheckpoint(from);
+  }
+  // Seen-txn rows are never pruned here yet: a txn already seen can come
+  // back in a NEW segment above any checkpoint (a retired replica's late
+  // segment, a rebind re-pushing an upload that was stored but never
+  // recorded), and with its first-delivery row gone it would apply twice.
+  // Pruning waits for a floor the applier refuses below anyway, such as the
+  // receive watermark (T13256), and then prunes by that floor, not by
+  // stream seq.
+  return pullStream(db, {
+    scope: tableScopeOf(t),
+    stream: t.streamId,
+    replica: replicaId,
+    initialCursor,
+    pull: async (cursor) => {
+      const page = await journal.pull(
+        {
+          after: cursor.after,
+          knowsAllReplicas: cursor.knowsAllReplicas,
+          replicas: { ...cursor.replicas },
+        },
+        key.signers,
+      );
+      return {
+        segments: page.segments.map((s) => ({
+          seq: s.seq,
+          replicaId: s.replicaId,
+          replicaSeq: s.replicaSeq,
+          deviceId: s.deviceId,
+          plaintext: s.plaintext,
+          schemaVersion: s.meta.schemaVersion,
+        })),
+        cursor: page.cursor,
+        head: page.head,
+      };
+    },
+    verify: (deviceId, txns) => {
+      const keys = signerKeys(key.signers, deviceId);
+      const first = keys[0];
+      if (first === undefined) return 0;
+      for (const k of keys) {
+        if (firstBadTxnSignature(k.publicKey, t.streamId, txns) === null) return null;
+      }
+      return firstBadTxnSignature(first.publicKey, t.streamId, txns);
+    },
+    seal: () => {
+      sealPending(db, { scope: tableScopeOf(t), replica: replicaId });
+    },
+  });
+}
+
 /** The keyed hash of an empty table (a table the parent lists that this store no longer has). */
 function buildEmptyHash(t: VaultTarget, table: string): string {
   return emptyVaultTableHash(hashKeyOf(t.dataKey), table);
@@ -2812,6 +2940,22 @@ export function pushSyncStream(
   return mapped(() => pushSyncStreamImpl(opts));
 }
 
+/**
+ * Pull this store's stream journal (journal spec §3.1; T12343 S5-1): verified
+ * segments after the store's cursor are decoded, signature-checked, staged
+ * (a re-delivered transaction never twice, a vault delta passed over) and
+ * applied. A store with no pull position starts after the checkpoint it last
+ * synced. `cleo cloud sync` (T12996) runs it with push.
+ *
+ * @param opts - Scope and overrides.
+ * @returns What was received, staged and applied.
+ * @throws {NexusAccountError} `E_NEXUS_SYNC_REFUSED` (replica mismatch, no position),
+ *   `E_NEXUS_VAULT_NOT_LINKED`, or a mapped API error.
+ */
+export function pullSyncStream(opts: NexusVaultCommandOptions = {}): Promise<PullStreamReport> {
+  return mapped(() => pullSyncStreamImpl(opts));
+}
+
 /**
  * `cleo cloud pull` / `cleo cloud restore`: verify and activate a snapshot.
  *

packages/core/src/store/__tests__/table-classification-gate.test.ts

@@ -199,6 +199,8 @@ const SYNC_REBASE_DDL =
   'packages/core/migrations/sync-journal/20261006120000_t13193-rebase-undo/migration.sql';
 const SYNC_SEGMENT_DDL =
   'packages/core/migrations/sync-journal/20261007120000_t12343-segment-outbox/migration.sql';
+const SYNC_PULL_DDL =
+  'packages/core/migrations/sync-journal/20261008120000_t12343-pull-cursor/migration.sql';
 const SYNC_JOURNAL_TABLES = {
   _sync_capture: { class: 'local-only', ddl: SYNC_CAPTURE_DDL },
   _sync_frame: { class: 'local-only', ddl: SYNC_CAPTURE_DDL },
@@ -217,6 +219,8 @@ const SYNC_JOURNAL_TABLES = {
   _sync_sequenced: { class: 'local-only', ddl: SYNC_REBASE_DDL },
   _sync_segment: { class: 'local-only', ddl: SYNC_SEGMENT_DDL },
   _sync_segment_txn: { class: 'local-only', ddl: SYNC_SEGMENT_DDL },
+  _sync_cursor: { class: 'local-only', ddl: SYNC_PULL_DDL },
+  _sync_seen_txn: { class: 'local-only', ddl: SYNC_PULL_DDL },
   _sync_clock: { class: 'local-only', ddl: SYNC_JOURNAL_DDL },
   _sync_meta: { class: 'local-only', ddl: SYNC_JOURNAL_DDL },
   _sync_replica: { class: 'local-only', ddl: SYNC_JOURNAL_DDL },