fix(files): reseed agent shadow on lead regain + agent-only compaction

Two multi-writer edge cases surfaced in review:

- rich-markdown-editor: a client that led, lost leadership, then regained it
  reused its stale shadow (which never saw the interim leader's ops), re-emitting
  ops for content already present. Tear the shadow down when a client observes it
  is not the leader, so a regain rebuilds fresh from the current doc.

- file-doc-store: compaction always stamped its snapshot REDIS_SNAPSHOT_ORIGIN
  (marks peers edited). A long agent-only stream crossing the threshold could
  fold preview content into a persist-eligible snapshot. Track whether a room
  integrated any real edit and stamp an agent-only snapshot REDIS_AGENT_ORIGIN
  so it stays no-persist.

Both covered by falsification-verified tests.

Author: waleedlatif1

apps/realtime/src/handlers/file-doc-store.test.ts

@@ -223,8 +223,14 @@ describe('FileDocStore', () => {
 
     const a = await newStore()
     // This task has integrated only up to entry 400 (all no-ops) — its local doc is empty and lags the
-    // two peer entries. Inject that lagging room directly.
-    ;(a as any).rooms.set(NAME, { doc: new Y.Doc(), lastId: '400-0', publishes: 0 })
+    // two peer entries. Inject that lagging room directly (a real edit was integrated → realEdited).
+    ;(a as any).rooms.set(NAME, {
+      doc: new Y.Doc(),
+      lastId: '400-0',
+      publishes: 0,
+      seededObserved: true,
+      realEdited: true,
+    })
     await (a as any).maybeCompact(NAME)
 
     // A fresh catch-up must still reconstruct the peer content — compaction must not have trimmed 401/402.
@@ -264,6 +270,42 @@ describe('FileDocStore', () => {
     bDoc.destroy()
   })
 
+  it('stamps a compaction snapshot of an agent-ONLY stream as an agent frame (never persisted)', async () => {
+    const streamKey = `filedoc:stream:${NAME}`
+    const noop = Buffer.from(Y.encodeStateAsUpdate(new Y.Doc())).toString('base64')
+    // A doc whose content is purely agent preview (no real edit integrated) — realEdited stays false.
+    const agentDoc = docWithText('agent-only preview body')
+    const entries = Array.from({ length: 400 }, (_, i) => ({
+      id: `${i + 1}-0`,
+      message: { u: noop },
+    }))
+    state.backing!.streams.set(streamKey, entries)
+    state.backing!.seq = 400
+
+    const a = await newStore()
+    ;(a as any).rooms.set(NAME, {
+      doc: agentDoc,
+      lastId: '400-0',
+      publishes: 0,
+      seededObserved: true,
+      realEdited: false,
+    })
+    await (a as any).maybeCompact(NAME)
+
+    // The snapshot must carry the AGENT marker, NOT the snapshot marker, so a peer catch-up applies it as
+    // REDIS_AGENT_ORIGIN and never marks the doc edited — the no-persist guarantee survives compaction.
+    const stream = state.backing!.streams.get(streamKey)!
+    const last = stream[stream.length - 1].message
+    expect(last.a).toBe('1')
+    expect(last.s).toBeUndefined()
+    // Content is still fully reconstructable from the compacted stream.
+    const doc = new Y.Doc()
+    Y.applyUpdate(doc, (await a.getStreamState(NAME))!)
+    expect(doc.getText('body').toString()).toBe('agent-only preview body')
+    doc.destroy()
+    agentDoc.destroy()
+  })
+
   it('retries a transient append failure so the edit is not lost from the shared log', async () => {
     const a = await newStore()
     state.backing!.failXAdd = 2 // first two xAdd attempts throw; the third must succeed
@@ -407,8 +449,20 @@ describe('FileDocStore', () => {
     const b = await newStore()
     const docA = new Y.Doc()
     Y.applyUpdate(docA, peerUpdates[0]) // A integrated up to 401
-    ;(a as any).rooms.set(NAME, { doc: docA, lastId: '401-0', publishes: 0 })
-    ;(b as any).rooms.set(NAME, { doc: new Y.Doc(), lastId: '400-0', publishes: 0 })
+    ;(a as any).rooms.set(NAME, {
+      doc: docA,
+      lastId: '401-0',
+      publishes: 0,
+      seededObserved: true,
+      realEdited: true,
+    })
+    ;(b as any).rooms.set(NAME, {
+      doc: new Y.Doc(),
+      lastId: '400-0',
+      publishes: 0,
+      seededObserved: true,
+      realEdited: true,
+    })
     await Promise.all([(a as any).maybeCompact(NAME), (b as any).maybeCompact(NAME)])
 
     const doc = new Y.Doc()

apps/realtime/src/handlers/file-doc-store.ts

@@ -33,7 +33,7 @@
  * @module
  */
 import { createLogger } from '@sim/logger'
-import { FILE_DOC_TIMEOUTS } from '@sim/realtime-protocol/file-doc'
+import { FILE_DOC_SEED, FILE_DOC_TIMEOUTS } from '@sim/realtime-protocol/file-doc'
 import { getErrorMessage } from '@sim/utils/errors'
 import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
@@ -180,13 +180,27 @@ function applyEntryToDoc(
   }
 }
 
+/** Whether a doc carries the seed flag (mirrors the relay's `isDocSeeded`), so the store can tell the
+ * one-time seed transition from a real post-seed edit without re-implementing the check divergently. */
+function isDocSeeded(doc: Y.Doc): boolean {
+  return doc.getMap(FILE_DOC_SEED.configMap).get(FILE_DOC_SEED.flag) === true
+}
+
 /** One locally-open room the store tracks: its doc and the last stream id applied to it. */
 interface StoreRoom {
   doc: Y.Doc
   /** The id of the last stream entry applied to `doc`; the tailer resumes strictly after it. */
   lastId: string
   /** Local publish count, to pace compaction checks. */
   publishes: number
+  /** Set once the doc has been observed seeded, so the seed transition itself is never mistaken for an
+   * edit (mirrors the relay's `seededObserved`). */
+  seededObserved: boolean
+  /** Whether the doc has integrated any REAL (non-agent, non-seed) edit. Compaction stamps its snapshot
+   * as an AGENT snapshot ({@link REDIS_AGENT_ORIGIN}, never persisted) until this is true, so a long
+   * agent-only stream that crosses the compaction threshold can't fold its preview content into a
+   * snapshot that marks peers edited. */
+  realEdited: boolean
 }
 
 /**
@@ -250,7 +264,13 @@ export class FileDocStore {
     if (!this.enabled || !this.write) return
     // Register BEFORE the async read so a concurrent publish/tailer for this room can't be missed —
     // the tailer resumes from `lastId`, which the catch-up advances.
-    const room: StoreRoom = { doc, lastId: '0', publishes: 0 }
+    const room: StoreRoom = {
+      doc,
+      lastId: '0',
+      publishes: 0,
+      seededObserved: false,
+      realEdited: false,
+    }
     this.rooms.set(name, room)
     try {
       const entries = await this.write.xRange(streamKey(name), '-', '+')
@@ -298,7 +318,13 @@ export class FileDocStore {
     }
     await this.write.expire(streamKey(name), STREAM_TTL_SEC).catch(() => {})
     const room = this.rooms.get(name)
-    if (room && ++room.publishes % COMPACT_CHECK_EVERY === 0) void this.maybeCompact(name)
+    if (room) {
+      // A local non-agent publish (a user edit or the awaited copilot durable merge) is a real edit; the
+      // seed never flows through here (it uses seedIfEmpty). Set it BEFORE the compaction check below so a
+      // real edit can never be folded into an agent (no-persist) snapshot due to a tail-back race.
+      if (!agent) room.realEdited = true
+      if (++room.publishes % COMPACT_CHECK_EVERY === 0) void this.maybeCompact(name)
+    }
   }
 
   /**
@@ -538,7 +564,16 @@ export class FileDocStore {
       : message[AGENT_FIELD]
         ? REDIS_AGENT_ORIGIN
         : REDIS_ORIGIN
+    const seededBefore = room.seededObserved
     applyEntryToDoc(room.doc, id, message, origin)
+    if (isDocSeeded(room.doc)) room.seededObserved = true
+    // Track a real edit integrated from the stream so compaction knows whether its snapshot represents
+    // real content or agent-only preview: a real snapshot (folds real edits), or a markerless edit
+    // applied AFTER the doc was already seeded (the seed transition itself never counts). Agent frames
+    // and agent snapshots (REDIS_AGENT_ORIGIN) never count.
+    if (origin === REDIS_SNAPSHOT_ORIGIN || (origin === REDIS_ORIGIN && seededBefore)) {
+      room.realEdited = true
+    }
   }
 
   /**
@@ -599,10 +634,14 @@ export class FileDocStore {
         // appended snapshot id instead would silently drop those un-integrated peer entries.
         const upTo = room.lastId
         const snapshot = Buffer.from(Y.encodeStateAsUpdate(room.doc)).toString('base64')
-        // Mark it a snapshot so a fresh catch-up task treats it as edited content, not a bare seed.
+        // Stamp the snapshot by what it folds: a real edit → SNAPSHOT_FIELD (a fresh catch-up treats it
+        // as edited content, not a bare seed). An agent-ONLY stream (no real edit yet) → AGENT_FIELD, so a
+        // peer catching up applies it as REDIS_AGENT_ORIGIN and never marks the doc edited — preserving
+        // the no-persist guarantee even when a long copilot stream alone crosses the compaction threshold.
+        const marker = room.realEdited ? SNAPSHOT_FIELD : AGENT_FIELD
         await this.write.xAdd(streamKey(name), '*', {
           [UPDATE_FIELD]: snapshot,
-          [SNAPSHOT_FIELD]: '1',
+          [marker]: '1',
         })
         // MINID keeps entries with id >= upTo: the snapshot, any un-integrated peer entries, and
         // `upTo` itself (redundant with the snapshot, harmless); it drops only the folded older deltas.

apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/apply-streamed-markdown.test.ts

@@ -99,6 +99,46 @@ describe('agent-stream applier', () => {
     expect(editor.getText()).toContain('Agent wrote this')
   })
 
+  it('a shadow reused after the live doc advanced duplicates content; a fresh one does not', () => {
+    // The invariant behind the component's leadership-regain teardown: a shadow tracks only ITS OWN
+    // reconciles, so once the live doc advances under another writer, REUSING that stale shadow re-emits
+    // ops for content already present (duplication). Seeding a FRESH shadow from the current doc fixes it.
+    const stale = track(makeCollabEditor())
+    const staleSession = beginAgentStream(stale.editor)! // seeded from the empty base
+    applyAgentStreamFrame(stale.editor, staleSession, 'Alpha paragraph.')
+    // Another writer advances the live doc while this shadow is NOT looking (a handoff to an interim leader).
+    stale.editor.commands.focus('end')
+    stale.editor.commands.insertContent('\n\nBeta paragraph.')
+    // Reusing the stale shadow (only knows "Alpha") to reconcile toward the full body re-inserts "Beta".
+    applyAgentStreamFrame(
+      stale.editor,
+      staleSession,
+      'Alpha paragraph.\n\nBeta paragraph.\n\nGamma paragraph.'
+    )
+    endAgentStream(staleSession)
+    const staleText = stale.editor.getText()
+    expect(staleText.match(/Beta paragraph/g)?.length).toBe(2) // duplicated — what the teardown prevents
+
+    // Fresh shadow re-seeded from the CURRENT doc (what a regaining leader does after teardown) emits only
+    // the genuine delta, so no content duplicates.
+    const fresh = track(makeCollabEditor())
+    const first = beginAgentStream(fresh.editor)!
+    applyAgentStreamFrame(fresh.editor, first, 'Alpha paragraph.')
+    fresh.editor.commands.focus('end')
+    fresh.editor.commands.insertContent('\n\nBeta paragraph.')
+    endAgentStream(first)
+    const regained = beginAgentStream(fresh.editor)! // re-seeded from the advanced doc
+    applyAgentStreamFrame(
+      fresh.editor,
+      regained,
+      'Alpha paragraph.\n\nBeta paragraph.\n\nGamma paragraph.'
+    )
+    endAgentStream(regained)
+    const freshText = fresh.editor.getText()
+    expect(freshText.match(/Beta paragraph/g)?.length).toBe(1)
+    expect(freshText).toContain('Gamma paragraph')
+  })
+
   it('preserves a concurrent peer edit to a region the agent snapshot does not include', () => {
     // This is the core "AI as a CRDT peer" guarantee: the agent relays only its OWN delta (computed
     // against a private shadow), never a whole-document reconcile that would revert a peer's edit.

apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/rich-markdown-editor.tsx

@@ -918,6 +918,15 @@ export function LoadedRichMarkdownEditor({
             collaboration &&
             !isAgentStreamLeader(collaboration.awareness, collaboration.doc.clientID)
           ) {
+            // Not (or no longer) the leader: discard any shadow this client holds. A shadow only tracks
+            // ITS OWN reconciles, so one kept across a leadership loss goes stale as the interim leader
+            // advances the shared doc; reusing it on a later regain would re-emit ops for content already
+            // present (duplication). Dropping it here means a regain rebuilds a FRESH shadow from the
+            // current doc via the `??=` below — upholding "a non-leader holds none."
+            if (agentStreamSessionRef.current) {
+              endAgentStream(agentStreamSessionRef.current)
+              agentStreamSessionRef.current = null
+            }
             streamRafRef.current = null
             return
           }
@@ -931,8 +940,10 @@ export function LoadedRichMarkdownEditor({
           const el = containerRef.current
           const pinnedToBottom = el ? el.scrollHeight - el.scrollTop - el.clientHeight < 80 : false
           // Open the shadow lazily HERE — only when THIS client actually leads — seeded from the CURRENT
-          // doc, so a handoff successor diffs against the prior leader's ops (no stale base) and a
-          // non-leader never builds one. Defensive: a ready collab editor always has a ySync binding.
+          // doc. A non-leader holds none (torn down above), so whether this client is a first-time leader
+          // or one REGAINING leadership, `??=` finds a null ref and rebuilds fresh from the current doc,
+          // already carrying the interim leader's ops (never a stale base). Defensive: a ready collab
+          // editor always has a ySync binding.
           agentStreamSessionRef.current ??= beginAgentStream(editor)
           const session = agentStreamSessionRef.current
           if (!session || !applyAgentStreamFrame(editor, session, pending)) {