diff --git a/.changeset/t12343-pull.md b/.changeset/t12343-pull.md
new file mode 100644
index 0000000000..912d961a62
--- /dev/null
+++ b/.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.
diff --git a/packages/contracts/src/nexus-account.ts b/packages/contracts/src/nexus-account.ts
index f5ef38f2e8..443e9c7ffa 100644
--- a/packages/contracts/src/nexus-account.ts
+++ b/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. */
diff --git a/packages/core/migrations/sync-journal/20261008120000_t12343-pull-cursor/migration.sql b/packages/core/migrations/sync-journal/20261008120000_t12343-pull-cursor/migration.sql
new file mode 100644
index 0000000000..f2a7e131ff
--- /dev/null
+++ b/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;
diff --git a/packages/core/src/cloud/__tests__/nexus-attach.test.ts b/packages/core/src/cloud/__tests__/nexus-attach.test.ts
index e180eb624d..780a40a5e1 100644
--- a/packages/core/src/cloud/__tests__/nexus-attach.test.ts
+++ b/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',
diff --git a/packages/core/src/cloud/__tests__/nexus-vault.test.ts b/packages/core/src/cloud/__tests__/nexus-vault.test.ts
index f4f3a979c2..91284af8b2 100644
--- a/packages/core/src/cloud/__tests__/nexus-vault.test.ts
+++ b/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 })));
diff --git a/packages/core/src/cloud/nexus-vault.ts b/packages/core/src/cloud/nexus-vault.ts
index cbfe7306b6..f4f2e967bd 100644
--- a/packages/core/src/cloud/nexus-vault.ts
+++ b/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.
  *
diff --git a/packages/core/src/store/__tests__/table-classification-gate.test.ts b/packages/core/src/store/__tests__/table-classification-gate.test.ts
index 7a9a7c1627..47f983574e 100644
--- a/packages/core/src/store/__tests__/table-classification-gate.test.ts
+++ b/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 },
diff --git a/packages/core/src/store/sync/__tests__/pull.test.ts b/packages/core/src/store/sync/__tests__/pull.test.ts
new file mode 100644
index 0000000000..ebca26a7c7
--- /dev/null
+++ b/packages/core/src/store/sync/__tests__/pull.test.ts
@@ -0,0 +1,275 @@
+/**
+ * Pulling a stream's journal (T12343 S5-1; journal spec §3.1, §2.8).
+ *
+ * Two real stores: A authors and persists segments with the real sealer,
+ * segment builder and transaction signatures; a fake stream serves them;
+ * B pulls, stages and applies.
+ *
+ * @task T12343
+ */
+
+import { mkdirSync, mkdtempSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join, resolve } from 'node:path';
+import type { DatabaseSync } from 'node:sqlite';
+import { SYNC_SCHEMA_VERSION } from '@cleocode/contracts/sync-schema.js';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { generateEd25519 } from '../../../cloud/crypto.js';
+import {
+  _resetDualScopeDbCache,
+  getDualScopeNativeDb,
+  openDualScopeDbAtPath,
+} from '../../dual-scope-db.js';
+import { finishCaptureFrame, openCaptureFrame, setCaptureEnabled } from '../capture.js';
+import { setSyncFlag } from '../flags.js';
+import {
+  type PulledStreamSegment,
+  pullStream,
+  readStreamCursor,
+  type StreamCursor,
+} from '../pull.js';
+import { sealPending } from '../sealer.js';
+import { buildSegment } from '../segments.js';
+import { firstBadTxnSignature, signTxn } from '../txn-signing.js';
+
+const SYNC_SCHEMA = resolve(import.meta.dirname, '../../../../migrations/sync-journal');
+const STREAM = 'project:0192ffff-7f00-7000-8000-00000000000f';
+const RA = '0192aaaa-7f00-7000-8000-00000000000a';
+const RB = '0192bbbb-7f00-7000-8000-00000000000b';
+const DEV_A = 'dev-a';
+const KEY_A = generateEd25519();
+let clock = Date.now();
+let dir: string;
+
+beforeEach(() => {
+  dir = mkdtempSync(join(tmpdir(), 'cleo-pull-'));
+  mkdirSync(join(dir, 'cleo'), { recursive: true });
+  vi.stubEnv('CLEO_HOME', join(dir, 'cleo'));
+  vi.stubEnv('CLEO_ROOT', undefined);
+  vi.stubEnv('CLEO_DIR', undefined);
+});
+
+afterEach(() => {
+  _resetDualScopeDbCache();
+  vi.unstubAllEnvs();
+  rmSync(dir, { recursive: true, force: true });
+});
+
+async function store(name: string): Promise<DatabaseSync> {
+  mkdirSync(join(dir, name, '.cleo'), { recursive: true });
+  const db = getDualScopeNativeDb(
+    await openDualScopeDbAtPath('project', join(dir, name, '.cleo', 'cleo.db')),
+  );
+  setCaptureEnabled(db, 'project', true, { schemaRoot: SYNC_SCHEMA });
+  setSyncFlag(db, 'sync.seal', true, { schemaRoot: SYNC_SCHEMA, allowUnreleased: true });
+  setSyncFlag(db, 'sync.pull', true, { schemaRoot: SYNC_SCHEMA, allowUnreleased: true });
+  return db;
+}
+
+const seal = (db: DatabaseSync, replica: string) => () => {
+  sealPending(db, {
+    scope: 'project',
+    replica,
+    now: () => ++clock,
+    env: {},
+    allowUnreleased: true,
+  });
+};
+
+function write(db: DatabaseSync, sql: string): void {
+  db.exec('BEGIN IMMEDIATE');
+  const frame = openCaptureFrame(db, 'write', null);
+  db.exec(sql);
+  finishCaptureFrame(db, frame);
+  db.exec('COMMIT');
+  seal(db, RA)();
+}
+
+const addTask = (id: string) =>
+  `INSERT INTO tasks_tasks (id, title, type, status, priority, uid, birth_fp) VALUES ('${id}', 'title ${id}', 'task', 'pending', 'medium', 'uid-${id}', 'fp-${id}')`;
+
+/** A's next segment, signed by A's device, as the stream serves it (the sealer is identity here). */
+function authorSegment(a: DatabaseSync): Uint8Array {
+  const seg = buildSegment(a, {
+    stream: STREAM,
+    replica: RA,
+    scope: 'project',
+    project: null,
+    sealer: (_seq, plaintext) => Buffer.from(plaintext),
+    signTxn: (stream, txn) => signTxn(KEY_A, stream, txn),
+    nowIso: new Date(++clock).toISOString(),
+  });
+  if (!seg) throw new Error('nothing to pack');
+  return seg.sealed;
+}
+
+/** A fake stream: segments in seq order, served after the cursor, one page at a time. */
+function fakeStream() {
+  const segments: PulledStreamSegment[] = [];
+  let pages = 0;
+  return {
+    segments,
+    get pages() {
+      return pages;
+    },
+    append(plaintext: Uint8Array, replicaSeq: number, replicaId = RA, deviceId = DEV_A): void {
+      segments.push({
+        seq: segments.length + 1,
+        replicaId,
+        replicaSeq,
+        deviceId,
+        plaintext,
+        schemaVersion: SYNC_SCHEMA_VERSION,
+      });
+    },
+    pull: async (cursor: StreamCursor) => {
+      pages += 1;
+      const page = segments.filter((s) => s.seq > cursor.after).slice(0, 2);
+      const replicas = { ...cursor.replicas };
+      for (const s of page)
+        replicas[s.replicaId] = { deviceId: s.deviceId, replicaSeq: s.replicaSeq };
+      return {
+        segments: page,
+        cursor: { after: page.at(-1)?.seq ?? cursor.after, knowsAllReplicas: true, replicas },
+        head: segments.length,
+      };
+    },
+  };
+}
+
+const START: StreamCursor = { after: 0, knowsAllReplicas: true, replicas: {} };
+
+const pullOpts = (b: DatabaseSync, stream: ReturnType<typeof fakeStream>) => ({
+  scope: 'project' as const,
+  stream: STREAM,
+  replica: RB,
+  pull: stream.pull,
+  verify: (deviceId: string, txns: Parameters<typeof firstBadTxnSignature>[2]) =>
+    deviceId === DEV_A ? firstBadTxnSignature(KEY_A.publicKey, STREAM, txns) : 0,
+  initialCursor: START,
+  now: () => ++clock,
+  env: {},
+  seal: seal(b, RB),
+});
+
+const n = (db: DatabaseSync, sql: string): number =>
+  Number((db.prepare(sql).get() as { n: number }).n);
+
+describe('pullStream (S5-1)', () => {
+  it("stages and applies another replica's segments in stream order, persisting the cursor; a rerun pulls nothing", async () => {
+    const a = await store('a');
+    const b = await store('b');
+    const stream = fakeStream();
+    write(a, addTask('T1'));
+    stream.append(authorSegment(a), 0);
+    write(a, addTask('T2'));
+    stream.append(authorSegment(a), 1);
+    write(a, "UPDATE tasks_tasks SET priority = 'high' WHERE id = 'T1'");
+    stream.append(authorSegment(a), 2);
+    const r = await pullStream(b, pullOpts(b, stream));
+    expect(r).toMatchObject({ segments: 3, staged: 3, redelivered: 0, after: 3, head: 3 });
+    expect(r.apply?.applied).toBe(3);
+    expect(b.prepare("SELECT priority FROM tasks_tasks WHERE id = 'T1'").get()).toEqual({
+      priority: 'high',
+    });
+    expect(readStreamCursor(b, STREAM)).toMatchObject({ after: 3 });
+    const again = await pullStream(b, pullOpts(b, stream));
+    expect(again).toMatchObject({ segments: 0, staged: 0, after: 3 });
+  });
+
+  it('a re-delivered transaction is never staged or applied twice', async () => {
+    const a = await store('a');
+    const b = await store('b');
+    const stream = fakeStream();
+    write(a, addTask('T1'));
+    const first = authorSegment(a);
+    stream.append(first, 0);
+    await pullStream(b, pullOpts(b, stream));
+    // The same transaction arrives again in a later segment (a replayed or re-sealed segment).
+    stream.append(first, 1);
+    const r = await pullStream(b, pullOpts(b, stream));
+    expect(r).toMatchObject({ segments: 1, staged: 0, redelivered: 1 });
+    expect(n(b, 'SELECT count(*) AS n FROM _sync_inbox')).toBe(1);
+    expect(n(b, "SELECT count(*) AS n FROM tasks_tasks WHERE id = 'T1'")).toBe(1);
+    expect(n(b, 'SELECT count(*) AS n FROM _sync_conflict')).toBe(0);
+  });
+
+  it('a vault delta segment carries no ops and is passed over', async () => {
+    const a = await store('a');
+    const b = await store('b');
+    const stream = fakeStream();
+    stream.append(
+      Buffer.from(
+        JSON.stringify({ kind: 'cleo-vault-delta/v1', base: 'cp-1', tables: { tasks_tasks: 2 } }),
+      ),
+      0,
+    );
+    write(a, addTask('T1'));
+    stream.append(authorSegment(a), 1);
+    const r = await pullStream(b, pullOpts(b, stream));
+    expect(r).toMatchObject({ segments: 2, vaultDeltas: 1, staged: 1, after: 2 });
+    expect(n(b, "SELECT count(*) AS n FROM tasks_tasks WHERE id = 'T1'")).toBe(1);
+  });
+
+  it('a refused segment stops the pull there: what precedes it is applied, the cursor stops before it, a retry re-stages nothing (T13307)', async () => {
+    const a = await store('a');
+    const b = await store('b');
+    const stream = fakeStream();
+    write(a, addTask('T1'));
+    stream.append(authorSegment(a), 0);
+    write(a, addTask('T2'));
+    stream.append(authorSegment(a), 1);
+    write(a, addTask('T3'));
+    stream.append(authorSegment(a), 2, RA, 'dev-mallory'); // page 2: served under another device's pin
+    const r = await pullStream(b, pullOpts(b, stream));
+    expect(r.refused).toMatch(/segment 3 of replica .*not signed by device dev-mallory/);
+    expect(r).toMatchObject({ segments: 2, staged: 2, after: 2 });
+    expect(r.apply?.applied).toBe(2);
+    expect(n(b, "SELECT count(*) AS n FROM tasks_tasks WHERE id IN ('T1', 'T2')")).toBe(2);
+    expect(n(b, "SELECT count(*) AS n FROM tasks_tasks WHERE id = 'T3'")).toBe(0);
+    expect(readStreamCursor(b, STREAM)).toMatchObject({ after: 2 });
+    const retry = await pullStream(b, pullOpts(b, stream));
+    expect(retry).toMatchObject({ segments: 0, staged: 0, after: 2 });
+    expect(retry.refused).toMatch(/segment 3/);
+  });
+
+  it('a refused segment mid-page: the segments before it in that page are staged and applied', async () => {
+    const a = await store('a');
+    const b = await store('b');
+    const stream = fakeStream();
+    write(a, addTask('T1'));
+    stream.append(authorSegment(a), 0);
+    stream.append(Buffer.from('not a segment'), 1); // same page as seq 1
+    const r = await pullStream(b, pullOpts(b, stream));
+    expect(r.refused).toMatch(/segment 2 of replica .*not a ledger segment/);
+    expect(r).toMatchObject({ segments: 1, staged: 1, after: 1 });
+    expect(n(b, "SELECT count(*) AS n FROM tasks_tasks WHERE id = 'T1'")).toBe(1);
+  });
+
+  it('seen-txn rows at or below the prune floor are dropped, later ones kept', async () => {
+    const a = await store('a');
+    const b = await store('b');
+    const stream = fakeStream();
+    write(a, addTask('T1'));
+    stream.append(authorSegment(a), 0);
+    write(a, addTask('T2'));
+    stream.append(authorSegment(a), 1);
+    await pullStream(b, pullOpts(b, stream));
+    expect(n(b, 'SELECT count(*) AS n FROM _sync_seen_txn')).toBe(2);
+    await pullStream(b, { ...pullOpts(b, stream), pruneSeenUpTo: 1 });
+    expect(b.prepare('SELECT seq FROM _sync_seen_txn').all()).toEqual([{ seq: 2 }]);
+  });
+
+  it('refuses with sync.pull off: nothing pulled, nothing staged', async () => {
+    const a = await store('a');
+    const b = await store('b');
+    const stream = fakeStream();
+    write(a, addTask('T1'));
+    stream.append(authorSegment(a), 0);
+    setSyncFlag(b, 'sync.pull', false, { schemaRoot: SYNC_SCHEMA });
+    const r = await pullStream(b, pullOpts(b, stream));
+    expect(r).toMatchObject({ refused: 'sync.pull is off', segments: 0, apply: null });
+    expect(stream.pages).toBe(0);
+    expect(n(b, 'SELECT count(*) AS n FROM _sync_inbox')).toBe(0);
+  });
+});
diff --git a/packages/core/src/store/sync/pull.ts b/packages/core/src/store/sync/pull.ts
new file mode 100644
index 0000000000..7a8c394007
--- /dev/null
+++ b/packages/core/src/store/sync/pull.ts
@@ -0,0 +1,347 @@
+/**
+ * Pull one stream's journal into the store (journal spec §3.1 "Receive",
+ * §2.9; T12343 S5-1).
+ *
+ * {@link pullStream} pages through the stream after the store's persisted
+ * `_sync_cursor` (the journal client verifies order, gaps, the replica-to-
+ * device pin, trust, hashes and decryption before anything is returned),
+ * and for each segment, in ONE local transaction per page:
+ * - decodes it: `deflate-raw(canonical JSON of LedgerTxn[])`, parsed with
+ *   the ledger schema; a vault delta segment (`cleo-vault-delta/v1`, a count
+ *   correction a vault push appended before the stream's genesis) carries no
+ *   ops and is passed over;
+ * - checks every transaction's signature against the segment's device key
+ *   (§2.8); one bad signature refuses the segment, and the pull stops there:
+ *   the segments before it are staged and applied, the segment and what
+ *   follows are not;
+ * - skips every transaction id already staged from this stream
+ *   (`_sync_seen_txn`): a re-delivered transaction is never staged twice, so
+ *   a counter delta is never applied twice;
+ * - stages the rest (`stageTxns`), schema-ahead ones included (the applier
+ *   holds them `refused-schema` and replays them after an upgrade);
+ * - advances the cursor.
+ *
+ * Then the applier applies what is staged, in stream order, own echoes
+ * included (they feed the fast path and the scoped rebase, §3.5).
+ *
+ * The page source is a port ({@link SegmentPuller}): the journal client in
+ * production (`cloud/nexus-vault.ts`), a fake in tests.
+ *
+ * @task T12343
+ * @module store/sync/pull
+ */
+
+import type { DatabaseSync } from 'node:sqlite';
+import { inflateRawSync } from 'node:zlib';
+import type { TableScope } from '@cleocode/contracts';
+import { LedgerTxn } from '@cleocode/contracts/ledger';
+import { type ApplyReport, type ApplyStagedOptions, applyStagedTxns } from './apply/applier.js';
+import { withImmediateTransaction } from './clock-store.js';
+import { isSyncFlagOn } from './flags.js';
+import { stageTxns } from './inbox.js';
+import { hasTable } from './schema.js';
+
+/** Where a pull stands (the journal client's `PullCursor`). */
+export interface StreamCursor {
+  /** The last stream seq staged. */
+  readonly after: number;
+  /** Whether `replicas` lists every replica with a segment at or below `after`. */
+  readonly knowsAllReplicas: boolean;
+  /** Per replica: the device pinned to it and its last replicaSeq staged. */
+  readonly replicas: Readonly<
+    Record<string, { readonly deviceId: string; readonly replicaSeq: number }>
+  >;
+}
+
+/** One verified, decrypted segment, as the journal client returns it. */
+export interface PulledStreamSegment {
+  readonly seq: number;
+  readonly replicaId: string;
+  readonly replicaSeq: number;
+  readonly deviceId: string;
+  readonly plaintext: Uint8Array;
+  readonly schemaVersion: number;
+}
+
+/** Pulls the next verified page after `cursor`. */
+export type SegmentPuller = (cursor: StreamCursor) => Promise<{
+  readonly segments: readonly PulledStreamSegment[];
+  readonly cursor: StreamCursor;
+  readonly head: number;
+}>;
+
+/** The index of a segment's first transaction its device did not sign, or null when every one is signed. */
+export type TxnVerifier = (deviceId: string, txns: readonly LedgerTxn[]) => number | null;
+
+/** The plaintext kind of a vault count-delta segment (`cloud/nexus-vault.ts`). */
+export const VAULT_DELTA_KIND = 'cleo-vault-delta/v1';
+
+/** Options for {@link pullStream}. */
+export interface PullStreamOptions {
+  readonly scope: TableScope;
+  readonly stream: string;
+  /** This store's bound replica (own echoes are recognised by it). */
+  readonly replica: string;
+  readonly pull: SegmentPuller;
+  readonly verify: TxnVerifier;
+  /** The cursor to start from when the store has none persisted (the checkpoint it was cut from or restored). */
+  readonly initialCursor: StreamCursor;
+  /** Wall clock. @defaultValue Date.now */
+  readonly now?: () => number;
+  /** Seals pending local captures before a rebase (the applier's `seal`). */
+  readonly seal: ApplyStagedOptions['seal'];
+  /** Applier options passed through (pages, budget). */
+  readonly apply?: Partial<Pick<ApplyStagedOptions, 'pageOps' | 'pageMs' | 'undoBudgetBytes'>>;
+  /** Environment for the `sync.pull` kill switch. @defaultValue process.env */
+  readonly env?: NodeJS.ProcessEnv;
+  /**
+   * Drop seen-txn rows first staged at or below this stream seq, or null to
+   * keep every row. Only a floor the applier refuses below is safe
+   * ({@link pruneSeenTxns}); the cloud pull passes none yet (T13256).
+   */
+  readonly pruneSeenUpTo?: number | null;
+}
+
+/** What {@link pullStream} did. */
+export interface PullStreamReport {
+  readonly stream: string;
+  /**
+   * Why the pull stopped short, or null: `sync.pull` off (nothing pulled), or
+   * a refused segment (a malformed body or a transaction its device did not
+   * sign), naming its seq, replica and device. Everything before a refused
+   * segment was staged and applied, and the cursor stops just before it.
+   */
+  readonly refused: string | null;
+  /** Segments received, vault deltas passed over, transactions staged, re-deliveries skipped. */
+  readonly segments: number;
+  readonly vaultDeltas: number;
+  readonly staged: number;
+  readonly redelivered: number;
+  /** The stream position the store has staged up to, and the server's head. */
+  readonly after: number;
+  readonly head: number;
+  /** What the applier did, or null when the pull was refused. */
+  readonly apply: ApplyReport | null;
+}
+
+/** A segment the store refuses to stage (a malformed body or a bad transaction signature). */
+export class SegmentRefusedError extends Error {
+  readonly code = 'E_SYNC_SEGMENT_REFUSED';
+}
+
+/** The store's persisted cursor for `stream`, or null. */
+export function readStreamCursor(db: DatabaseSync, stream: string): StreamCursor | null {
+  if (!hasTable(db, '_sync_cursor')) return null;
+  const row = db.prepare('SELECT cursor_json FROM _sync_cursor WHERE stream = ?').get(stream) as
+    | { cursor_json: string }
+    | undefined;
+  return row ? (JSON.parse(row.cursor_json) as StreamCursor) : null;
+}
+
+/** Decode a segment's transactions, or `null` for a vault delta segment. */
+function decode(seg: PulledStreamSegment): LedgerTxn[] | null {
+  let text: string;
+  try {
+    text = inflateRawSync(seg.plaintext).toString('utf8');
+  } catch {
+    // Not deflated: a vault delta segment is plain JSON.
+    try {
+      const raw: unknown = JSON.parse(Buffer.from(seg.plaintext).toString('utf8'));
+      if (
+        typeof raw === 'object' &&
+        raw !== null &&
+        'kind' in raw &&
+        raw.kind === VAULT_DELTA_KIND
+      ) {
+        return null;
+      }
+    } catch {
+      // fall through to the refusal
+    }
+    // @sync-invariant none:input-shape a segment body that is neither a ledger segment nor a vault delta is never staged
+    throw new SegmentRefusedError(
+      `segment ${seg.seq} of replica ${seg.replicaId}: not a ledger segment`,
+    );
+  }
+  const parsed = LedgerTxn.array().safeParse(JSON.parse(text));
+  if (!parsed.success) {
+    // @sync-invariant none:input-shape a ledger segment that does not parse is never staged
+    throw new SegmentRefusedError(
+      `segment ${seg.seq} of replica ${seg.replicaId}: transactions do not parse`,
+    );
+  }
+  return parsed.data;
+}
+
+/** A segment's transactions (null for a vault delta), or why it is refused. */
+function refusalOf(
+  seg: PulledStreamSegment,
+  verify: TxnVerifier,
+): { readonly txns: LedgerTxn[] | null } | string {
+  let txns: LedgerTxn[] | null;
+  try {
+    txns = decode(seg);
+  } catch (err) {
+    if (err instanceof SegmentRefusedError) return err.message;
+    throw err;
+  }
+  if (txns !== null) {
+    const bad = verify(seg.deviceId, txns);
+    if (bad !== null) {
+      return `segment ${seg.seq} of replica ${seg.replicaId}: transaction ${bad} is not signed by device ${seg.deviceId}`;
+    }
+  }
+  return { txns };
+}
+
+/** `cursor` advanced over `segs` (a page cut short by a refused segment). */
+function advance(cursor: StreamCursor, segs: readonly PulledStreamSegment[]): StreamCursor {
+  const replicas = { ...cursor.replicas };
+  for (const s of segs) replicas[s.replicaId] = { deviceId: s.deviceId, replicaSeq: s.replicaSeq };
+  return {
+    after: segs.at(-1)?.seq ?? cursor.after,
+    knowsAllReplicas: cursor.knowsAllReplicas,
+    replicas,
+  };
+}
+
+/**
+ * Drop the seen-txn rows of `stream` first staged at or below `seq`, keeping
+ * `_sync_seen_txn` bounded (§3.1). Only safe below a floor the applier
+ * refuses any txn under: a seen txn can return in a NEW segment above a
+ * checkpoint (a retired replica's late segment, a rebind's re-push), so a
+ * checkpoint's coversSeq alone is not such a floor. No caller prunes until
+ * that floor exists (T13256).
+ *
+ * @param db - The store.
+ * @param stream - The stream.
+ * @param seq - The floor, inclusive.
+ * @returns How many rows were dropped.
+ */
+export function pruneSeenTxns(db: DatabaseSync, stream: string, seq: number): number {
+  if (!hasTable(db, '_sync_seen_txn')) return 0;
+  return Number(
+    db.prepare('DELETE FROM _sync_seen_txn WHERE stream = ? AND seq <= ?').run(stream, seq).changes,
+  );
+}
+
+/**
+ * Pull and stage a stream's new segments, then apply (module docs). Must run
+ * outside a transaction. A refused segment stops the pull: everything
+ * before it is staged and applied, the cursor stops just before it, and the
+ * report's `refused` names it (T13307), so the next pull reaches it again
+ * without re-staging anything.
+ *
+ * @param db - The store.
+ * @param o - {@link PullStreamOptions}.
+ * @returns What was received, staged and applied.
+ */
+export async function pullStream(
+  db: DatabaseSync,
+  o: PullStreamOptions,
+): Promise<PullStreamReport> {
+  const now = o.now ?? Date.now;
+  if (!isSyncFlagOn(db, 'sync.pull', o.env ?? process.env)) {
+    return {
+      stream: o.stream,
+      refused: 'sync.pull is off',
+      segments: 0,
+      vaultDeltas: 0,
+      staged: 0,
+      redelivered: 0,
+      after: readStreamCursor(db, o.stream)?.after ?? o.initialCursor.after,
+      head: 0,
+      apply: null,
+    };
+  }
+  let cursor = readStreamCursor(db, o.stream) ?? o.initialCursor;
+  let segments = 0;
+  let vaultDeltas = 0;
+  let staged = 0;
+  let redelivered = 0;
+  let head = cursor.after;
+  let refusedSegment: string | null = null;
+  // Seen-txn rows at or below the floor can no longer be re-delivered.
+  if (o.pruneSeenUpTo !== undefined && o.pruneSeenUpTo !== null) {
+    pruneSeenTxns(db, o.stream, o.pruneSeenUpTo);
+  }
+  for (;;) {
+    const page = await o.pull(cursor);
+    head = page.head;
+    const nowIso = new Date(now()).toISOString();
+    // Decode and verify the page before writing anything of it, up to the
+    // first segment refused (T13307): what precedes it is staged and applied,
+    // the cursor stops just before it, and the refusal is reported.
+    const decoded: Array<{ seg: PulledStreamSegment; txns: LedgerTxn[] | null }> = [];
+    for (const seg of page.segments) {
+      const why = refusalOf(seg, o.verify);
+      if (typeof why === 'string') {
+        refusedSegment = why;
+        break;
+      }
+      decoded.push({ seg, txns: why.txns });
+    }
+    const pageCursor =
+      refusedSegment === null
+        ? page.cursor
+        : advance(
+            cursor,
+            decoded.map((d) => d.seg),
+          );
+    withImmediateTransaction(db, () => {
+      const seen = db.prepare(
+        'INSERT INTO _sync_seen_txn (stream, txn, seq) VALUES (?, ?, ?) ON CONFLICT (stream, txn) DO NOTHING',
+      );
+      for (const { seg, txns } of decoded) {
+        segments += 1;
+        if (txns === null) {
+          vaultDeltas += 1;
+          continue;
+        }
+        // A transaction already staged under an earlier segment is re-delivered: skip it.
+        const fresh = txns.filter((t) => Number(seen.run(o.stream, t.txn, seg.seq).changes) === 1);
+        redelivered += txns.length - fresh.length;
+        if (fresh.length === 0) continue;
+        staged += stageTxns(
+          db,
+          o.stream,
+          {
+            seq: seg.seq,
+            replicaId: seg.replicaId,
+            replicaSeq: seg.replicaSeq,
+            deviceId: seg.deviceId,
+            schemaVersion: seg.schemaVersion,
+            txns: fresh,
+          },
+          nowIso,
+        );
+      }
+      db.prepare(
+        'INSERT INTO _sync_cursor (stream, cursor_json, updated_at) VALUES (?, ?, ?) ' +
+          'ON CONFLICT (stream) DO UPDATE SET cursor_json = excluded.cursor_json, updated_at = excluded.updated_at',
+      ).run(o.stream, JSON.stringify(pageCursor), nowIso);
+    });
+    cursor = pageCursor;
+    if (refusedSegment !== null) break;
+    if (page.segments.length === 0 || cursor.after >= page.head) break;
+  }
+  const apply = applyStagedTxns(db, {
+    scope: o.scope,
+    stream: o.stream,
+    replica: o.replica,
+    now,
+    seal: o.seal,
+    ...(o.apply ?? {}),
+  });
+  return {
+    stream: o.stream,
+    refused: refusedSegment,
+    segments,
+    vaultDeltas,
+    staged,
+    redelivered,
+    after: cursor.after,
+    head,
+    apply,
+  };
+}
diff --git a/packages/core/src/store/table-classification.ts b/packages/core/src/store/table-classification.ts
index d1dba5ad7a..212cf159d0 100644
--- a/packages/core/src/store/table-classification.ts
+++ b/packages/core/src/store/table-classification.ts
@@ -196,6 +196,18 @@ const PROJECT_TABLES: Readonly<Record<string, TableRegistryEntry>> = {
     source: 'journal spec t12342-t12343-journal-design §2.8 (T12343)',
     note: 'the transactions each persisted segment carries, in order. This device only. Created lazily by store/sync/schema.ts when a sync.* flag is first enabled (T12343)',
   },
+  _sync_cursor: {
+    class: 'local-only',
+    status: 'optional-transient',
+    source: 'journal spec t12342-t12343-journal-design §3.1 Receive (T12343)',
+    note: "where this store's pull stands per stream: the PullCursor the journal client verifies the next page against, advanced in the transaction that stages what it covers. This device only. Created lazily by store/sync/schema.ts when a sync.* flag is first enabled (T12343)",
+  },
+  _sync_seen_txn: {
+    class: 'local-only',
+    status: 'optional-transient',
+    source: 'journal spec t12342-t12343-journal-design §3.1 Receive (T12343 S5-1)',
+    note: 'every transaction id staged from a stream, so a re-delivered transaction is never staged twice (a counter delta would double). This device only. Created lazily by store/sync/schema.ts when a sync.* flag is first enabled (T12343)',
+  },
   _sync_row_meta: {
     class: 'local-only',
     status: 'optional-transient',
@@ -1437,6 +1449,18 @@ const GLOBAL_TABLES: Readonly<Record<string, TableRegistryEntry>> = {
     source: 'journal spec t12342-t12343-journal-design §2.8 (T12343)',
     note: 'the transactions each persisted segment carries, in order. This device only. Created lazily by store/sync/schema.ts when a sync.* flag is first enabled (T12343)',
   },
+  _sync_cursor: {
+    class: 'local-only',
+    status: 'optional-transient',
+    source: 'journal spec t12342-t12343-journal-design §3.1 Receive (T12343)',
+    note: "where this store's pull stands per stream: the PullCursor the journal client verifies the next page against, advanced in the transaction that stages what it covers. This device only. Created lazily by store/sync/schema.ts when a sync.* flag is first enabled (T12343)",
+  },
+  _sync_seen_txn: {
+    class: 'local-only',
+    status: 'optional-transient',
+    source: 'journal spec t12342-t12343-journal-design §3.1 Receive (T12343 S5-1)',
+    note: 'every transaction id staged from a stream, so a re-delivered transaction is never staged twice (a counter delta would double). This device only. Created lazily by store/sync/schema.ts when a sync.* flag is first enabled (T12343)',
+  },
   _sync_row_meta: {
     class: 'local-only',
     status: 'optional-transient',
diff --git a/scripts/lint-no-raw-table-writes.mjs b/scripts/lint-no-raw-table-writes.mjs
index b0b0b36cab..30e8ec2e8f 100644
--- a/scripts/lint-no-raw-table-writes.mjs
+++ b/scripts/lint-no-raw-table-writes.mjs
@@ -396,6 +396,18 @@ export const EXEMPT = {
         'capture undo (local-only): dropped when a failed or raced genesis cut is undone and no stream is cut, so undo is off again (T13296)',
     },
   },
+  'packages/core/src/store/sync/pull.ts': {
+    _sync_cursor: {
+      count: 1,
+      reason:
+        'pull cursor (local-only): advanced in the transaction that stages the segments it covers (§3.1 step 3, T12343)',
+    },
+    _sync_seen_txn: {
+      count: 2,
+      reason:
+        'staged transaction ids (local-only): a re-delivered transaction is never staged twice, and rows past the latest verified checkpoint are pruned (§3.1, T12343 S5-1)',
+    },
+  },
   'packages/core/src/store/sync/push.ts': {
     _sync_meta: {
       count: 2,