fix(T13867): a snapshot-only store is never up to date while the journal moved past the head

.changeset/t13867-journal-unread-stale.md

@@ -0,0 +1,24 @@
+---
+id: t13867-journal-unread-stale
+tasks: [T13867]
+kind: fix
+summary: a store that reads only snapshots is no longer reported up to date while the change journal has moved past the newest snapshot
+---
+
+A store that does not read the change journal (it restored a snapshot and
+never ran `cleo sync enable push`) holds only the newest snapshot. When other
+devices then wrote through the journal, `cleo cloud pull` still said
+`up-to-date` and `cleo cloud verify` said `match`, while every journal change
+was missing. Now that pull refuses with `E_NEXUS_VAULT_BEHIND`, verify says
+`behind`, and a restore warns (`W_NEXUS_VAULT_JOURNAL_UNREAD`), each naming
+how many journal segments are missing and the way forward: `cleo sync enable
+push`, or a vault push from a device that writes the journal.
+
+On a journal stream, `cleo cloud verify` no longer reports `ahead` only
+because a table that snapshots alone carry (the file inventory, other
+databases, the device and project registries) differs on this machine. Those
+tables are still listed, with a `W_NEXUS_VAULT_SNAPSHOT_ONLY` warning.
+
+Every request now sends `x-cleo-stream-reader: 3`, so a Cleo Nexus server can
+refuse a journal stream to a client that reads only snapshots. Released
+v2026.10.19 does not send it.

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

@@ -80,7 +80,13 @@ import {
   recordBareAccounts,
   setSyncFlag,
 } from '../../store/sync/flags.js';
-import { cutGenesis, genesisCutOf, genesisPending } from '../../store/sync/genesis.js';
+import {
+  cutGenesis,
+  GENESIS_CUT_KEY_PREFIX,
+  genesisCutOf,
+  genesisPending,
+  readGenesisCut,
+} from '../../store/sync/genesis.js';
 import { readStreamCursor } from '../../store/sync/pull.js';
 import {
   activeReplica,
@@ -109,7 +115,7 @@ import {
   uuidv7,
   verifyEd25519,
 } from '../crypto.js';
-import { type FetchLike, Http, NexusError } from '../http.js';
+import { type FetchLike, Http, NEXUS_STREAM_READER_HEADER, NexusError } from '../http.js';
 import {
   cursorFromCheckpoint,
   Journal,
@@ -423,6 +429,8 @@ class FakeNexus {
   beforeSegment: { deviceId: string; run: () => Promise<void> } | null = null;
   /** Codes the next checkpoint creates are refused with, in order. */
   refuseCheckpoint: string[] = [];
+  /** As Cleo Nexus with T13876: a v3 stream's head and checkpoint reads need the stream-reader header. */
+  streamReaderRequired = false;
   /** Runs once before the next checkpoint create by `deviceId` (simulates a concurrent author). */
   beforeCheckpoint: { deviceId: string; run: () => Promise<void> } | null = null;
   /** Runs once after the server stored a checkpoint, before the response reaches the client. */
@@ -502,6 +510,19 @@ class FakeNexus {
       const token = (new Headers(init?.headers).get('authorization') ?? '').replace(/^Bearer /, '');
       const device = [...this.devices.values()].find((d) => d.token === token);
       if (!device) throw new ApiFail(401, 'E_UNAUTHENTICATED');
+      if (this.streamReaderRequired && method === 'GET') {
+        const sm = url.pathname.match(/^\/v1\/streams\/([^/]+)(\/checkpoints)?$/);
+        const st = sm ? this.streams.get(decodeURIComponent(sm[1] ?? '')) : undefined;
+        const headCp = st?.checkpoints.find((c) => c.checkpointId === st.headCheckpointId);
+        if (
+          headCp &&
+          manifestVersion(headCp.manifest) === 3 &&
+          new Headers(init?.headers).get(NEXUS_STREAM_READER_HEADER.name) !==
+            NEXUS_STREAM_READER_HEADER.value
+        ) {
+          throw new ApiFail(409, 'E_STREAM_VERSION');
+        }
+      }
       const body =
         typeof init?.body === 'string' ? (JSON.parse(init.body) as Record<string, unknown>) : {};
       if (method !== 'GET') this.writes.push(`${method} ${url.pathname}`);
@@ -6163,6 +6184,67 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => {
       return { a, aDb, genesis, b, bDb };
     }
 
+    it('T13867: a store that reads snapshots only is behind once the journal moves past the head: pull refuses, verify says behind, never up-to-date or match', async () => {
+      fake.streamReaderRequired = true;
+      const { a, aDb } = await journalDevices(false);
+      // B restores the genesis and never joins the journal: it reads snapshots only.
+      const { b, bDb } = await restoredJoiner();
+      // The restored bundle carries no genesis cut of A's: B does not pass for a journal reader.
+      expect(await on(b, () => readGenesisCut(bDb, STREAM))).toBeUndefined();
+      expect((await on(b, () => restoreNexusVault(vopts(b, { mode: 'pull' })))).status).toBe(
+        'up-to-date',
+      );
+      expect((await on(b, () => verifyNexusVault(vopts(b)))).verdict).toBe('match');
+      // A writes through the journal: a segment past the head checkpoint.
+      await on(a, async () => (await storeOf(aDb)).exec(insertTask('T3', 'from A')));
+      await sync(a);
+      const pulled = await failure(on(b, () => restoreNexusVault(vopts(b, { mode: 'pull' }))));
+      expect(pulled.code).toBe('E_NEXUS_VAULT_BEHIND');
+      expect(pulled.message).toContain('reads snapshots only');
+      expect(pulled.fix).toContain('cleo sync enable push');
+      const verified = await on(b, () => verifyNexusVault(vopts(b)));
+      expect(verified.verdict).toBe('behind');
+      expect(verified.remedy).toContain('cleo sync enable push');
+      // A genesis cut alone (as a restored bundle could carry A's) does not make B a journal reader.
+      await on(b, async () => {
+        const db = await storeOf(bDb);
+        db.exec(
+          'CREATE TABLE IF NOT EXISTS _sync_meta (key TEXT PRIMARY KEY, value TEXT, updated_at TEXT)',
+        );
+        db.prepare(
+          'INSERT INTO _sync_meta (key, value, updated_at) VALUES (?, ?, ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value',
+        ).run(`${GENESIS_CUT_KEY_PREFIX}${STREAM}`, '0', '2026-10-17');
+      });
+      expect(await on(b, () => readGenesisCut(bDb, STREAM))).toBe(0);
+      const cutOnly = await failure(on(b, () => restoreNexusVault(vopts(b, { mode: 'pull' }))));
+      expect(cutOnly.code).toBe('E_NEXUS_VAULT_BEHIND');
+      expect((await on(b, () => verifyNexusVault(vopts(b)))).verdict).toBe('behind');
+      // A device that writes the journal is not held behind by it.
+      expect((await on(a, () => verifyNexusVault(vopts(a)))).verdict).toBe('match');
+      // This client sent the stream-reader header throughout; a client without it (released
+      // v2026.10.19) is refused a v3 stream's reads by a T13876 server, never shown it as current.
+      const old = await fake.fetch(`${API}/v1/streams/${encodeURIComponent(STREAM)}/checkpoints`, {
+        headers: { authorization: `Bearer ${b.token}` },
+      });
+      expect(old.status).toBe(409);
+      expect(JSON.stringify(await old.json())).toContain('E_STREAM_VERSION');
+    });
+
+    it('T13867: on a journal head, a table only snapshots carry is listed but does not decide the verdict', async () => {
+      const { a, aDb } = await journalDevices(false);
+      await on(a, async () =>
+        (await storeOf(aDb)).exec(
+          "INSERT INTO schedules (schedule_id, cron_expr, title) VALUES ('s-t13867', '0 * * * *', 'machine-local schedule')",
+        ),
+      );
+      const v = await on(a, () => verifyNexusVault(vopts(a)));
+      expect(v.tables.find((r) => r.table === 'schedules')?.match).toBe(false);
+      expect(v.verdict).toBe('match');
+      expect(v.warnings.find((w) => w.code === 'W_NEXUS_VAULT_SNAPSHOT_ONLY')?.message).toContain(
+        'schedules',
+      );
+    });
+
     it('AC1: with sync.push on, cloud push checkpoints at the sealed journal hwm with no delta segment; the server v3 rule accepts it', async () => {
       const { a, aDb, genesis } = await journalDevices(true);
       // A write not sealed yet: the push seals and pushes it first.

packages/core/src/cloud/http.ts

@@ -59,6 +59,14 @@ export function isSecureUrl(url: string): boolean {
   }
 }
 
+/**
+ * The header saying this client reads journal (checkpoint/v3) streams (T13867).
+ * Cleo Nexus refuses a v3 stream's checkpoint and head reads to a request
+ * without it (T13876): such a client reads only snapshots and would report a
+ * stale store as current. Released v2026.10.19 does not send it.
+ */
+export const NEXUS_STREAM_READER_HEADER = { name: 'x-cleo-stream-reader', value: '3' } as const;
+
 /**
  * Minimal envelope-aware HTTP client. It retries only failures that are safe to retry. Every API
  * write is idempotent or lineage-checked, so a retried write either no-ops or is refused explicitly.
@@ -116,6 +124,7 @@ export class Http {
       try {
         const headers: Record<string, string> = { authorization: `Bearer ${this.o.token}` };
         if (this.o.deviceId) headers['x-cleo-device-id'] = this.o.deviceId;
+        headers[NEXUS_STREAM_READER_HEADER.name] = NEXUS_STREAM_READER_HEADER.value;
         if (body !== undefined) headers['content-type'] = 'application/json';
         res = await this.fetch(`${this.o.baseUrl}${path}`, {
           method,

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

@@ -582,6 +582,37 @@ async function replaySegments(
 const JOURNAL_STREAM_REMEDY =
   'on this stream local changes travel through the change journal (`sync.push`), and with it on a vault push checkpoints the journal (T12999); a store that has not joined it joins with `cleo sync enable push`, which keeps its own rows and merges them into the stream (T13466); `cleo cloud pull`, `cleo cloud restore` and `cleo cloud verify` still work';
 
+/** What a store that reads only snapshots does about journal changes past the head (T13867). */
+const JOURNAL_UNREAD_REMEDY =
+  'join the change journal with `cleo sync enable push` to receive them, or run `cleo cloud push` on a device that writes the journal (its push checkpoints the journal, T12999), then pull again';
+
+/**
+ * The change-journal segments past a checkpoint/v3 head that this store does
+ * not hold (T13867): it reads snapshots only, so the head is all it has, and
+ * every segment past the head's `coversSeq` is missing from it. A store is on
+ * the journal when it holds a genesis cut of the stream AND `sync.push` or
+ * `sync.pull` is on: a cut alone, as a restored bundle could carry, does not
+ * make it one. 0 for any other head, or a store on the journal.
+ */
+async function journalUnread(
+  t: VaultTarget,
+  headSeq: number,
+  headCp: Checkpoint | null | undefined,
+): Promise<number> {
+  if (!headCp || !isJournalSnapshot(headCp) || headSeq <= headCp.coversSeq) return 0;
+  if (fs.existsSync(t.dbPath)) {
+    const { openNativeDatabase } = await import('../store/sqlite-native.js');
+    const db = openNativeDatabase(t.dbPath, { readonly: true, enableWal: false });
+    try {
+      const on = isSyncFlagOn(db, 'sync.push') || isSyncFlagOn(db, 'sync.pull');
+      if (on && genesisCutOf(db, t.streamId) !== undefined) return 0;
+    } finally {
+      db.close();
+    }
+  }
+  return headSeq - headCp.coversSeq;
+}
+
 /** Whether a snapshot is checkpoint/v3: the change journal writes its stream (T13034). */
 function isJournalSnapshot(cp: Checkpoint | null | undefined): boolean {
   return cp !== null && cp !== undefined && manifestVersion(cp.manifest) === 3;
@@ -3788,6 +3819,22 @@ async function restoreNexusVaultImpl(opts: NexusVaultRestoreOptions): Promise<Cl
   const names = await deviceNames(conn);
   if (opts.mode === 'pull' && synced?.lastCheckpointId === target && opts.force !== true) {
     const cp = cps.find((c) => c.checkpointId === target) ?? null;
+    const unread =
+      target === head.headCheckpointId
+        ? await journalUnread(
+            t,
+            head.headSeq,
+            trusted.find((c) => c.checkpointId === target),
+          )
+        : 0;
+    if (unread > 0) {
+      // @sync-invariant none:local-only a read-only refusal: nothing is written
+      throw vaultError(
+        'E_NEXUS_VAULT_BEHIND',
+        `${t.streamId}: the change journal holds ${unread} segment(s) past snapshot ${target}, and this store reads snapshots only, so it is not up to date`,
+        JOURNAL_UNREAD_REMEDY,
+      );
+    }
     warnings.push(...conn.state.drainWarnings());
     return {
       apiUrl: conn.apiUrl,
@@ -4109,6 +4156,16 @@ async function restoreNexusVaultImpl(opts: NexusVaultRestoreOptions): Promise<Cl
         });
       }
     }
+    const unread =
+      restored.checkpoint.checkpointId === head.headCheckpointId
+        ? await journalUnread(t, head.headSeq, restored.checkpoint)
+        : 0;
+    if (unread > 0) {
+      warnings.push({
+        code: 'W_NEXUS_VAULT_JOURNAL_UNREAD',
+        message: `the change journal holds ${unread} segment(s) past snapshot ${restored.checkpoint.checkpointId}, which this store does not have: ${JOURNAL_UNREAD_REMEDY}`,
+      });
+    }
     warnings.push(...conn.state.drainWarnings());
     return {
       apiUrl: conn.apiUrl,
@@ -4365,16 +4422,33 @@ async function verifyNexusVaultImpl(
       return { ...r, cloudRows, match: (r.localRows ?? 0) === cloudRows };
     });
   }
+  // On a journal head, tables only snapshots carry (files, other databases, the
+  // row-identity-exempt tables) hold machine-local state: listed, never the verdict (T13867).
+  const journalHead = isJournalSnapshot(headCp);
+  const snapshotOnly = new Set(Object.keys(ROW_IDENTITY_EXEMPT[tableScopeOf(t)]));
+  const decides = (r: CloudVaultTableDiff): boolean =>
+    !journalHead || !(isSnapshotAuthoritativeKey(r.table) || snapshotOnly.has(r.table));
+  const asideDiffer = tables.filter((r) => !r.match && !decides(r)).map((r) => r.table);
+  if (asideDiffer.length > 0) {
+    warnings.push({
+      code: 'W_NEXUS_VAULT_SNAPSHOT_ONLY',
+      message: `${asideDiffer.length} table(s) only snapshots carry differ from the newest snapshot (machine-local, or not checkpointed yet), so they do not decide the verdict: ${asideDiffer.slice(0, 8).join(', ')}`,
+    });
+  }
+  const unread = await journalUnread(t, head.headSeq, headCp);
   let verdict: CloudVerifyResult['verdict'];
   if (headUntrusted) verdict = 'untrusted';
   else if (!headCp) verdict = 'empty';
-  else if (tables.every((x) => x.match)) verdict = 'match';
+  else if (unread > 0) verdict = 'behind';
+  else if (tables.filter(decides).every((x) => x.match)) verdict = 'match';
   else if (journalSide !== null) {
     const { localChanged, cloudAdvanced } = journalSide;
     verdict = localChanged && cloudAdvanced ? 'diverged' : cloudAdvanced ? 'behind' : 'ahead';
   } else {
     const localChanged = last
-      ? compareWithSynced(local, last.manifest, git, fileChanges).some((d) => !d.match)
+      ? compareWithSynced(local, last.manifest, git, fileChanges).some(
+          (d) => !d.match && decides(d),
+        )
       : true;
     const cloudAdvanced = synced?.lastCheckpointId !== headCp.checkpointId;
     verdict = localChanged && cloudAdvanced ? 'diverged' : cloudAdvanced ? 'behind' : 'ahead';
@@ -4433,7 +4507,6 @@ async function verifyNexusVaultImpl(
   // journal writes takes no push (T13034).
   const headFormat = formatWarning(t, headCp);
   if (headFormat) warnings.push(headFormat);
-  const journalHead = isJournalSnapshot(headCp);
   const remedy =
     deepFailure !== null
       ? `the cloud copy failed the deep check (${deepFailure}): do not pull it; see which device pushed the newest snapshot with \`cleo cloud vault\` and \`cleo cloud activity\`, then push a fresh snapshot from a machine whose store passes \`cleo cloud verify\``
@@ -4442,7 +4515,9 @@ async function verifyNexusVaultImpl(
         : headCp && vaultManifestFormat(headCp.manifest, hashKeyOf(t.dataKey)) === 'unknown'
           ? `the newest snapshot records a vault manifest format this CLEO does not recognise (this CLEO hashes format ${VAULT_MANIFEST_FORMAT_VERSION}): if a newer CLEO pushed it, upgrade CLEO on this machine, then verify again`
           : verdict === 'behind'
-            ? 'run `cleo cloud pull` to bring this machine to the newest snapshot'
+            ? unread > 0
+              ? `the change journal holds ${unread} segment(s) past the newest snapshot, which this store reads snapshots only and does not have: ${JOURNAL_UNREAD_REMEDY}`
+              : 'run `cleo cloud pull` to bring this machine to the newest snapshot'
             : verdict === 'ahead'
               ? journalHead
                 ? `this machine changed since the newest snapshot, which is checkpoint/v3 (the change journal writes this stream): ${JOURNAL_STREAM_REMEDY}`