fix(files): agent stream frames skip the relay's durable persist

Cursor round 1 (High): client-applied stream frames broadcast over the sync
channel, so the relay stamped a socket origin and ran schedulePersist — durably
writing partial agent content mid-stream, attributed to the watching user (the old
server-merge applied with no origin and never did). Restore that behavior:

- new FILE_DOC_MESSAGE_TYPE.SYNC_NO_PERSIST wire tag; the provider tags
  AGENT_STREAM_ORIGIN updates with it (normal user edits stay SYNC)
- the relay applies it under an AgentSyncOrigin (carries the socket id for
  broadcast exclusion, but is not a plain string) so originSocketId() is null →
  no edited/schedulePersist/lastEditorUserId; excludeSocketId() still excludes the
  sender, and the update still publishes to the stream so peers converge
- the copilot's final edit_content write remains the authoritative durable persist
- tests: relay applies+fans-out but never persists a SYNC_NO_PERSIST frame
  (verified it fails if applied as a socket edit); provider tags agent edits

Author: waleedlatif1

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

@@ -299,6 +299,36 @@ describe('setupWorkspaceFileDocHandlers', () => {
     expect(mockFetchFileDocPersist).toHaveBeenCalled()
   })
 
+  it('applies + fans out an agent-streamed frame (SYNC_NO_PERSIST) but never persists it', async () => {
+    mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server'))
+    const { io, sent } = createIo()
+    const { handlers } = setup('socket-1', io)
+    await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
+    await flushMicrotasks()
+
+    const before = sent.length
+    const edit = new Y.Doc()
+    edit.getText(FILE_DOC_FIELD).insert(0, 'agent streamed this')
+    handlers[FILE_DOC_EVENTS.MESSAGE](
+      frame(FILE_DOC_MESSAGE_TYPE.SYNC_NO_PERSIST, (e) =>
+        syncProtocol.writeUpdate(e, Y.encodeStateAsUpdate(edit))
+      )
+    )
+    await flushMicrotasks()
+
+    // It fans out to the room excluding the sender, so a collaborator sees the stream live...
+    const fanout = sent
+      .slice(before)
+      .filter((m) => m.event === FILE_DOC_EVENTS.MESSAGE && m.except === 'socket-1')
+    expect(fanout.length).toBeGreaterThan(0)
+
+    // ...but it must NOT mark the doc dirty: a last-disconnect flush never persists agent content (the
+    // copilot's final edit_content write is the authoritative durable persist).
+    cleanupFileDocForSocket('socket-1', io, true)
+    await flushMicrotasks()
+    expect(mockFetchFileDocPersist).not.toHaveBeenCalled()
+  })
+
   it('stops on a persist conflict without clobbering (single attempt, durable left authoritative)', async () => {
     mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server'))
     // A persist reports an out-of-band change (If-Match conflict). The relay must NOT re-persist against

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

@@ -177,6 +177,25 @@ function originSocketId(origin: unknown): string | null {
   return typeof origin === 'string' ? origin : null
 }
 
+/**
+ * The transaction origin stamped on an agent-streamed frame (a {@link FILE_DOC_MESSAGE_TYPE.SYNC_NO_PERSIST}
+ * apply). It carries the emitting socket id for broadcast exclusion, but is deliberately NOT a plain
+ * string — so `originSocketId` returns `null` for it and the update never triggers `edited`/`schedulePersist`
+ * (the copilot's final `edit_content` write is the durable persist).
+ */
+interface AgentSyncOrigin {
+  readonly agentSocketId: string
+}
+
+function isAgentSyncOrigin(origin: unknown): origin is AgentSyncOrigin {
+  return typeof origin === 'object' && origin !== null && 'agentSocketId' in origin
+}
+
+/** The socket id to exclude when relaying an update — a client socket edit OR an agent-streamed frame. */
+function excludeSocketId(origin: unknown): string | null {
+  return originSocketId(origin) ?? (isAgentSyncOrigin(origin) ? origin.agentSocketId : null)
+}
+
 /**
  * Broadcast an AWARENESS frame to the room ACROSS tasks via the Socket.IO Redis adapter. Awareness
  * (cursors/selection) is ephemeral and needs no convergence or replay, so the adapter's cross-task
@@ -684,9 +703,10 @@ function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom {
     const encoder = encoding.createEncoder()
     encoding.writeVarUint(encoder, FILE_DOC_MESSAGE_TYPE.SYNC)
     syncProtocol.writeUpdate(encoder, update)
-    // Fan out to THIS task's clients only (excluding the origin socket if local). Cross-task delivery
-    // rides the shared stream — every task's tailer applies + runs its own local fan-out.
-    broadcastLocal(io, name, encoding.toUint8Array(encoder), originSocketId(origin))
+    // Fan out to THIS task's clients only (excluding the origin socket if local — a user edit OR an
+    // agent-streamed frame). Cross-task delivery rides the shared stream — every task's tailer applies +
+    // runs its own local fan-out.
+    broadcastLocal(io, name, encoding.toUint8Array(encoder), excludeSocketId(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
@@ -781,6 +801,22 @@ function handleMessage(socket: AuthenticatedSocket, data: unknown) {
         }
         break
       }
+      case FILE_DOC_MESSAGE_TYPE.SYNC_NO_PERSIST: {
+        // An agent-streamed frame: apply + fan out to peers (so a collaborator sees the stream live) but
+        // do NOT treat it as a durable user edit. Unlike SYNC, we do NOT set `lastEditorUserId`, and the
+        // apply uses an {@link AgentSyncOrigin} (not the bare socket id) so `originSocketId` is `null` in
+        // `doc.on('update')` — skipping `edited`/`schedulePersist`. `excludeSocketId` still reads the
+        // carried socket id, so the sender is excluded from the relay fan-out. The copilot's final
+        // `edit_content` write remains the authoritative durable persist.
+        const encoder = encoding.createEncoder()
+        encoding.writeVarUint(encoder, FILE_DOC_MESSAGE_TYPE.SYNC)
+        const agentOrigin: AgentSyncOrigin = { agentSocketId: socket.id }
+        syncProtocol.readSyncMessage(decoder, encoder, room.doc, agentOrigin)
+        if (encoding.length(encoder) > 1) {
+          socket.emit(FILE_DOC_EVENTS.MESSAGE, encoding.toUint8Array(encoder))
+        }
+        break
+      }
       case FILE_DOC_MESSAGE_TYPE.AWARENESS: {
         const update = decoding.readVarUint8Array(decoder)
         // Enforce presence ownership: a socket may only publish/remove awareness

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

@@ -12,7 +12,7 @@ const COLLAB_DOC_FIELD = 'default'
  * the `ySyncPluginKey` origin that local user edits use, so the Collaboration UndoManager — which
  * tracks only `ySyncPluginKey` — excludes streamed ops from the user's undo stack.
  */
-const AGENT_STREAM_ORIGIN = Symbol('agent-stream')
+export const AGENT_STREAM_ORIGIN = Symbol('agent-stream')
 
 /**
  * A private Yjs replica the agent stream reconciles against, so a stream writes into the live doc as a

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

@@ -12,6 +12,7 @@ import { describe, expect, it, vi } from 'vitest'
 import * as awarenessProtocol from 'y-protocols/awareness'
 import * as syncProtocol from 'y-protocols/sync'
 import * as Y from 'yjs'
+import { AGENT_STREAM_ORIGIN } from './apply-streamed-markdown'
 import { FileDocProvider } from './file-doc-provider'
 
 /** A minimal fake Socket.IO client whose server→client events can be fired in tests. */
@@ -125,6 +126,19 @@ describe('FileDocProvider', () => {
     expect(messages[0][0]).toBe(FILE_DOC_MESSAGE_TYPE.SYNC)
   })
 
+  it('tags agent-streamed edits as SYNC_NO_PERSIST so the relay skips the durable persist', () => {
+    const { doc, emit } = createProvider(true)
+    emit.mockClear()
+
+    // An agent-streamed frame is applied under AGENT_STREAM_ORIGIN; it must still reach the server (peers
+    // see it live) but as SYNC_NO_PERSIST, so the relay fans it out without treating it as a user edit.
+    doc.transact(() => doc.getText('default').insert(0, 'agent'), AGENT_STREAM_ORIGIN)
+
+    const messages = emittedMessages(emit)
+    expect(messages.length).toBe(1)
+    expect(messages[0][0]).toBe(FILE_DOC_MESSAGE_TYPE.SYNC_NO_PERSIST)
+  })
+
   it('does not echo updates it applied from the server', () => {
     const { provider, emit, fire } = createProvider(true)
     fire(FILE_DOC_EVENTS.JOIN_SUCCESS, { fileId: 'file-1' })

apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/file-doc-provider.ts

@@ -14,6 +14,7 @@ import type { Socket } from 'socket.io-client'
 import * as awarenessProtocol from 'y-protocols/awareness'
 import * as syncProtocol from 'y-protocols/sync'
 import type * as Y from 'yjs'
+import { AGENT_STREAM_ORIGIN } from './apply-streamed-markdown'
 
 /**
  * Events emitted by {@link FileDocProvider}.
@@ -276,8 +277,15 @@ export class FileDocProvider extends ObservableV2<FileDocProviderEvents> {
     if (this.fatal) return
     // Updates we applied from the server carry `this` as origin — don't echo them.
     if (origin === this) return
+    // Agent-streamed frames must reach peers (so a collaborator sees the stream live) but must NOT be
+    // treated by the server as a durable user edit — the copilot's final `edit_content` write is the
+    // authoritative persist. Tag them so the relay applies + fans out but skips persist bookkeeping.
+    const messageType =
+      origin === AGENT_STREAM_ORIGIN
+        ? FILE_DOC_MESSAGE_TYPE.SYNC_NO_PERSIST
+        : FILE_DOC_MESSAGE_TYPE.SYNC
     const encoder = encoding.createEncoder()
-    encoding.writeVarUint(encoder, FILE_DOC_MESSAGE_TYPE.SYNC)
+    encoding.writeVarUint(encoder, messageType)
     syncProtocol.writeUpdate(encoder, update)
     this.socket.emit(FILE_DOC_EVENTS.MESSAGE, encoding.toUint8Array(encoder))
   }

packages/realtime-protocol/src/file-doc.ts

@@ -42,6 +42,15 @@ export const FILE_DOC_EVENTS = {
 export const FILE_DOC_MESSAGE_TYPE = {
   SYNC: 0,
   AWARENESS: 1,
+  /**
+   * Client → server: a Yjs sync UPDATE (same framing as {@link FILE_DOC_MESSAGE_TYPE.SYNC}) that the
+   * server must apply and fan out to peers WITHOUT treating it as a durable user edit — no
+   * `schedulePersist`, no `edited`/`lastEditorUserId` bookkeeping. Used for agent-streamed frames: the
+   * copilot's final `edit_content` write is the authoritative durable persist, so the live stream must
+   * not also durably write partial content (attributed to the watching user). The server never sends
+   * this type; replies always use {@link FILE_DOC_MESSAGE_TYPE.SYNC}.
+   */
+  SYNC_NO_PERSIST: 2,
 } as const
 
 export type FileDocMessageType = (typeof FILE_DOC_MESSAGE_TYPE)[keyof typeof FILE_DOC_MESSAGE_TYPE]