Conversation sync done properly: fork-shaped seeded import adopted by the session controller, paged transfer, replicas of remote chats for offline continuation, move/divergence ledger with queued notices delivered on reconnect, native sidebar menu item + row markers, toasts

lib/session-sync.mjs

@@ -0,0 +1,177 @@
+// Conversation sync between devices, modelled on git:
+//
+//  - A session log is an append-only list of events with dense seqs from 0. The "commit id" of a
+//    log at seq N is sha256 over (N, event N). Because logs never rewrite, two copies with the
+//    same commit id at N share the whole prefix 0..N.
+//  - Continuing a session on another device = a fork: the target gets the prefix through the
+//    last completed turn (the "base", like a branch point) as a seeded session of its own, and
+//    the source records "moved to <device>/<session> at <base>".
+//  - If the source gets new events after the base (someone kept chatting there, e.g. while the
+//    devices could not see each other), the two branches have diverged → a conflict is raised
+//    on both sides. Nothing is merged or deleted automatically; both branches stay usable.
+//  - Each device keeps replicas ("mirrors") of recent remote sessions, so a session can be
+//    continued here even while its device is offline. The move notice is queued and delivered
+//    the moment that device is reachable again.
+import crypto from 'node:crypto'
+import fs from 'node:fs'
+import path from 'node:path'
+import zlib from 'node:zlib'
+
+/** JSON bytes per transferred page; keeps every link frame far below the 64 MiB limit. */
+export const PAGE_BYTES = 4 * 1024 * 1024
+
+/** Commit id of a log at `seq` (the event there must exist). */
+export function commitId(events, seq) {
+  const ev = events[seq]
+  if (!ev || ev.seq !== seq) return null
+  return crypto.createHash('sha256').update(`${seq}\n`).update(JSON.stringify(ev)).digest('hex').slice(0, 20)
+}
+
+/** Last seq of a log, -1 when empty. */
+export const tipSeq = (events) => (events.length ? events[events.length - 1].seq : -1)
+
+/**
+ * Inclusive seq of the latest completed-turn boundary (same rule as DSH's own fork): the last
+ * `turn/end` plus any trailing events up to the next turn/user message.
+ */
+export function completedBoundary(events) {
+  let lastEnd = -1
+  for (let i = events.length - 1; i >= 0; i--) if (events[i].type === 'turn/end') { lastEnd = i; break }
+  if (lastEnd < 0) return undefined
+  let boundary = events[lastEnd].seq
+  for (const next of events.slice(lastEnd + 1)) {
+    if (next.type === 'turn/start' || (next.type === 'user/message' && next.surfaceOp === 'append') || next.type === 'agent/inbox/spliced') break
+    boundary = next.seq
+  }
+  return boundary
+}
+
+/**
+ * Fork seed through `boundary` (inclusive), terminated by the inherited-cut marker exactly as
+ * `buildForkSeed` does for a completed boundary (no open turns to close there).
+ */
+export function forkSeed(events, boundary) {
+  const prefix = events.slice(0, boundary + 1)
+  prefix.push({ type: 'session/end-seed', seq: boundary + 1, time: events[boundary].time, data: { inherited: true } })
+  return prefix
+}
+
+/** Validate a transferred log: dense seqs from 0. */
+export function assertDense(events) {
+  for (let i = 0; i < events.length; i++) {
+    if (!events[i] || events[i].seq !== i) throw new Error(`session log is not contiguous at index ${i}`)
+  }
+}
+
+/** One page of events starting at `fromSeq`, bounded by PAGE_BYTES (always ≥ 1 event). */
+export function page(events, fromSeq, budget = PAGE_BYTES) {
+  const out = []
+  let bytes = 0
+  for (let i = Math.max(0, fromSeq); i < events.length; i++) {
+    const n = JSON.stringify(events[i]).length
+    if (out.length && bytes + n > budget) break
+    out.push(events[i]); bytes += n
+  }
+  const next = fromSeq + out.length
+  return { events: out, nextSeq: next, done: next >= events.length }
+}
+
+/**
+ * Compare a session's current log with the point it was moved away at.
+ * @returns 'clean' (no new events after the base), 'diverged' (new events after the base) or
+ *   'rewritten' (the base event itself differs — should never happen with append-only logs).
+ */
+export function divergence(events, atSeq, atCommit) {
+  if (atSeq >= 0 && atCommit && commitId(events, atSeq) !== atCommit) return 'rewritten'
+  return tipSeq(events) > atSeq ? 'diverged' : 'clean'
+}
+
+/** Short, user-facing title without newlines. */
+export const shortTitle = (t, fallback = 'session') => String(t || fallback).replace(/\s+/g, ' ').trim().slice(0, 120)
+
+// ------------------------------------------------------------------ replicas on disk
+/** Replicas of remote sessions: <dir>/<device>/<sessionId>.json (meta) + .jsonl.gz (events). */
+export class MirrorStore {
+  constructor(dir) { this.dir = dir }
+  _base(device, sessionId) {
+    const safe = (s) => String(s).replace(/[^A-Za-z0-9._-]/g, '_')
+    return path.join(this.dir, safe(device), safe(sessionId))
+  }
+  meta(device, sessionId) {
+    try { return JSON.parse(fs.readFileSync(`${this._base(device, sessionId)}.json`, 'utf8')) } catch { return null }
+  }
+  /** Device ids that have replicas. */
+  devices() {
+    try { return fs.readdirSync(this.dir, { withFileTypes: true }).filter((d) => d.isDirectory()).map((d) => d.name) } catch { return [] }
+  }
+  list(device) {
+    const d = path.join(this.dir, String(device).replace(/[^A-Za-z0-9._-]/g, '_'))
+    let names = []
+    try { names = fs.readdirSync(d).filter((n) => n.endsWith('.json')) } catch { return [] }
+    return names.map((n) => { try { return JSON.parse(fs.readFileSync(path.join(d, n), 'utf8')) } catch { return null } }).filter(Boolean)
+  }
+  events(device, sessionId) {
+    try {
+      const raw = zlib.gunzipSync(fs.readFileSync(`${this._base(device, sessionId)}.jsonl.gz`)).toString('utf8')
+      return raw ? raw.split('\n').filter(Boolean).map((l) => JSON.parse(l)) : []
+    } catch { return null }
+  }
+  /** Replace a replica atomically. */
+  save(device, sessionId, events, info = {}) {
+    const base = this._base(device, sessionId)
+    fs.mkdirSync(path.dirname(base), { recursive: true, mode: 0o700 })
+    const body = zlib.gzipSync(Buffer.from(events.map((e) => JSON.stringify(e)).join('\n'), 'utf8'))
+    const tmp = `${base}.${process.pid}.tmp`
+    fs.writeFileSync(tmp, body, { mode: 0o600 }); fs.renameSync(tmp, `${base}.jsonl.gz`)
+    const meta = {
+      device, sessionId, ...info,
+      tipSeq: tipSeq(events), tipCommit: events.length ? commitId(events, tipSeq(events)) : null,
+      syncedAt: Date.now(),
+    }
+    fs.writeFileSync(`${tmp}.m`, JSON.stringify(meta, null, 2), { mode: 0o600 }); fs.renameSync(`${tmp}.m`, `${base}.json`)
+    return meta
+  }
+  remove(device, sessionId) {
+    const base = this._base(device, sessionId)
+    for (const f of [`${base}.json`, `${base}.jsonl.gz`]) { try { fs.unlinkSync(f) } catch { /* gone */ } }
+  }
+  /** Keep only the `keep` most recently updated replicas of a device. */
+  prune(device, keep) {
+    const all = this.list(device).sort((a, b) => (b.updatedAt || 0) - (a.updatedAt || 0))
+    for (const m of all.slice(keep)) this.remove(device, m.sessionId)
+  }
+}
+
+// ------------------------------------------------------------------ lineage ledger
+/**
+ * Per-device ledger, persisted in the plugin state:
+ *   lineage[sessionId] = {
+ *     origin?: { device, sessionId, baseSeq, baseCommit, title, importedAt }   // this copy was continued from there
+ *     moved?:  { device, sessionId, atSeq, atCommit, at, by }                  // this session continues there
+ *     conflict?: { kind: 'diverged'|'rewritten', localTip, atSeq, detectedAt, other: {device, sessionId} }
+ *   }
+ *   outbox[key] = notice for a device that was unreachable  (key = `${notice.to}:${notice.sessionId}`)
+ */
+export class Ledger {
+  constructor(store) { this.store = store }
+  get data() { return (this.store.data.sessionSync ||= { lineage: {}, outbox: {} }) }
+  all() { return this.data.lineage }
+  get(sessionId) { return this.data.lineage[sessionId] || null }
+  patch(sessionId, patch) {
+    const cur = this.data.lineage[sessionId] || {}
+    const next = { ...cur, ...patch }
+    for (const k of Object.keys(next)) if (next[k] === undefined) delete next[k]
+    this.data.lineage[sessionId] = next
+    this.store.save()
+    return next
+  }
+  queue(notice) {
+    this.data.outbox[`${notice.to}:${notice.sessionId}`] = { ...notice, queuedAt: Date.now() }
+    this.store.save()
+  }
+  pending() { return Object.values(this.data.outbox) }
+  done(notice) {
+    delete this.data.outbox[`${notice.to}:${notice.sessionId}`]
+    this.store.save()
+  }
+}