fix(files): tag agent stream frames no-persist across replicas

A peer task tailing an agent-streamed preview frame previously applied it
as REDIS_ORIGIN, marking the seeded room edited and making a transient
startup-race duplicate eligible for that task's last-disconnect flush. Mark
agent frames with a stream field so peers apply them as REDIS_AGENT_ORIGIN,
excluded from the edited/persist gate. The copilot's durable edit_content
write stays the sole authority over file bytes.

Author: waleedlatif1

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

@@ -100,7 +100,7 @@ function makeClient(): any {
 
 vi.mock('redis', () => ({ createClient: () => makeClient() }))
 
-import { FileDocStore } from '@/handlers/file-doc-store'
+import { FileDocStore, REDIS_AGENT_ORIGIN, REDIS_ORIGIN } from '@/handlers/file-doc-store'
 
 const REDIS_URL = 'redis://fake'
 const NAME = 'workspace-file-doc:file-1'
@@ -239,6 +239,31 @@ describe('FileDocStore', () => {
     expect(stream[stream.length - 1].message.s).toBe('1')
   })
 
+  it('tags an agent-streamed frame so a peer tailer applies it as REDIS_AGENT_ORIGIN (never persisted)', async () => {
+    const streamKey = `filedoc:stream:${NAME}`
+    const a = await newStore()
+    const b = await newStore()
+    const bDoc = new Y.Doc()
+    // Capture the origin the tailer stamps each applied entry with — the persistence gate keys off it.
+    const origins: unknown[] = []
+    bDoc.on('update', (_u: Uint8Array, origin: unknown) => origins.push(origin))
+    await b.attachRoom(NAME, bDoc)
+
+    // A normal edit tails as REDIS_ORIGIN (a peer edit that CAN be persisted).
+    a.publish(NAME, updateFor('user edit'))
+    await vi.waitFor(() => expect(origins).toContain(REDIS_ORIGIN), { timeout: 2000 })
+
+    // An agent-streamed frame is published WITH the agent flag: the stream entry carries the marker, and
+    // the peer tailer applies it as REDIS_AGENT_ORIGIN — excluded from the relay's edited/persist gate.
+    a.publish(NAME, updateFor('agent frame'), true)
+    await vi.waitFor(() => expect(origins).toContain(REDIS_AGENT_ORIGIN), { timeout: 2000 })
+    const stream = state.backing!.streams.get(streamKey)!
+    expect(stream.some((e) => e.message.a === '1')).toBe(true)
+    // The normal edit's entry carries no agent marker.
+    expect(stream.filter((e) => e.message.a === '1')).toHaveLength(1)
+    bDoc.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

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

@@ -90,6 +90,16 @@ export const REDIS_ORIGIN = Symbol('file-doc-redis')
  */
 export const REDIS_SNAPSHOT_ORIGIN = Symbol('file-doc-redis-snapshot')
 
+/**
+ * Origin for an AGENT-STREAMED frame applied from the stream (a copilot output token relayed via
+ * {@link FILE_DOC_MESSAGE_TYPE.SYNC_NO_PERSIST}). A peer task tails these to stay live mid-stream, but
+ * they are transient preview content the copilot's durable `edit_content` write reconciles — so the
+ * relay's edit-tracker must NOT mark the doc edited on them (a startup-race duplicate between two stream
+ * leaders would otherwise become eligible for a peer task's persist). Behaves like {@link REDIS_ORIGIN}
+ * otherwise (already in the stream — never re-published).
+ */
+export const REDIS_AGENT_ORIGIN = Symbol('file-doc-redis-agent')
+
 const STREAM_PREFIX = 'filedoc:stream:'
 /** Cluster-wide "durable version the live doc is synced to" (the persist If-Match token). */
 const SYNC_VERSION_PREFIX = 'filedoc:syncver:'
@@ -103,6 +113,9 @@ const UPDATE_FIELD = 'u'
 /** Marks a stream entry as a compaction SNAPSHOT (folds seed + edits), so the tailer applies it with
  * {@link REDIS_SNAPSHOT_ORIGIN}. Present only on snapshot entries. */
 const SNAPSHOT_FIELD = 's'
+/** Marks a stream entry as an AGENT-STREAMED preview frame, so the tailer applies it with
+ * {@link REDIS_AGENT_ORIGIN} (never marks the doc edited). Present only on agent-frame entries. */
+const AGENT_FIELD = 'a'
 
 /** Sentinel token a DISABLED store returns from a lock acquire, so single-replica callers proceed
  * without special-casing; {@link FileDocStore.releaseLock} treats it as a no-op. Not a real UUID, so it
@@ -264,12 +277,14 @@ export class FileDocStore {
    * an edit from the shared log. Only the `xAdd` is retried; the TTL refresh + compaction check are
    * post-write best-effort and never re-trigger the append. Throws if the append ultimately fails.
    */
-  private async appendUpdate(name: string, update: Uint8Array): Promise<void> {
+  private async appendUpdate(name: string, update: Uint8Array, agent = false): Promise<void> {
     if (!this.write) return
     const encoded = Buffer.from(update).toString('base64')
+    const fields: Record<string, string> = { [UPDATE_FIELD]: encoded }
+    if (agent) fields[AGENT_FIELD] = '1'
     for (let attempt = 0; attempt <= PUBLISH_MAX_RETRIES; attempt++) {
       try {
-        await this.write.xAdd(streamKey(name), '*', { [UPDATE_FIELD]: encoded })
+        await this.write.xAdd(streamKey(name), '*', fields)
         break
       } catch (error) {
         if (attempt === PUBLISH_MAX_RETRIES) {
@@ -288,11 +303,12 @@ export class FileDocStore {
 
   /**
    * Fire-and-forget append for the hot keystroke path (`doc.on('update')`): converges peers without
-   * blocking the relay. Retries internally; never throws. No-op when disabled.
+   * blocking the relay. Retries internally; never throws. No-op when disabled. Pass `agent: true` for
+   * a copilot preview frame so peer tasks tail it as {@link REDIS_AGENT_ORIGIN} and never persist it.
    */
-  publish(name: string, update: Uint8Array): void {
+  publish(name: string, update: Uint8Array, agent = false): void {
     if (!this.enabled || !this.write) return
-    void this.appendUpdate(name, update).catch(() => {}) // already logged inside appendUpdate
+    void this.appendUpdate(name, update, agent).catch(() => {}) // already logged inside appendUpdate
   }
 
   /**
@@ -515,8 +531,13 @@ export class FileDocStore {
   private applyEntry(room: StoreRoom, id: string, message: Record<string, string>): void {
     room.lastId = id
     // A compaction snapshot folds seed + edits into one frame; stamp it so the relay's edit-tracker
-    // treats a fresh catch-up from it as edited (a snapshot only exists once real edits accumulated).
-    const origin = message[SNAPSHOT_FIELD] ? REDIS_SNAPSHOT_ORIGIN : REDIS_ORIGIN
+    // treats a fresh catch-up from it as edited (a snapshot only exists once real edits accumulated). An
+    // agent-streamed preview frame is stamped separately so the tracker NEVER marks it edited.
+    const origin = message[SNAPSHOT_FIELD]
+      ? REDIS_SNAPSHOT_ORIGIN
+      : message[AGENT_FIELD]
+        ? REDIS_AGENT_ORIGIN
+        : REDIS_ORIGIN
     applyEntryToDoc(room.doc, id, message, origin)
   }
 

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

@@ -45,7 +45,12 @@ import * as syncProtocol from 'y-protocols/sync'
 import * as Y from 'yjs'
 import { resolveAvatarUrl } from '@/handlers/avatar'
 import { fetchFileDocMerge, fetchFileDocPersist, fetchFileDocSeed } from '@/handlers/file-doc-app'
-import { getFileDocStore, REDIS_ORIGIN, REDIS_SNAPSHOT_ORIGIN } from '@/handlers/file-doc-store'
+import {
+  getFileDocStore,
+  REDIS_AGENT_ORIGIN,
+  REDIS_ORIGIN,
+  REDIS_SNAPSHOT_ORIGIN,
+} from '@/handlers/file-doc-store'
 import { resolveRoomJoinAuth } from '@/handlers/room-join-auth'
 import type { AuthenticatedSocket } from '@/middleware/auth'
 import type { IRoomManager } from '@/rooms'
@@ -708,23 +713,29 @@ function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom {
       origin === AGENT_SYNC_ORIGIN ? null : originSocketId(origin)
     )
     // Share every locally-originated update to the stream so peers converge. Skip updates that already
-    // came FROM the stream (REDIS_ORIGIN / REDIS_SNAPSHOT_ORIGIN) and SEED_ORIGIN — the seed is published
-    // EXPLICITLY and AWAITED under the seed lock (so it lands before the lock releases), which a
-    // fire-and-forget publish here couldn't guarantee.
-    if (origin !== REDIS_ORIGIN && origin !== REDIS_SNAPSHOT_ORIGIN && origin !== SEED_ORIGIN)
-      getFileDocStore().publish(name, update)
+    // came FROM the stream (REDIS_ORIGIN / REDIS_SNAPSHOT_ORIGIN / REDIS_AGENT_ORIGIN) and SEED_ORIGIN —
+    // the seed is published EXPLICITLY and AWAITED under the seed lock (so it lands before the lock
+    // releases), which a fire-and-forget publish here couldn't guarantee. An agent-streamed frame
+    // (AGENT_SYNC_ORIGIN) is published WITH the agent marker so peer tasks tail it as REDIS_AGENT_ORIGIN
+    // and never mark the doc edited on it (see the edit-tracker below).
+    if (
+      origin !== REDIS_ORIGIN &&
+      origin !== REDIS_SNAPSHOT_ORIGIN &&
+      origin !== REDIS_AGENT_ORIGIN &&
+      origin !== SEED_ORIGIN
+    )
+      getFileDocStore().publish(name, update, origin === AGENT_SYNC_ORIGIN)
     // Edit tracking for persistence. Mark the doc dirty on any update applied AFTER it was seeded — a
     // local user edit (socket origin) OR a peer's edit relayed via the tailer (REDIS_ORIGIN) — so
     // whichever task is last to leave persists real edits, even one that only tailed them. A compaction
     // snapshot on catch-up (REDIS_SNAPSHOT_ORIGIN) also counts: it folds real edits into one frame, so a
     // fresh task catching up purely from it must not treat the doc as unedited. The seed transition
-    // itself is never counted, so a seeded-but-unedited doc is never projected back over the file. NOTE:
-    // an agent-streamed frame ({@link FILE_DOC_MESSAGE_TYPE.SYNC_NO_PERSIST}) is not counted on the
-    // ORIGINATING task (it applies under {@link AGENT_SYNC_ORIGIN} — see the handler), but in the multi-replica
-    // path it is published to the stream and PEER tasks apply it as REDIS_ORIGIN, indistinguishable from a
-    // peer edit, so it marks `edited` there. That only ever causes an extra idempotent persist of content
-    // the copilot's final `edit_content` write persists durably anyway (safe over-persist, never a lost
-    // edit); fully suppressing it would require tagging the stream entry as no-persist across replicas.
+    // itself is never counted, so a seeded-but-unedited doc is never projected back over the file. An
+    // agent-streamed frame ({@link FILE_DOC_MESSAGE_TYPE.SYNC_NO_PERSIST}) is never counted anywhere: on
+    // the ORIGINATING task it applies under {@link AGENT_SYNC_ORIGIN}, and across replicas it is published
+    // WITH the agent marker so PEER tasks tail it as REDIS_AGENT_ORIGIN — neither is in the edited set
+    // below. So a transient startup-race duplicate between two stream leaders is never eligible for
+    // persistence; the copilot's durable `edit_content` write remains the sole authority over file bytes.
     const seededBefore = room.seededObserved
     if (isDocSeeded(room.doc)) room.seededObserved = true
     if (