fix(brain): compact sync log on installs with no brain-sync peers

server/services/brainSyncLog.js

@@ -237,9 +237,38 @@ export async function getChangesSince(sinceSeq, limit = 100) {
 }
 
 /**
- * Compact the log by dropping entries below minSeq
+ * Replay entries for a single (type, id) according to runtime LWW rules,
+ * returning the surviving terminal entry.
+ *
+ * Incumbent wins ties (same timestamp) matching applyRemoteRecord.
+ * Entries missing updatedAt are skipped.
+ */
+function replayTerminal(entries) {
+  let accepted = null;
+  for (const e of [...entries].sort((a, b) => a.seq - b.seq)) {
+    const ts = e?.record?.updatedAt;
+    if (ts == null) continue;
+    if (accepted == null || ts > accepted.record.updatedAt) accepted = e;
+  }
+  return accepted;
+}
+
+/**
+ * Compact the sync log using a compatibility-preserving compaction representation.
+ *
+ * Deltas at or above `minSeq` (unconsumed by at least one active peer) are
+ * preserved verbatim in sequence order.
+ *
+ * For history below `minSeq` (or all entries when minSeq is 0, e.g. on installs
+ * with no brain-sync peers), redundant intermediate updates are pruned by
+ * retaining only the surviving terminal LWW entry per (type, id).
+ *
+ * The durable sequence counter (maxSeq) is always determined from disk state
+ * under the log mutex, preventing index skew after a failed append, and is
+ * preserved so initSyncLog recovers the monotonic sequence counter across
+ * restarts.
  */
-export async function compactLog(minSeq) {
+export async function compactLog(minSeq = 0) {
   return withLock(async () => {
     await ensureBrainDir();
     // Load first: the rebuild below marks the index loaded, so skipping this
@@ -248,33 +277,126 @@ export async function compactLog(minSeq) {
     if (!existsSync(SYNC_LOG_FILE)) return 0;
 
     const content = await readFile(SYNC_LOG_FILE, 'utf-8');
-    const lines = content.trim().split('\n').filter(l => l.trim());
-    const kept = [];
-    let dropped = 0;
-
-    for (const line of lines) {
-      const entry = safeJSONParse(line, null);
-      if (!entry || entry.seq < minSeq) {
-        dropped++;
-        continue;
+    const rawLines = content.trim().split('\n').filter(l => l.trim());
+    if (rawLines.length === 0) return 0;
+
+    const parsedLines = [];
+    let maxDurableSeq = 0;
+    let maxSeqEntry = null;
+
+    for (const rawLine of rawLines) {
+      const entry = safeJSONParse(rawLine, null);
+      if (entry && typeof entry.seq === 'number') {
+        if (entry.seq > maxDurableSeq) {
+          maxDurableSeq = entry.seq;
+          maxSeqEntry = entry;
+        }
+        parsedLines.push({ rawLine, entry, seq: entry.seq });
+      } else {
+        // Line without numeric seq (e.g. malformed or unindexed note)
+        parsedLines.push({ rawLine, entry: null, seq: null });
+      }
+    }
+
+    const floor = typeof minSeq === 'number' && Number.isFinite(minSeq)
+      ? Math.max(0, Math.min(minSeq, maxDurableSeq))
+      : 0;
+
+    const preservedTail = [];
+    const tailKeys = new Set();
+    const olderEntriesByKey = new Map();
+    const unindexedOrUntypedOlder = [];
+
+    for (const item of parsedLines) {
+      const { entry, seq } = item;
+      if (seq !== null && floor > 0 && seq >= floor) {
+        preservedTail.push(item);
+        if (entry?.type && entry?.id) {
+          tailKeys.add(`${entry.type}/${entry.id}`);
+        }
+      } else if (entry?.type && entry?.id) {
+        const key = `${entry.type}/${entry.id}`;
+        if (!olderEntriesByKey.has(key)) olderEntriesByKey.set(key, []);
+        olderEntriesByKey.get(key).push(item);
+      } else {
+        if (floor === 0 || seq === null) {
+          unindexedOrUntypedOlder.push(item);
+        }
+      }
+    }
+
+    // Replay terminal winning state for older keys
+    const keptOlder = [];
+    const olderWinnersByKey = new Map();
+    for (const [key, items] of olderEntriesByKey) {
+      const entries = items.map(i => i.entry);
+      const olderWinner = replayTerminal(entries);
+      if (!olderWinner) continue;
+
+      // If this key also appears in the preserved tail, check whether any tail
+      // operation strictly supersedes the pre-floor LWW winner (updatedAt > olderWinner.updatedAt).
+      // If the tail carries ONLY stale/losing operations (e.g. olderWinner is a Jan-02 delete
+      // and tail has an echoed Jan-01 create), we MUST retain olderWinner before the verbatim
+      // tail so fresh / delta-only peers do not accept the stale create and resurrect the record.
+      if (tailKeys.has(key)) {
+        const tailItems = preservedTail.filter(i => i.entry && `${i.entry.type}/${i.entry.id}` === key);
+        const supersededByTail = tailItems.some(i => {
+          const tailTs = i.entry?.record?.updatedAt;
+          return tailTs != null && olderWinner.record?.updatedAt != null && tailTs > olderWinner.record.updatedAt;
+        });
+        if (supersededByTail) {
+          continue;
+        }
+      }
+
+      const matchingItem = items.find(i => i.entry === olderWinner)
+        || { rawLine: JSON.stringify(olderWinner), entry: olderWinner, seq: olderWinner.seq };
+      keptOlder.push(matchingItem);
+      olderWinnersByKey.set(key, matchingItem);
+    }
+
+    const kept = [...unindexedOrUntypedOlder, ...keptOlder, ...preservedTail];
+
+    // Ensure the durable max sequence is preserved so restart recovery and cursors hold
+    if (maxSeqEntry && !kept.some(i => i.seq === maxSeqEntry.seq)) {
+      if (!maxSeqEntry.type || !maxSeqEntry.id) {
+        kept.push({ rawLine: JSON.stringify(maxSeqEntry), entry: maxSeqEntry, seq: maxSeqEntry.seq });
+      } else {
+        const key = `${maxSeqEntry.type}/${maxSeqEntry.id}`;
+        const winner = olderWinnersByKey.get(key);
+        if (winner && winner.entry) {
+          winner.entry.seq = maxSeqEntry.seq;
+          winner.seq = maxSeqEntry.seq;
+          winner.rawLine = JSON.stringify(winner.entry);
+        } else {
+          kept.push({ rawLine: JSON.stringify(maxSeqEntry), entry: maxSeqEntry, seq: maxSeqEntry.seq });
+        }
       }
-      kept.push({ line, seq: entry.seq });
     }
 
-    const newContent = kept.length > 0 ? kept.map(k => k.line).join('\n') + '\n' : '';
+    // Sort kept entries: items with numeric seq sorted by seq
+    kept.sort((a, b) => {
+      if (a.seq !== null && b.seq !== null) return a.seq - b.seq;
+      return 0;
+    });
+
+    const dropped = rawLines.length - kept.length;
+    if (dropped <= 0) return 0;
+
+    const newContent = kept.map(i => i.rawLine).join('\n') + '\n';
     await atomicWrite(SYNC_LOG_FILE, newContent);
 
-    // Rebuild the index from what we just wrote — the offsets all moved. A kept
-    // line without a numeric seq keeps its bytes but stays OUT of the index:
-    // a non-numeric seq in the array would break firstIndexAfter's binary
-    // search and hide every entry before it from peers.
+    // Rebuild index offsets from what was written
     offsets = [];
     let offset = 0;
-    for (const { line, seq } of kept) {
-      if (typeof seq === 'number') offsets.push({ seq, offset });
-      offset += Buffer.byteLength(line, 'utf8') + 1;
+    for (const { rawLine, seq } of kept) {
+      if (typeof seq === 'number') {
+        offsets.push({ seq, offset });
+      }
+      offset += Buffer.byteLength(rawLine, 'utf8') + 1;
     }
     fileSize = offset;
+    currentSeq = maxDurableSeq;
     pendingNewline = false;
     indexLoaded = true;
 

server/services/brainSyncLog.test.js

@@ -20,7 +20,11 @@ function getTempRoot() {
 
 vi.mock('../lib/fileUtils.js', async () => {
   const actual = await vi.importActual('../lib/fileUtils.js');
-  return makePathsProxy(actual, { dataRoot: () => getTempRoot() });
+  const proxy = makePathsProxy(actual, { dataRoot: () => getTempRoot() });
+  return {
+    ...proxy,
+    atomicWrite: vi.fn(actual.atomicWrite)
+  };
 });
 
 // Spy on createReadStream so the tests can assert WHICH bytes were read — that
@@ -39,6 +43,7 @@ vi.mock('fs/promises', async () => {
 
 import { createReadStream } from 'fs';
 import { appendFile } from 'fs/promises';
+import { atomicWrite } from '../lib/fileUtils.js';
 import {
   initSyncLog,
   getCurrentSeq,
@@ -392,9 +397,11 @@ describe('brainSyncLog', () => {
     });
 
     it('serves reads from the rebuilt index immediately after compaction', async () => {
-      for (let i = 0; i < 5; i++) {
-        await appendChange('create', 'people', `p${i}`, { name: `Person ${i}` }, 'inst-1');
-      }
+      await appendChange('create', 'people', 'p0', { name: 'Person 0 v1' }, 'inst-1');
+      await appendChange('update', 'people', 'p0', { name: 'Person 0 v2' }, 'inst-1');
+      await appendChange('update', 'people', 'p0', { name: 'Person 0 v3' }, 'inst-1');
+      await appendChange('create', 'people', 'p1', { name: 'Person 1' }, 'inst-1');
+      await appendChange('create', 'people', 'p2', { name: 'Person 2' }, 'inst-1');
 
       expect(await compactLog(3)).toBe(2);
 
@@ -423,16 +430,142 @@ describe('brainSyncLog', () => {
     });
 
     it('keeps appending at the right offset after compaction', async () => {
-      for (let i = 0; i < 4; i++) {
-        await appendChange('create', 'people', `p${i}`, { name: `Person ${i}` }, 'inst-1');
-      }
+      await appendChange('create', 'people', 'p0', { name: 'Person 0 v1' }, 'inst-1');
+      await appendChange('update', 'people', 'p0', { name: 'Person 0 v2' }, 'inst-1');
+      await appendChange('create', 'people', 'p1', { name: 'Person 1' }, 'inst-1');
+      await appendChange('create', 'people', 'p2', { name: 'Person 2' }, 'inst-1');
       await compactLog(3);
-      await appendChange('create', 'people', 'p5', { name: 'Person 5' }, 'inst-1');
+      await appendChange('create', 'people', 'p3', { name: 'Person 3' }, 'inst-1');
 
       vi.clearAllMocks();
       const result = await getChangesSince(4);
       expect(lastReadStart()).toBe(lineOffset(2));
       expect(result.changes.map(c => c.seq)).toEqual([5]);
     });
+
+    it('prunes intermediate update churn to terminal survivors and preserves max seq on installs with no peers (#5439)', async () => {
+      await appendChange('create', 'people', 'p0', { name: 'Person 0 v1', updatedAt: '2026-01-01T00:00:00.000Z' }, 'inst-1');
+      await appendChange('update', 'people', 'p0', { name: 'Person 0 v2', updatedAt: '2026-01-02T00:00:00.000Z' }, 'inst-1');
+      await appendChange('update', 'people', 'p0', { name: 'Person 0 v3', updatedAt: '2026-01-03T00:00:00.000Z' }, 'inst-1');
+      await appendChange('create', 'people', 'p1', { name: 'Person 1', updatedAt: '2026-01-01T00:00:00.000Z' }, 'inst-1');
+      await appendChange('delete', 'people', 'p1', { updatedAt: '2026-01-04T00:00:00.000Z' }, 'inst-1');
+      await appendChange('create', 'people', 'p2', { name: 'Person 2', updatedAt: '2026-01-05T00:00:00.000Z' }, 'inst-1');
+      expect(getCurrentSeq()).toBe(6);
+
+      // Compact with floor 0 (compatibility-preserving compaction representation)
+      const dropped = await compactLog(0);
+      expect(dropped).toBe(3); // 2 intermediate p0 updates + 1 p1 create superseded by delete
+
+      // Verify file on disk contains the 3 terminal survivors
+      const lines = readFileSync(syncLogPath(), 'utf8').trim().split('\n');
+      expect(lines).toHaveLength(3);
+      const parsed = lines.map(l => JSON.parse(l));
+      expect(parsed.map(p => ({ id: p.id, op: p.op }))).toEqual([
+        { id: 'p0', op: 'update' },
+        { id: 'p1', op: 'delete' },
+        { id: 'p2', op: 'create' }
+      ]);
+      expect(parsed.map(p => p.seq)).toEqual([3, 5, 6]);
+
+      // All active entities are available for delta pulls (e.g. pre-#1077 / fresh peers)
+      const deltaResult = await getChangesSince(0);
+      expect(deltaResult.changes).toHaveLength(3);
+
+      // Simulate restart: re-initialize the log
+      await initSyncLog();
+      expect(getCurrentSeq()).toBe(6);
+
+      // Subsequent appends continue monotonic sequence numbers
+      const nextEntry = await appendChange('create', 'people', 'p3', { name: 'Person 3' }, 'inst-1');
+      expect(nextEntry.seq).toBe(7);
+      expect(getCurrentSeq()).toBe(7);
+    });
+
+    it('returns 0 and does not call atomicWrite when no entries are dropped (#5439)', async () => {
+      for (let i = 0; i < 3; i++) {
+        await appendChange('create', 'people', `p${i}`, { name: `Person ${i}`, updatedAt: `2026-01-0${i + 1}T00:00:00.000Z` }, 'inst-1');
+      }
+      atomicWrite.mockClear();
+
+      // Calling compactLog with minSeq <= 1 drops nothing
+      const dropped = await compactLog(1);
+      expect(dropped).toBe(0);
+      expect(atomicWrite).not.toHaveBeenCalled();
+    });
+
+    it('determines compaction floor strictly under mutex from durable state after a failed append', async () => {
+      for (let i = 0; i < 3; i++) {
+        await appendChange('create', 'people', `p${i}`, { name: `Person ${i}`, updatedAt: `2026-01-0${i + 1}T00:00:00.000Z` }, 'inst-1');
+      }
+      expect(getCurrentSeq()).toBe(3);
+
+      // Simulate appendFile disk failure during appendChange
+      appendFile.mockRejectedValueOnce(new Error('Disk I/O error'));
+      await expect(appendChange('create', 'people', 'p3', { name: 'Person 3' }, 'inst-1')).rejects.toThrow('Disk I/O error');
+
+      // In-memory currentSeq was reserved/incremented to 4, but disk holds up to seq 3
+      expect(getCurrentSeq()).toBe(4);
+
+      // Compacting under mutex re-syncs durable state from disk and does NOT wipe the log
+      const dropped = await compactLog(4);
+      expect(dropped).toBe(0);
+
+      // All 3 durable entries on disk remain intact
+      const lines = readFileSync(syncLogPath(), 'utf8').trim().split('\n');
+      expect(lines).toHaveLength(3);
+      expect(getCurrentSeq()).toBe(3);
+
+      // Sequence recovery across restart recovers the durable seq 3
+      await initSyncLog();
+      expect(getCurrentSeq()).toBe(3);
+      const next = await appendChange('create', 'people', 'p3', { name: 'Person 3' }, 'inst-1');
+      expect(next.seq).toBe(4);
+    });
+
+    it('replays LWW rules correctly in terminal compaction', async () => {
+      writeLog(
+        '{"seq":1,"op":"create","type":"links","id":"x","record":{"updatedAt":"2026-01-01T00:00:00.000Z"}}\n' +
+        '{"seq":2,"op":"delete","type":"links","id":"x","record":{"updatedAt":"2026-01-02T00:00:00.000Z"}}\n' +
+        '{"seq":99,"op":"create","type":"links","id":"x","record":{"updatedAt":"2026-01-01T00:00:00.000Z"}}\n'
+      );
+      await initSyncLog();
+      expect(await compactLog(0)).toBe(2);
+
+      const lines = readFileSync(syncLogPath(), 'utf8').trim().split('\n').map(l => JSON.parse(l));
+      expect(lines).toHaveLength(1);
+      expect(lines[0].op).toBe('delete'); // LWW winner
+      expect(lines[0].record.updatedAt).toBe('2026-01-02T00:00:00.000Z');
+      expect(lines[0].seq).toBe(99); // max seq preserved
+    });
+
+    it('preserves pre-floor LWW winner before verbatim tail when tail is stale under positive floor', async () => {
+      // seq 1: create links x (Jan-01)
+      // seq 2: delete links x (Jan-02) -> winning delete
+      // seq 99: stale create links x (Jan-01) -> stale echoed create in tail
+      writeLog(
+        '{"seq":1,"op":"create","type":"links","id":"x","record":{"updatedAt":"2026-01-01T00:00:00.000Z"}}\n' +
+        '{"seq":2,"op":"delete","type":"links","id":"x","record":{"updatedAt":"2026-01-02T00:00:00.000Z"}}\n' +
+        '{"seq":99,"op":"create","type":"links","id":"x","record":{"updatedAt":"2026-01-01T00:00:00.000Z"}}\n'
+      );
+      await initSyncLog();
+
+      // Compact with positive floor (50). Tail (seq >= 50) is kept verbatim,
+      // and seq 2 (delete Jan-02) is preserved before the tail so a fresh peer
+      // replaying from since=0 rejects the stale create and avoids resurrecting the record.
+      const dropped = await compactLog(50);
+      expect(dropped).toBe(1); // seq 1 is dropped, seq 2 and seq 99 are kept
+
+      const lines = readFileSync(syncLogPath(), 'utf8').trim().split('\n').map(l => JSON.parse(l));
+      expect(lines).toHaveLength(2);
+      expect(lines[0].seq).toBe(2);
+      expect(lines[0].op).toBe('delete');
+      expect(lines[0].record.updatedAt).toBe('2026-01-02T00:00:00.000Z');
+      expect(lines[1].seq).toBe(99);
+      expect(lines[1].op).toBe('create');
+
+      // A fresh peer pulling from since=0 gets seq 2 (delete) then seq 99 (stale create)
+      const fromZero = await getChangesSince(0);
+      expect(fromZero.changes.map(c => c.seq)).toEqual([2, 99]);
+    });
   });
 });

server/services/syncOrchestrator.js

@@ -867,14 +867,19 @@ export async function syncAllPeers() {
   // runs (vs. the old skip-when-empty) so a 0 floor is an explicit "keep all",
   // and the anti-entropy reconcile (Part 1) re-converges anyone genuinely behind.
   const brainPeers = peers.filter(p => p.enabled && p.instanceId && getEffectiveCategories(p).brain);
+  let minSeq = 0;
   if (brainPeers.length > 0) {
     const consumedSeqs = brainPeers.map(p => {
       const consumed = p.remoteSyncSeqs?.cursorForYou?.brainSeq;
       return typeof consumed === 'number' && Number.isFinite(consumed) && consumed >= 0 ? consumed : 0;
     });
-    const minSeq = Math.min(...consumedSeqs);
-    await brainSyncLog.compactLog(minSeq);
+    minSeq = Math.min(...consumedSeqs);
   }
+  // When brain peers exist, minSeq preserves unconsumed deltas above the floor.
+  // When no brain peers are enabled (or on standalone installs, #5439), floor 0
+  // runs compatibility-preserving compaction: pruning intermediate update churn
+  // while retaining terminal state for inbound, asymmetric, or pre-#1077 consumers.
+  await brainSyncLog.compactLog(minSeq);
 }
 
 /**