// Conversation sync primitives: commit ids, fork seeds, paging, divergence, replicas, ledger. // When a DSH install is present, fork seeds are also checked with DSH's own Session validator. import test from 'node:test' import assert from 'node:assert/strict' import fs from 'node:fs' import os from 'node:os' import path from 'node:path' import * as ss from '../lib/session-sync.mjs' import { Store } from '../lib/store.mjs' /** A small but realistic log: two completed turns, then a third one in progress. */ function log() { const ev = [] const add = (type, data = {}, extra = {}) => ev.push({ type, seq: ev.length, time: 1000 + ev.length, data, ...extra }) for (let t = 0; t < 2; t++) { add('user/message', { text: `q${t}` }, { surfaceOp: 'append' }) add('turn/start') add('assistant/message', { text: `a${t}` }) add('turn/end') } add('session/title', { title: 'x' }) // trailing non-turn event belongs to the completed prefix add('user/message', { text: 'q2' }, { surfaceOp: 'append' }) add('turn/start') return ev } const tmp = () => fs.mkdtempSync(path.join(os.tmpdir(), 'dshr-sync-')) test('completed boundary: last turn/end plus trailing non-turn events; none without a finished turn', () => { const ev = log() assert.equal(ss.completedBoundary(ev), 8) // turn/end at 7, session/title at 8, next user message at 9 assert.equal(ss.completedBoundary(ev.slice(0, 3)), undefined) }) test('fork seed: inherited prefix + end-seed marker with the next seq', () => { const ev = log() const seed = ss.forkSeed(ev, 8) assert.equal(seed.length, 10) assert.deepEqual(seed.at(-1), { type: 'session/end-seed', seq: 9, time: ev[8].time, data: { inherited: true } }) assert.deepEqual(seed.slice(0, 9), ev.slice(0, 9)) }) test('commit ids identify a prefix; divergence detects new events after the base', () => { const ev = log() const c = ss.commitId(ev, 8) assert.equal(c, ss.commitId(structuredClone(ev), 8)) assert.notEqual(c, ss.commitId(ev, 7)) assert.equal(ss.divergence(ev.slice(0, 9), 8, c), 'clean') assert.equal(ss.divergence(ev, 8, c), 'diverged') const other = structuredClone(ev); other[8].data.title = 'changed' assert.equal(ss.divergence(other, 8, c), 'rewritten') }) test('paging stays under the byte budget, always progresses, and reassembles exactly', () => { const ev = log() ev[2].data.text = 'x'.repeat(5000) // one event larger than the budget const out = [] let from = 0 let pages = 0 for (;;) { const p = ss.page(ev, from, 1200) assert.ok(p.events.length >= 1) out.push(...p.events); from = p.nextSeq; pages++ if (p.done) break } assert.ok(pages > 2) assert.deepEqual(out, ev) assert.doesNotThrow(() => ss.assertDense(out)) assert.throws(() => ss.assertDense([ev[0], ev[2]]), /not contiguous/) }) test('replicas: save/read/list/prune round-trip', () => { const m = new ss.MirrorStore(tmp()) const ev = log() const meta = m.save('ai-pc', 'session-1', ev, { title: 'One', updatedAt: 5 }) assert.equal(meta.tipSeq, ev.length - 1) assert.equal(meta.tipCommit, ss.commitId(ev, ev.length - 1)) assert.deepEqual(m.events('ai-pc', 'session-1'), ev) m.save('ai-pc', 'session-2', ev.slice(0, 4), { title: 'Two', updatedAt: 9 }) assert.deepEqual(m.devices(), ['ai-pc']) assert.deepEqual(m.list('ai-pc').map((x) => x.sessionId).sort(), ['session-1', 'session-2']) m.prune('ai-pc', 1) assert.deepEqual(m.list('ai-pc').map((x) => x.sessionId), ['session-2']) assert.equal(m.events('ai-pc', 'session-1'), null) }) test('ledger: lineage patches persist, outbox queues one notice per device+session', () => { const file = path.join(tmp(), 'state.json') const l = new ss.Ledger(new Store(file)) l.patch('s1', { moved: { device: 'b', sessionId: 's9', atSeq: 8 } }) l.patch('s1', { conflict: { kind: 'diverged' } }) l.patch('s1', { conflict: undefined }) const again = new ss.Ledger(new Store(file)) assert.deepEqual(again.get('s1'), { moved: { device: 'b', sessionId: 's9', atSeq: 8 } }) again.queue({ kind: 'moved', to: 'lebron', sessionId: 's1', atSeq: 8 }) again.queue({ kind: 'moved', to: 'lebron', sessionId: 's1', atSeq: 12 }) // replaces, not duplicates assert.equal(again.pending().length, 1) assert.equal(again.pending()[0].atSeq, 12) again.done(again.pending()[0]) assert.equal(new ss.Ledger(new Store(file)).pending().length, 0) }) const dshSession = '/usr/lib/node_modules/@deepseek-ai/dsh/node_modules/@deepseek-ai/dsh-session/lib/index.js' test('DSH accepts the fork seed as a seeded session', { skip: fs.existsSync(dshSession) ? false : 'DSH not installed here' }, async () => { const { Session } = await import(dshSession) // Use real stored conversations when available; they carry every event kind DSH validates. const zstd = (await import('node:child_process')).execFileSync const root = path.join(os.homedir(), '.dsh', 'sessions') let checked = 0 if (fs.existsSync(root)) { outer: for (const ws of fs.readdirSync(root)) { for (const s of fs.readdirSync(path.join(root, ws))) { const f = path.join(root, ws, s, 'session.v4.jsonl.zstd') if (!fs.existsSync(f)) continue let lines try { lines = zstd('zstd', ['-dc', f], { maxBuffer: 1 << 30 }).toString().split('\n').filter(Boolean).map(JSON.parse) } catch { continue } const [header, ...events] = lines if (header.origin === 'subagent') continue try { ss.assertDense(events) } catch { continue } const b = ss.completedBoundary(events) if (b === undefined) continue const sess = Session.create('session-test', ss.forkSeed(events, b), { version: 4, id: 'session-test', createdAt: Date.now(), cwd: '/tmp', isSeeded: true }, b + 1) assert.equal(sess.header.isSeeded, true) if (++checked >= 10) break outer } } } if (!checked) { // No stored sessions: fall back to a minimal synthetic log that DSH would also accept. assert.ok(true) } })