feat(sync): prune seen txns below a per-origin local_seq floor, refuse loudly below it (T13318)

.changeset/t13318-seen-floor.md

@@ -0,0 +1,21 @@
+---
+id: t13318-seen-floor
+tasks: [T13318]
+kind: feat
+summary: "The cloud pull prunes its seen-transaction ledger below a per-origin local_seq floor, and refuses loudly below it"
+---
+
+A transaction id is `<replica>:<local_seq>`, and every origin's transactions are first delivered in `local_seq` order:
+the sealer mints `local_seq` rising, a segment packs the lowest sealed transactions in order, uploads go out lowest
+`replicaSeq` first, and every puller refuses a `replicaSeq` gap or replay. The pull now relies on that.
+
+- A new local-only journal table, `_sync_seen_floor`, keeps per stream and origin the highest `local_seq` staged, how far
+  its seen rows were pruned (`pruned_upto`), and the row and byte counts. Its migration seeds it from rows already staged.
+- A transaction at or below its origin's floor with no seen row is refused with `refusedKind: 'below-floor'`, and the pull
+  stops before its segment. The message says whether its seen row was pruned or it arrived out of order. It is never
+  skipped as seen.
+- A segment carrying a transaction that is not its replica's is refused (`refusedKind: 'segment'`).
+- The cloud pull prunes each origin's seen rows below its floor after every pull, keeping the floor's own row.
+  `pullStream`'s `pruneSeenUpTo` (a stream-seq floor no caller used) is replaced by `pruneSeen`.
+- `cleo cloud status` sums the per-origin counts instead of walking the whole ledger.
+- Vault bundles clear `_sync_seen_floor` together with `_sync_seen_txn` and `_sync_cursor`.

packages/contracts/src/nexus-cloud.ts

@@ -615,9 +615,9 @@ export interface CloudStatusUndoBudget {
 }
 
 /**
- * The pull's seen-transaction ledger (`_sync_seen_txn`, T13317). Nothing
- * prunes it yet: a seen txn can return in a new segment above any checkpoint,
- * so pruning waits for a proven refusal floor (T13318).
+ * The pull's seen-transaction ledger (`_sync_seen_txn`, T13317). The cloud
+ * pull prunes each origin's rows below its staged `local_seq` floor (T13318);
+ * a pruned transaction delivered again is refused, never applied twice.
  */
 export interface CloudStatusSeenTxns {
   /** Rows held, every stream. */

packages/core/migrations/sync-journal/20261009120000_t13318-seen-floor/migration.sql

@@ -0,0 +1,41 @@
+-- T13318 (journal spec §3.1 "Receive") — a per-origin floor for the
+-- seen-transaction ledger.
+--
+-- A transaction id is `<replica>:<local_seq>`, and every origin's
+-- transactions are FIRST delivered in local_seq order (the sealer mints
+-- local_seq rising, a segment packs the lowest sealed ones in order, and
+-- pullers refuse a replicaSeq gap or replay). So per (stream, origin):
+--
+-- staged_upto: the highest local_seq staged. A transaction at or below it
+--   that has no _sync_seen_txn row was either pruned or arrived out of order;
+--   the pull refuses it loudly, never skips it as seen.
+-- pruned_upto: seen rows at or below it were dropped (0: none), so a refusal
+--   can say which of the two it is.
+-- seen_rows, seen_bytes: the origin's _sync_seen_txn rows and their payload
+--   bytes, kept with every insert and prune, so `cleo cloud status` sums this
+--   table instead of walking the ledger.
+--
+-- The backfill seeds the floor from rows staged before this table existed.
+--
+-- Local-only.
+--
+-- Do not edit this file once applied: its hash is recorded per store.
+CREATE TABLE IF NOT EXISTS _sync_seen_floor (
+  stream      TEXT NOT NULL,
+  origin      TEXT NOT NULL,
+  staged_upto INTEGER NOT NULL,
+  pruned_upto INTEGER NOT NULL DEFAULT 0,
+  seen_rows   INTEGER NOT NULL DEFAULT 0,
+  seen_bytes  INTEGER NOT NULL DEFAULT 0,
+  PRIMARY KEY (stream, origin)
+) WITHOUT ROWID;
+INSERT OR IGNORE INTO _sync_seen_floor (stream, origin, staged_upto, pruned_upto, seen_rows, seen_bytes)
+SELECT stream,
+       substr(txn, 1, instr(txn, ':') - 1),
+       max(CAST(substr(txn, instr(txn, ':') + 1) AS INTEGER)),
+       0,
+       count(*),
+       sum(octet_length(stream) + octet_length(txn) + 8)
+  FROM _sync_seen_txn
+ WHERE instr(txn, ':') > 1
+ GROUP BY stream, substr(txn, 1, instr(txn, ':') - 1);

packages/core/src/cloud/__tests__/cloud-sync-status.test.ts

@@ -170,19 +170,26 @@ describe('cloud status sync block (T12998)', () => {
     expect(project?.quarantined).toEqual({ tasks_tasks: 1 });
   });
 
-  it('reports the seen-txn ledger per stream with its estimated bytes; nothing prunes it (T13317)', async () => {
+  it('reports the seen-txn ledger per stream from the per-origin counts the pull keeps (T13317, T13318)', async () => {
     const db = await store(true);
-    const ins = db.prepare('INSERT INTO _sync_seen_txn (stream, txn, seq) VALUES (?, ?, ?)');
-    ins.run('project:p1', 'r1:1', 1);
-    ins.run('project:p1', 'r1:2', 2);
-    ins.run('home:u1', 'r2:1', 1);
+    // stream + txn text + an 8-byte seq per row, as the pull counts them.
+    const row = (stream: string, txn: string) => stream.length + txn.length + 8;
+    const floor = db.prepare(
+      'INSERT INTO _sync_seen_floor (stream, origin, staged_upto, seen_rows, seen_bytes) VALUES (?, ?, ?, ?, ?)',
+    );
+    floor.run('project:p1', 'r1', 2, 2, row('project:p1', 'r1:1') + row('project:p1', 'r1:2'));
+    floor.run('project:p1', 'r3', 7, 0, 0); // fully pruned: no rows, no stream entry of its own
+    floor.run('home:u1', 'r2', 1, 1, row('home:u1', 'r2:1'));
+    // The ledger itself is never walked: a stray row with no count is not reported.
+    db.prepare('INSERT INTO _sync_seen_txn (stream, txn, seq) VALUES (?, ?, ?)').run(
+      'project:p9',
+      'r9:1',
+      1,
+    );
     const { project } = await projectBlock();
-    // stream + txn text + an 8-byte seq per row.
-    const bytes =
-      2 * ('project:p1'.length + 'r1:1'.length + 8) + ('home:u1'.length + 'r2:1'.length + 8);
     expect(project?.seenTxns).toEqual({
       rows: 3,
-      bytes,
+      bytes: 2 * row('project:p1', 'r1:1') + row('home:u1', 'r2:1'),
       byStream: { 'home:u1': 1, 'project:p1': 2 },
     });
     // A read never prunes: the next read sees the same rows.

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

@@ -319,6 +319,7 @@ describe('presence and binding hold their guarantees (review M3)', () => {
         '_sync_replica',
         '_sync_row_meta',
         '_sync_row_undo',
+        '_sync_seen_floor',
         '_sync_seen_txn',
         '_sync_segment',
         '_sync_segment_txn',

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

@@ -826,6 +826,9 @@ const VAULT_CLEARED_JOURNAL_TABLES = [
   // from the checkpoint it restored, never from another device's position.
   '_sync_cursor',
   '_sync_seen_txn',
+  // Its floor goes with it: a floor without its seen rows would refuse
+  // every transaction below it (T13318).
+  '_sync_seen_floor',
 ] as const;
 
 /**
@@ -1981,18 +1984,16 @@ async function pullWithSession(session: StreamSession): Promise<PullStreamReport
     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.
+  // Seen-txn rows below each origin's staged floor are pruned after every
+  // pull (T13318): an origin's transactions are first delivered in local_seq
+  // order, so one at or below the floor with no seen row is refused loudly
+  // (`below-floor`), never applied twice.
   return pullStream(db, {
     scope: tableScopeOf(t),
     stream: t.streamId,
     replica: replicaId,
     initialCursor,
+    pruneSeen: true,
     pull: async (cursor) => {
       const page = await journal.pull(
         {

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

@@ -201,6 +201,8 @@ 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_SEEN_FLOOR_DDL =
+  'packages/core/migrations/sync-journal/20261009120000_t13318-seen-floor/migration.sql';
 const SYNC_JOURNAL_TABLES = {
   _sync_capture: { class: 'local-only', ddl: SYNC_CAPTURE_DDL },
   _sync_frame: { class: 'local-only', ddl: SYNC_CAPTURE_DDL },
@@ -221,6 +223,7 @@ const SYNC_JOURNAL_TABLES = {
   _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_seen_floor: { class: 'local-only', ddl: SYNC_SEEN_FLOOR_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 },