feat(sync): a restored store joins a journaled stream; two-device sync end to end (T13312)

.changeset/t13312-journal-join.md

@@ -0,0 +1,30 @@
+---
+id: t13312-journal-join
+tasks: [T13312, T12999, T12996]
+kind: feat
+summary: "`cleo sync enable push` joins a stream another device journaled, from the journal checkpoint this store restored"
+---
+
+A second device can now enter a stream whose change journal another device already started (T13312, T12999). Before this, a
+store restored from the journal checkpoint carried no row meta and no join state, so `cleo sync enable push` refused it and
+there was no way in.
+
+- **Restore keeps the journal's row state.** On a journal checkpoint, the restore keeps the snapshot's `_sync_row_meta` and
+  `_sync_field_leave` rows instead of clearing them as machine-local.
+- **The join.** On a stream whose head is a journal checkpoint this store restored, `cleo sync enable push` returns status
+  `joined`:
+  - the store's cut is placed at its current capture position;
+  - the ledgers are baselined;
+  - push, pull and undo are turned on;
+  - pulling resumes right after the restored checkpoint.
+  No second genesis is pushed.
+- **A changed store is refused, never folded.** The join is refused with `E_NEXUS_SYNC_REFUSED` when:
+  - any table's rows differ from the checkpoint's manifest;
+  - any row no longer matches its restored row meta;
+  - a capture is still waiting to seal;
+  - a write landed after the check, which is caught inside the cut's own transaction, so it is never baselined as
+    checkpoint state.
+  The remedy is to restore the checkpoint again (`cleo cloud restore --force`), then join. A refused join changes nothing: capture is left as it was found, and sealing stays off.
+- **A store that never restored the journal checkpoint** gets `E_NEXUS_SYNC_STREAM_JOURNALED`, naming the restore-then-join
+  path. The pull refusal for a store behind the journal genesis (T13306) names the same path.
+- `sync.push` and `sync.pull` stay unreleased.

packages/contracts/src/nexus-vault.ts

@@ -218,9 +218,11 @@ export interface CloudSyncPushEnableResult {
   streamId: string;
   /**
    * `enabled`: the cut was recorded and its genesis checkpoint stored; `resumed`: a cut recorded
-   * by an earlier run whose checkpoint push failed was pushed now; `already`: push was already on.
+   * by an earlier run whose checkpoint push failed was pushed now; `already`: push was already on;
+   * `joined`: this store holds the stream's journal checkpoint another device cut, and now pushes its
+   * own writes and pulls the stream's after it, with no genesis of its own (T13312).
    */
-  status: 'enabled' | 'resumed' | 'already';
+  status: 'enabled' | 'resumed' | 'already' | 'joined';
   /** The genesis cut: the highest capture seq the checkpoint carries. */
   cut: number;
   /** Transactions sealed before the cut, folded into the checkpoint, and rows given genesis row meta. */

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

@@ -68,6 +68,7 @@ import {
 import { setCaptureEnabled } from '../../store/sync/capture.js';
 import { isSyncFlagOn, setSyncFlag } from '../../store/sync/flags.js';
 import { cutGenesis, genesisCutOf, genesisPending } from '../../store/sync/genesis.js';
+import { readStreamCursor } from '../../store/sync/pull.js';
 import { ensureProjectReplica, storeHwm } from '../../store/sync/replica.js';
 import { readDeviceRegistry } from '../../store/sync/replica-registry.js';
 import { ensureSyncSchema } from '../../store/sync/schema.js';
@@ -5241,9 +5242,15 @@ describe('cloud project link stores a new project key with its registration (onb
 describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => {
   const SYNC_JOURNAL = path.resolve(import.meta.dirname, '../../../migrations/sync-journal');
 
-  /** A machine whose project store is a real cleo.db with capture and seal on, bound and linked. */
-  async function journalMachine(): Promise<{ m: Machine; dbPath: string }> {
-    const m = await machine('a', DEVICE_A, REPLICA_A);
+  /**
+   * A machine whose project store is a real cleo.db with capture and seal on, bound and linked.
+   * Device A by default; device B attaches to the project A registered.
+   */
+  async function journalMachine(device: 'a' | 'b' = 'a'): Promise<{ m: Machine; dbPath: string }> {
+    const m =
+      device === 'a'
+        ? await machine('a', DEVICE_A, REPLICA_A)
+        : await machine('b', DEVICE_B, REPLICA_B);
     const cleo = path.join(m.root, '.cleo');
     fs.mkdirSync(cleo, { recursive: true });
     fs.writeFileSync(path.join(cleo, 'project-id'), `${LOCAL_PROJECT}\n`);
@@ -5267,14 +5274,34 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => {
       return ensureProjectReplica(db, { dbPath, mode: 'live' }).replicaId;
     });
     m.replicaId = replicaId;
-    fake.addProject(REMOTE_PROJECT, { [replicaId]: DEVICE_A });
+    if (device === 'a') fake.addProject(REMOTE_PROJECT, { [replicaId]: DEVICE_A });
+    else fake.replicas.get(REMOTE_PROJECT)?.set(replicaId, DEVICE_B);
     link(m);
     return { m, dbPath };
   }
 
   const storeOf = async (dbPath: string) =>
     getDualScopeNativeDb(await openDualScopeDbAtPath('project', dbPath));
 
+  /**
+   * Device B with A's journal checkpoint restored, attached under its own bound
+   * replica, and not yet joined.
+   */
+  async function restoredJoiner(): Promise<{ b: Machine; bDb: string }> {
+    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 () => ensureProjectReplica(await storeOf(bDb), { dbPath: bDb, mode: 'live' }).replicaId,
+    );
+    b.replicaId = bound;
+    fake.replicas.get(REMOTE_PROJECT)?.set(bound, DEVICE_B);
+    link(b);
+    return { b, bDb };
+  }
+
   it('cuts the store and pushes a checkpoint/v3 genesis the server accepts; push turns on and the vault steps aside', async () => {
     const { m, dbPath } = await journalMachine();
     const r = await on(m, () => enableSyncPush(vopts(m, { allowUnreleased: true })));
@@ -5305,13 +5332,22 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => {
     const vault = await failure(on(m, () => pushNexusVault(vopts(m))));
     expect(vault.code).toBe('E_NEXUS_VAULT_STREAM_UPGRADED');
     expect(s.checkpoints).toHaveLength(1);
-    // A store that never cut this stream (its cut forgotten) meets a journal head: it joins by
-    // pulling, never with a second genesis.
+    // A store that holds the journal checkpoint but forgot its cut joins it, never with a
+    // second genesis (T13312).
     await on(m, async () => {
       (await storeOf(dbPath)).exec("DELETE FROM _sync_meta WHERE key LIKE 'genesis_%'");
     });
-    const second = await failure(on(m, () => enableSyncPush(vopts(m, { allowUnreleased: true }))));
+    expect((await on(m, () => enableSyncPush(vopts(m, { allowUnreleased: true })))).status).toBe(
+      'joined',
+    );
+    expect(s.checkpoints).toHaveLength(1);
+    // A store that never restored that checkpoint is told to restore it, then join.
+    const { m: other } = await journalMachine('b');
+    const second = await failure(
+      on(other, () => enableSyncPush(vopts(other, { allowUnreleased: true }))),
+    );
     expect(second.code).toBe('E_NEXUS_SYNC_STREAM_JOURNALED');
+    expect(second.fix).toContain('cleo cloud restore');
     expect(s.checkpoints).toHaveLength(1);
   });
 
@@ -5652,6 +5688,179 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => {
         0,
       );
     });
+    // The remedy works: B restores the journal checkpoint, joins, and holds A's folded change.
+    const restored = await on(b, () => restoreNexusVault(vopts(b, { mode: 'pull', force: true })));
+    expect(restored.status).toBe('restored');
+    // The placed store is a new instance bound to a new replica (T13109); `cleo cloud pull`
+    // relinks it (the CLI's relink), which this machine helper does by hand.
+    const rebound = restored.replica?.current;
+    expect(rebound).toBeDefined();
+    b.replicaId = rebound ?? '';
+    fake.replicas.get(REMOTE_PROJECT)?.set(b.replicaId, DEVICE_B);
+    link(b);
+    expect((await on(b, () => enableSyncPush(vopts(b, { allowUnreleased: true })))).status).toBe(
+      'joined',
+    );
+    expect(
+      await on(b, async () =>
+        (await storeOf(bDb)).prepare("SELECT title FROM tasks_tasks WHERE id = 'G1'").get(),
+      ),
+    ).toEqual({ title: 'title G1' });
+    expect((await on(b, () => pullSyncStream(vopts(b)))).refused).toBeNull();
+  });
+
+  it("two devices (T13312, T12996): B restores A's genesis and joins; cloud sync carries each device's writes to the other and converges a concurrent status change", async () => {
+    const { m: a, dbPath: aDb } = await journalMachine();
+    await on(a, () => enableSyncPush(vopts(a, { allowUnreleased: true })));
+    await on(a, async () => {
+      setSyncFlag(await storeOf(aDb), 'sync.pull', true, { allowUnreleased: true });
+    });
+    const { b, bDb } = await restoredJoiner();
+    const joined = await on(b, () => enableSyncPush(vopts(b, { allowUnreleased: true })));
+    expect(joined.status).toBe('joined');
+    const journalCps = fake
+      .stream(STREAM)
+      .checkpoints.filter((c) => manifestVersion(c.manifest) === 3);
+    expect(journalCps).toHaveLength(1);
+    // B pulls on from the checkpoint it restored.
+    const genesisCp = journalCps[0];
+    if (!genesisCp) throw new Error('no journal checkpoint');
+    expect(await on(b, async () => readStreamCursor(await storeOf(bDb), STREAM))).toEqual(
+      cursorFromCheckpoint(genesisCp),
+    );
+
+    const sync = async (m: Machine) => {
+      const r = await on(m, () => cloudSync(vopts(m, { scope: 'project', allowUnreleased: true })));
+      expect(r.streams[0]?.refused ?? null, `${m.name}: ${r.streams[0]?.refused}`).toBeNull();
+      return r.streams[0];
+    };
+    const value = async (m: Machine, dbFile: string, sql: string) =>
+      on(m, async () => (await storeOf(dbFile)).prepare(sql).get());
+
+    // Each device edits; both sync; each sees the other's write.
+    await on(a, async () => {
+      (await storeOf(aDb)).exec("UPDATE tasks_tasks SET title = 'from A' WHERE id = 'T1'");
+    });
+    await on(b, async () => {
+      (await storeOf(bDb)).exec(
+        "INSERT INTO tasks_tasks (id, title, type, status, priority, uid, birth_fp) VALUES ('T9', 'from B', 'task', 'pending', 'medium', 'uid-T9', 'fp-T9')",
+      );
+    });
+    await sync(a);
+    await sync(b);
+    await sync(a);
+    expect(await value(b, bDb, "SELECT title FROM tasks_tasks WHERE id = 'T1'")).toEqual({
+      title: 'from A',
+    });
+    expect(await value(a, aDb, "SELECT title FROM tasks_tasks WHERE id = 'T9'")).toEqual({
+      title: 'from B',
+    });
+
+    // A concurrent status change of one task: the typed rules decide it the same way on both.
+    await on(a, async () => {
+      (await storeOf(aDb)).exec(
+        "UPDATE tasks_tasks SET status = 'done', pipeline_stage = 'contribution', completed_at = '2026-10-07T00:00:00.000Z' WHERE id = 'T2'",
+      );
+    });
+    await on(b, async () => {
+      (await storeOf(bDb)).exec(
+        "UPDATE tasks_tasks SET status = 'cancelled', pipeline_stage = 'cancelled', cancelled_at = '2026-10-07T00:00:01.000Z', cancellation_reason = 'dup' WHERE id = 'T2'",
+      );
+    });
+    await sync(a);
+    await sync(b);
+    await sync(a);
+    await sync(b);
+    const statusSql =
+      "SELECT status, pipeline_stage, completed_at, cancelled_at, cancellation_reason FROM tasks_tasks WHERE id = 'T2'";
+    const onA = await value(a, aDb, statusSql);
+    const onB = await value(b, bDb, statusSql);
+    expect(onB).toEqual(onA);
+    // Both are absorbing, so the later write (B's cancel) wins its LWW; the
+    // stamps move with it as one group and the stage follows the status.
+    expect(onA).toEqual({
+      status: 'cancelled',
+      pipeline_stage: 'cancelled',
+      completed_at: null,
+      cancelled_at: '2026-10-07T00:00:01.000Z',
+      cancellation_reason: 'dup',
+    });
+  });
+
+  it('a joined store with push off syncs pull-only: synced, the push leg skipped (T13312)', async () => {
+    const { m: a, dbPath: aDb } = await journalMachine();
+    await on(a, () => enableSyncPush(vopts(a, { allowUnreleased: true })));
+    const { b, bDb } = await restoredJoiner();
+    expect((await on(b, () => enableSyncPush(vopts(b, { allowUnreleased: true })))).status).toBe(
+      'joined',
+    );
+    await on(b, async () => {
+      setSyncFlag(await storeOf(bDb), 'sync.push', false, { allowUnreleased: true });
+    });
+    await on(a, async () => {
+      (await storeOf(aDb)).exec("UPDATE tasks_tasks SET title = 'from A' WHERE id = 'T1'");
+    });
+    await on(a, () => pushSyncStream(vopts(a, { allowUnreleased: true })));
+    const r = await on(b, () => cloudSync(vopts(b, { scope: 'project', allowUnreleased: true })));
+    expect(r.streams[0]).toMatchObject({ status: 'synced', skipped: ['sync.push is off'] });
+    expect(
+      await on(b, async () =>
+        (await storeOf(bDb)).prepare("SELECT title FROM tasks_tasks WHERE id = 'T1'").get(),
+      ),
+    ).toEqual({ title: 'from A' });
+  });
+
+  it('a join is refused, never folded, when the store changed since its restore (T13312)', async () => {
+    const { m: a } = await journalMachine();
+    await on(a, () => enableSyncPush(vopts(a, { allowUnreleased: true })));
+    const { b, bDb } = await restoredJoiner();
+    await on(b, async () => {
+      (await storeOf(bDb)).exec("UPDATE tasks_tasks SET title = 'edited on B' WHERE id = 'T1'");
+    });
+    const err = await failure(on(b, () => enableSyncPush(vopts(b, { allowUnreleased: true }))));
+    expect(err.code).toBe('E_NEXUS_SYNC_REFUSED');
+    expect(err.message).toContain('tasks_tasks');
+    expect(err.fix).toContain('cleo cloud restore --force');
+    await on(b, async () => {
+      const db = await storeOf(bDb);
+      expect(genesisCutOf(db, STREAM)).toBeUndefined();
+      expect(isSyncFlagOn(db, 'sync.push', {})).toBe(false);
+    });
+    expect(fake.stream(STREAM).segments).toHaveLength(0);
+  });
+
+  it('a join is refused, never sealed past, while a capture waits to seal (T13312)', async () => {
+    const { m: a } = await journalMachine();
+    await on(a, () => enableSyncPush(vopts(a, { allowUnreleased: true })));
+    const { b, bDb } = await restoredJoiner();
+    // A write captured but not sealed: joining would seal it as if the checkpoint authored it.
+    await on(b, async () => {
+      (await storeOf(bDb)).exec(
+        "INSERT INTO _sync_capture (tbl, op, rk, uid, img, at_ms) VALUES ('tasks_tasks', 'U', 'T1', 'uid-T1', '{}', 0)",
+      );
+    });
+    const err = await failure(on(b, () => enableSyncPush(vopts(b, { allowUnreleased: true }))));
+    expect(err.code).toBe('E_NEXUS_SYNC_REFUSED');
+    expect(err.message).toContain('captures still waiting to seal');
+  });
+
+  it('a join is refused when the restored row meta does not match the rows (T13312)', async () => {
+    const { m: a } = await journalMachine();
+    await on(a, () => enableSyncPush(vopts(a, { allowUnreleased: true })));
+    const { b, bDb } = await restoredJoiner();
+    await on(b, async () => {
+      (await storeOf(bDb)).exec(
+        "DELETE FROM _sync_row_meta WHERE tbl = 'tasks_tasks' AND uid = (SELECT uid FROM _sync_row_meta WHERE tbl = 'tasks_tasks' ORDER BY uid LIMIT 1)",
+      );
+    });
+    const err = await failure(on(b, () => enableSyncPush(vopts(b, { allowUnreleased: true }))));
+    expect(err.code).toBe('E_NEXUS_SYNC_REFUSED');
+    expect(err.message).toContain('tasks_tasks');
+    await on(b, async () => {
+      const db = await storeOf(bDb);
+      expect(genesisCutOf(db, STREAM)).toBeUndefined();
+      expect(isSyncFlagOn(db, 'sync.seal', {})).toBe(false);
+    });
   });
 
   it('S4-2: a device clock ahead of the server pauses push', async () => {