diff --git a/apps/realtime/src/handlers/file-doc-store.test.ts b/apps/realtime/src/handlers/file-doc-store.test.ts index 7a6aba13015..1662a8a3426 100644 --- a/apps/realtime/src/handlers/file-doc-store.test.ts +++ b/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' @@ -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. @@ -239,6 +245,84 @@ 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('latches realEdited synchronously so a concurrent compaction can never mislabel a real edit', async () => { + // The data-loss race: a real edit sits in room.doc synchronously, but if realEdited were set only + // AFTER appendUpdate's awaits, a concurrent agent-triggered compaction could snapshot that content and + // stamp it an agent (no-persist) frame — losing the edit. The latch must be set in the same tick. + const a = await newStore() + const doc = new Y.Doc() + await a.attachRoom(NAME, doc) + const room = (a as any).rooms.get(NAME) + expect(room.realEdited).toBe(false) + // Kick off a real (non-agent) append but do NOT await it: realEdited must already be true before the + // xAdd/expire awaits resolve, so any compaction racing on the awaits sees the real edit. + const pending = (a as any).appendUpdate(NAME, updateFor('real user edit')) + expect(room.realEdited).toBe(true) + await pending + doc.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 @@ -382,8 +466,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() diff --git a/apps/realtime/src/handlers/file-doc-store.ts b/apps/realtime/src/handlers/file-doc-store.ts index b6cf65ce9e5..3865fb1a6d9 100644 --- a/apps/realtime/src/handlers/file-doc-store.ts +++ b/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' @@ -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 @@ -167,6 +180,12 @@ 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 @@ -174,6 +193,14 @@ interface StoreRoom { 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 } /** @@ -237,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), '-', '+') @@ -264,12 +297,25 @@ 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 { + private async appendUpdate(name: string, update: Uint8Array, agent = false): Promise { if (!this.write) return + // Latch realEdited SYNCHRONOUSLY — before the first await — for a real (non-agent) publish. The edit + // already sits in room.doc (applied in doc.on('update') before publish was called), so if this set + // were deferred past the xAdd/expire awaits a CONCURRENT agent-frame-triggered maybeCompact could read + // realEdited=false, snapshot the doc (which already holds this real edit), and stamp it an agent + // (no-persist) snapshot — a lost edit. Setting it in the same synchronous tick as the doc mutation + // makes "room.doc holds a real edit ⇒ realEdited" hold before any compaction (always async) can run. + // Monotonic latch, so an eager set is safe; the seed never flows through here (it uses seedIfEmpty). + if (!agent) { + const editedRoom = this.rooms.get(name) + if (editedRoom) editedRoom.realEdited = true + } const encoded = Buffer.from(update).toString('base64') + const fields: Record = { [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 +334,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,9 +562,23 @@ export class FileDocStore { private applyEntry(room: StoreRoom, id: string, message: Record): 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 + 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 + } } /** @@ -578,10 +639,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. diff --git a/apps/realtime/src/handlers/file-doc.multireplica.test.ts b/apps/realtime/src/handlers/file-doc.multireplica.test.ts index 17c17ffe371..df56ffdc4c0 100644 --- a/apps/realtime/src/handlers/file-doc.multireplica.test.ts +++ b/apps/realtime/src/handlers/file-doc.multireplica.test.ts @@ -60,7 +60,7 @@ describe('applyMarkdownToLiveFileDoc — multi-replica (store-enabled) ordering' mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc())) }) - it('drops a stale-base streaming snapshot against the SHARED synced version and never records it', async () => { + it('drops a stale durable write against the SHARED synced version', async () => { // A durable write (e.g. a concurrent human save on another process) records the shared synced version. expect(await applyMarkdownToLiveFileDoc('file-1', '# durable', { version: 100 })).toBe( 'applied' @@ -68,23 +68,19 @@ describe('applyMarkdownToLiveFileDoc — multi-replica (store-enabled) ordering' expect(fakeStore.setSyncedVersion).toHaveBeenCalledWith(ROOM_NAME, 100) mockFetchFileDocMerge.mockClear() - // A streaming snapshot built from an older base (50) than the SHARED synced version is stale — - // rejected under the lock before any diff is built, so it can't clobber the durable write. - expect( - await applyMarkdownToLiveFileDoc('file-1', '# stale-base stream', { baseVersion: 50 }) - ).toBe('stale') + // A durable write with an OLDER version than the SHARED synced version is stale — rejected under the + // lock before any diff is built, so it can't regress the doc across replicas. + expect(await applyMarkdownToLiveFileDoc('file-1', '# older durable', { version: 50 })).toBe( + 'stale' + ) expect(mockFetchFileDocMerge).not.toHaveBeenCalled() - // A streaming snapshot whose base is the current shared version applies (nothing newer to clobber)... - expect( - await applyMarkdownToLiveFileDoc('file-1', '# current-base stream', { baseVersion: 100 }) - ).toBe('applied') - // ...but is never recorded: a later durable write at 150 still applies. + // A newer durable write applies and advances the shared synced version. expect(await applyMarkdownToLiveFileDoc('file-1', '# durable again', { version: 150 })).toBe( 'applied' ) expect(fakeStore.setSyncedVersion).toHaveBeenCalledWith(ROOM_NAME, 150) - // setSyncedVersion fired only for the two durable writes, never for a streaming snapshot. + // setSyncedVersion fired only for the two applied durable writes, never for the stale one. expect(fakeStore.setSyncedVersion).toHaveBeenCalledTimes(2) }) }) diff --git a/apps/realtime/src/handlers/file-doc.test.ts b/apps/realtime/src/handlers/file-doc.test.ts index 7ccbe3d2254..5121638962a 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -299,6 +299,37 @@ 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 WHOLE room — no socket excluded — so peers AND a same-socket sibling provider see + // the stream live (the emitting provider no-ops on its own echo). + const fanout = sent + .slice(before) + .filter((m) => m.event === FILE_DOC_EVENTS.MESSAGE && m.except === undefined) + 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 @@ -569,39 +600,6 @@ describe('setupWorkspaceFileDocHandlers', () => { expect(mockFetchFileDocMerge).not.toHaveBeenCalled() }) - it('drops a streaming snapshot whose base predates a newer durable write, but never records it', async () => { - mockFetchFileDocSeed.mockResolvedValue(seedResult('# Original')) // seed version 1 - const { io } = createIo() - const { handlers } = setup('socket-1', io) - await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) - await flushMicrotasks() - - mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc())) - - // A durable write (e.g. a concurrent human save) lands and is recorded as the synced version. - expect(await applyMarkdownToLiveFileDoc('file-1', '# durable', { version: 100 })).toBe( - 'applied' - ) - - // A streaming snapshot built from an OLDER base (50) — copilot loaded the file before that durable - // write — is stale: applying it would diff the live doc back toward the copilot content and clobber - // the durable write, which a later persist would then write over the file. - expect( - await applyMarkdownToLiveFileDoc('file-1', '# stale-base stream', { baseVersion: 50 }) - ).toBe('stale') - - // A streaming snapshot whose base IS the current durable version applies — nothing newer to clobber. - expect( - await applyMarkdownToLiveFileDoc('file-1', '# current-base stream', { baseVersion: 100 }) - ).toBe('applied') - - // ...and a streaming merge is never recorded as the synced version: a later durable write at 150 still - // applies (only durable writes move the synced version; the final edit_content write reconciles). - expect(await applyMarkdownToLiveFileDoc('file-1', '# durable again', { version: 150 })).toBe( - 'applied' - ) - }) - it('serializes concurrent merges for the same file (second waits for the first)', async () => { mockFetchFileDocSeed.mockResolvedValue(seedResult('# Original')) const { io } = createIo() @@ -715,23 +713,33 @@ describe('setupWorkspaceFileDocHandlers', () => { expect(b.socket.join).toHaveBeenCalledWith(ROOM_NAME) }) - it('clears a departed caret when a socket rejoins the room with a new client id', async () => { + it('a socket owns MULTIPLE client ids (co-mounted providers) and relays awareness for each', async () => { + // The shared workspace socket hosts one provider per collaborative view, so the chat file preview + // and the standalone Files editor for the same file each JOIN with their own Yjs client id over ONE + // socket. Ownership is per client id: BOTH announcements must relay. (The old one-owner-per-socket + // model let the later JOIN overwrite the earlier, dropping its awareness — which broke the + // single-writer agent-stream election, letting a peer also self-elect and duplicate streamed text.) const { io, sent } = createIo() - const { frame: awFrame } = awarenessFrame(500, 'A') const a = setup('socket-a', io) await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 500 }) - a.handlers[FILE_DOC_EVENTS.MESSAGE](awFrame) - sent.length = 0 - await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 501 }) - - // The old client (500) caret removal is broadcast to the room. - const removal = sent.find( - (m) => - m.event === FILE_DOC_EVENTS.MESSAGE && - (m.payload as Uint8Array)[0] === FILE_DOC_MESSAGE_TYPE.AWARENESS - ) - expect(removal).toBeDefined() + expect(joinSuccessFileId(a.socket)).toBe('file-1') + + const relayedFor = (clientId: number) => { + sent.length = 0 + a.handlers[FILE_DOC_EVENTS.MESSAGE](awarenessFrame(clientId, `c${clientId}`).frame) + return sent.find( + (m) => + m.event === FILE_DOC_EVENTS.MESSAGE && + (m.payload as Uint8Array)[0] === FILE_DOC_MESSAGE_TYPE.AWARENESS + ) + } + // The FIRST provider's client id (500) is still owned after the second joins — its awareness relays. + expect(relayedFor(500)).toBeDefined() + // The second provider's client id (501) relays too. + expect(relayedFor(501)).toBeDefined() + // A client id this socket does NOT own is still dropped (ownership is not blanket-allowed). + expect(relayedFor(999)).toBeUndefined() }) it('preserves the existing caret when a rebind to a foreign client id is rejected', async () => { diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 039e2ab6661..a907cbb9cda 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/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' @@ -85,12 +90,16 @@ const MERGE_LOCK_RETRIES = Math.ceil( (MERGE_LOCK_TTL_MS + FILE_DOC_TIMEOUTS.mergeRequestMs) / MERGE_LOCK_RETRY_MS ) -/** A socket's presence ownership within a room. */ +/** One presence ownership within a room: a (socket, clientID) pair. */ interface FileDocOwner { /** - * The awareness clientID the socket declared at join. It owns exactly this one - * and may only publish/remove awareness for it, so an authenticated peer cannot - * forge or clear another collaborator's presence. + * An awareness clientID this socket declared at join. The socket may only publish/remove awareness + * for a clientID it owns, so an authenticated peer cannot forge or clear another collaborator's + * presence. A single socket can own SEVERAL clientIDs at once — the shared workspace socket hosts one + * provider per mounted collaborative view, so e.g. the chat file preview and the standalone Files + * editor for the same file each bind their own Yjs clientID over the one socket. The election that + * picks a single agent-stream writer depends on every such provider's awareness propagating, so + * ownership is tracked per clientID, not one-per-socket (which would drop the later joiner's frames). */ clientId: number /** The owning user — used to tell a reconnect (same user reusing its Yjs client @@ -107,8 +116,9 @@ interface FileDocRoom { fileId: string doc: Y.Doc awareness: awarenessProtocol.Awareness - /** socketId → its presence ownership. */ - owners: Map + /** socketId → (clientId → its presence ownership). A socket owns one entry per collaborative provider + * it mounted for this file (see {@link FileDocOwner}); an empty inner map is never kept. */ + owners: Map> /** True once the server-side seed fetch has started, so concurrent joins don't each fetch. * Reset on a fetch FAILURE so a later join can retry (a genuinely empty file stays empty). */ serverSeedStarted: boolean @@ -177,6 +187,17 @@ 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). A non-string sentinel, so `originSocketId` returns `null` for it and the update never triggers + * `edited`/`schedulePersist` (the copilot's final `edit_content` write is the durable persist). Unlike a + * client edit, an agent frame is broadcast to the WHOLE room (its originating socket is NOT excluded), so a + * second {@link FileDocProvider} on the same socket — e.g. the chat preview alongside the Files editor — + * also receives the mid-stream ops. The emitting provider no-ops on its own echo (the ops are already + * applied locally), so broadcasting back to the sender is harmless. + */ +const AGENT_SYNC_ORIGIN = Symbol('file-doc-agent-sync') + /** * 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 @@ -349,7 +370,12 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr */ function broadcastFileDocPresence(io: Server, name: string, room: FileDocRoom) { const users: FileDocPresenceUser[] = [] - for (const [socketId, owner] of room.owners) { + // One entry PER SOCKET (session), not per clientID: a socket's several providers are the same + // authenticated user, so any of its owners carries the identity; the client dedupes per user for the + // avatar stack (see the roster comment above). An empty inner map is never stored, so `owner` exists. + for (const [socketId, clientMap] of room.owners) { + const owner = clientMap.values().next().value + if (!owner) continue users.push({ socketId, userId: owner.userId, @@ -517,12 +543,11 @@ const fileDocMergeChains = new Map>() /** * How a merge is positioned on the file's version line — mirrors the sim-side `LiveFileDocMergeOrder` - * wire fields. A durable `version` is checked AND recorded; a streaming `baseVersion` (the durable version - * the snapshot was built from) is checked only — dropped if a newer durable write has since landed. + * wire field. A durable `version` is checked (applied only if newer than the doc's current version) AND + * recorded as the synced version. */ interface MergeOrder { version?: number - baseVersion?: number } /** @@ -569,16 +594,14 @@ async function mergeMarkdownIntoRoom( name: string, fileId: string, markdown: string, - { version, baseVersion }: MergeOrder + { version }: MergeOrder ): Promise<'applied' | 'no-live-room' | 'merge-unavailable' | 'stale'> { const store = getFileDocStore() // The durable version this merge carries is now incorporated in the live doc — record it (cluster-wide // in Redis for multi-task, plus this task's room) so the persist If-Match guard treats this write as // synced rather than an out-of-band conflict. AWAITED so the version is durable before the merge lock - // releases, so the next lock holder's staleness check (below) reads a consistent value. Only a durable - // `version` is recorded — a streaming `baseVersion` orders the merge (below) but is never a checkpoint, - // so the synced version stays pinned to the last durable write. + // releases, so the next lock holder's staleness check (below) reads a consistent value. const recordVersion = async () => { if (version === undefined) return const room = fileDocRooms.get(name) @@ -590,29 +613,13 @@ async function mergeMarkdownIntoRoom( } // Order this merge on the file's version line, where `current` is the durable version the doc already - // incorporates. Both keys are DB-monotonic `contentUpdatedAt` values (no wall-clock), so ordering is - // immune to clock skew: - // - A durable `version` is stale if it is NOT strictly newer than `current` — a newer durable write - // already landed (possibly on another process, out of dispatch order); applying its older markdown - // would regress the doc while the monotonic token stays high. - // - A streaming `baseVersion` (the durable version the snapshot was built from) is stale if `current` - // has moved PAST it — a newer durable write landed since the snapshot's base, so diffing the live - // doc back toward the snapshot would clobber that write's content (which a later persist, still - // holding the current If-Match token, would then write over the durable file). This is what stops a - // concurrent human edit from being silently lost; the per-process caller chain cannot see it. - // A merge with neither key is never stale (legacy, unordered). Only a durable `version` is recorded, - // so a streaming snapshot never advances the synced version — the final `edit_content` write does. - // - // Known, accepted limitation: two INDEPENDENT copilot streams editing the SAME file at once share one - // base version, so neither is stale relative to the other and their snapshots can interleave in the live - // doc. This is transient only — each stream's final durable write is version-ordered and reconciles the - // doc, so the steady state is deterministic (last durable wins) and the durable file is never corrupted. - // Ordering two independent snapshot streams would need a shared sequence they don't have; the fully - // robust form (a per-file streaming lease, or embedding the version in each stream entry) is a scoped - // follow-up, not a durability fix owed here. + // incorporates. `version` is a DB-monotonic `contentUpdatedAt` value (no wall-clock), so ordering is + // immune to clock skew: a durable `version` is stale if it is NOT strictly newer than `current` — a + // newer durable write already landed (possibly on another process, out of dispatch order); applying its + // older markdown would regress the doc while the monotonic token stays high. A merge with no `version` + // is never stale (legacy, unordered). const isStale = (current: number): boolean => { if (version !== undefined) return version <= current - if (baseVersion !== undefined) return current > baseVersion return false } @@ -703,25 +710,42 @@ 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. + // A client edit excludes its own sender socket (echo suppression). An agent frame broadcasts to the + // WHOLE room — no socket excluded — so a same-socket sibling provider (chat preview + Files editor) + // stays live mid-stream; the emitting provider no-ops on its own echo. + broadcastLocal( + io, + name, + encoding.toUint8Array(encoder), + 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: - // in the multi-replica path a copilot merge is NOT excluded — it round-trips through the stream as - // REDIS_ORIGIN, indistinguishable from a peer edit, so it marks `edited`. That only ever causes an - // extra idempotent persist of content copilot already wrote directly (safe over-persist, never a lost - // edit); the single-replica path applies the merge locally with no origin and does not count it. + // 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 ( @@ -785,8 +809,9 @@ function handleMessage(socket: AuthenticatedSocket, data: unknown) { switch (messageType) { case FILE_DOC_MESSAGE_TYPE.SYNC: { - // Attribute a server-side persist of the resulting edit to the actual editor (blob metadata). - const editor = room.owners.get(socket.id)?.userId + // Attribute a server-side persist of the resulting edit to the actual editor (blob metadata). A + // socket's providers are all the same user, so any owner's userId identifies the editor. + const editor = room.owners.get(socket.id)?.values().next().value?.userId if (editor) room.lastEditorUserId = editor const encoder = encoding.createEncoder() encoding.writeVarUint(encoder, FILE_DOC_MESSAGE_TYPE.SYNC) @@ -800,13 +825,29 @@ function handleMessage(socket: AuthenticatedSocket, data: unknown) { } break } + case FILE_DOC_MESSAGE_TYPE.SYNC_NO_PERSIST: { + // An agent-streamed frame: apply + fan out to the room (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 {@link AGENT_SYNC_ORIGIN} (a non-string sentinel) so `originSocketId` is `null` in + // `doc.on('update')` — skipping `edited`/`schedulePersist`, and broadcasting to the WHOLE room + // (including the sender socket, so a same-socket sibling provider stays live). The copilot's final + // `edit_content` write remains the authoritative durable persist. + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, FILE_DOC_MESSAGE_TYPE.SYNC) + syncProtocol.readSyncMessage(decoder, encoder, room.doc, AGENT_SYNC_ORIGIN) + 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 - // for the clientID it bound at join, so a peer cannot spoof or clear - // another collaborator's caret. - const owned = room.owners.get(socket.id)?.clientId - if (owned === undefined || awarenessUpdateClientIds(update).some((id) => id !== owned)) { + // Enforce presence ownership: a socket may only publish/remove awareness for a clientID it bound + // at join, so a peer cannot spoof or clear another collaborator's caret. A socket can own SEVERAL + // clientIDs (one per mounted provider), so the frame is accepted only if EVERY id it carries is + // owned by this socket. + const owned = room.owners.get(socket.id) + if (owned === undefined || awarenessUpdateClientIds(update).some((id) => !owned.has(id))) { logger.warn('Dropping awareness frame for an unowned client id', { socketId: socket.id }) return } @@ -844,12 +885,16 @@ export function cleanupFileDocForSocket(socketId: string, io: Server, endOfLife const room = fileDocRooms.get(name) if (!room) return - const owner = room.owners.get(socketId) + // The socket may own several clientIDs (one per provider it mounted for this file); drop them ALL. + // The client only emits LEAVE / disconnects once its LAST provider for the file tears down, so a + // per-socket cleanup here is correct — an earlier single-provider unmount already cleared its own + // caret via its awareness removal. + const clientMap = room.owners.get(socketId) room.owners.delete(socketId) - if (owner !== undefined) { - // Fires the awareness `update` handler with a non-socket origin → the removal - // is broadcast to every remaining client, so the departed caret vanishes. - awarenessProtocol.removeAwarenessStates(room.awareness, [owner.clientId], null) + if (clientMap !== undefined && clientMap.size > 0) { + // Fires the awareness `update` handler with a non-socket origin → the removals + // are broadcast to every remaining client, so the departed carets vanish. + awarenessProtocol.removeAwarenessStates(room.awareness, [...clientMap.keys()], null) // Refresh the roster for whoever remains (server-authenticated identity). broadcastFileDocPresence(io, name, room) } @@ -951,20 +996,27 @@ export function setupWorkspaceFileDocHandlers( // rejected. This runs BEFORE any teardown of the socket's current binding below, so a // rejected rebind — even during a document switch — leaves the socket's existing document // and caret untouched. - for (const [otherSid, owner] of entry.owners) { - if (owner.clientId !== clientId || otherSid === socket.id) continue + for (const [otherSid, clientMap] of entry.owners) { + if (otherSid === socket.id) continue + const owner = clientMap.get(clientId) + if (owner === undefined) continue if (owner.userId !== userId) { emitJoinError(socket, fileId, 'Client id already in use', 'CLIENT_ID_IN_USE', false) return } - // Fully evict the stale prior socket of the same user — owner + caret AND its room - // mapping + Socket.IO membership — so it can no longer send document (sync) frames: - // handleMessage's SYNC path gates on socketToRoomName, not owners. Done inline rather - // than via cleanupFileDocForSocket, which could destroyRoomIfIdle the room we're joining. - entry.owners.delete(otherSid) - awarenessProtocol.removeAwarenessStates(entry.awareness, [owner.clientId], null) - socketToRoomName.delete(otherSid) - io.in(otherSid).socketsLeave(name) + // Same user reclaiming its client id on a stale prior socket: evict just THAT clientID's binding + // + caret from the old socket. If that leaves the old socket with no providers, also drop its + // room mapping + Socket.IO membership so it can no longer send document (sync) frames + // (handleMessage's SYNC path gates on socketToRoomName, not owners); an old socket that still + // hosts OTHER providers keeps them. Done inline rather than via cleanupFileDocForSocket, which + // could destroyRoomIfIdle the room we're joining. + clientMap.delete(clientId) + awarenessProtocol.removeAwarenessStates(entry.awareness, [clientId], null) + if (clientMap.size === 0) { + entry.owners.delete(otherSid) + socketToRoomName.delete(otherSid) + io.in(otherSid).socketsLeave(name) + } } // Only now that the rebind is guaranteed to succeed, leave a previously-joined document if @@ -976,14 +1028,18 @@ export function setupWorkspaceFileDocHandlers( cleanupFileDocForSocket(socket.id, io) } - // Accepted: a same socket rebinding to a NEW client id clears its old caret - // so it doesn't linger as a ghost after the binding is overwritten. - const previous = entry.owners.get(socket.id) - if (previous !== undefined && previous.clientId !== clientId) { - awarenessProtocol.removeAwarenessStates(entry.awareness, [previous.clientId], null) + // ADD this provider's clientID to the socket's ownership set (do NOT overwrite a sibling provider + // on the same socket — that lone-owner overwrite is exactly what dropped the chat preview's + // awareness when the Files editor co-mounted). A re-JOIN of the same clientID is idempotent. A + // single provider that later unmounts clears its own caret via its awareness removal; the whole + // set is dropped on the socket's LEAVE/disconnect (client emits LEAVE only after its LAST provider + // for the file tears down). + let clientMap = entry.owners.get(socket.id) + if (clientMap === undefined) { + clientMap = new Map() + entry.owners.set(socket.id, clientMap) } - - entry.owners.set(socket.id, { clientId, userId, userName, avatarUrl }) + clientMap.set(clientId, { clientId, userId, userName, avatarUrl }) socketToRoomName.set(socket.id, name) socket.join(name) diff --git a/apps/realtime/src/routes/http.ts b/apps/realtime/src/routes/http.ts index b37e9061107..e81ed3150ad 100644 --- a/apps/realtime/src/routes/http.ts +++ b/apps/realtime/src/routes/http.ts @@ -203,22 +203,22 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) { return } - // Merge a copilot edit into a file's LIVE collaborative document so it streams into open editors - // (Stage C). Returns `{ applied }`: when false, no seeded live room exists and the caller writes - // the file directly instead. Any live user edits are preserved — the app builds a minimal CRDT diff. + // Merge a durable file write into a file's LIVE collaborative document so open editors reconcile to + // it (Stage C) — this is the stream-end/durable reconcile, not token-by-token streaming (that is now + // applied client-side by the open editor). Returns `{ applied }`: when false, no seeded live room + // exists and the caller writes the file directly instead. Live user edits are preserved — the app + // builds a minimal CRDT diff. if (req.method === 'POST' && req.url === '/api/file-doc/apply-edit') { try { const body = await readRequestBody(req) - const { fileId, markdown, version, baseVersion } = JSON.parse(body) + const { fileId, markdown, version } = JSON.parse(body) if (!isNonEmptyString(fileId) || typeof markdown !== 'string') { return sendError(res, 'Invalid fileId or markdown', 400) } // `version` (the durable updatedAt this markdown was written with) records that the live doc now // incorporates that durable version, so the persist If-Match guard won't flag it as a conflict. - // `baseVersion` is a streaming snapshot's causal base: dropped if a newer durable write landed. const result = await applyMarkdownToLiveFileDoc(fileId, markdown, { version: typeof version === 'number' ? version : undefined, - baseVersion: typeof baseVersion === 'number' ? baseVersion : undefined, }) res.writeHead(200, { 'Content-Type': 'application/json' }) res.end(JSON.stringify({ applied: result === 'applied' })) diff --git a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/file-viewer.tsx b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/file-viewer.tsx index bba37374240..006012191ac 100644 --- a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/file-viewer.tsx +++ b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/file-viewer.tsx @@ -108,6 +108,7 @@ interface FileViewerProps { streamingContent?: string isAgentEditing?: boolean streamIsIncremental?: boolean + streamOperation?: string disableStreamingAutoScroll?: boolean previewContextKey?: string /** @@ -150,6 +151,7 @@ function FileViewerContent({ streamingContent, isAgentEditing, streamIsIncremental, + streamOperation, disableStreamingAutoScroll = false, previewContextKey, collaborative, @@ -196,6 +198,7 @@ function FileViewerContent({ streamingContent={streamingContent} isAgentEditing={isAgentEditing} streamIsIncremental={streamIsIncremental} + streamOperation={streamOperation} disableStreamingAutoScroll={disableStreamingAutoScroll} previewContextKey={previewContextKey} collaborative={collaborative} diff --git a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/agent-stream-leader.test.ts b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/agent-stream-leader.test.ts new file mode 100644 index 00000000000..59ae7f50649 --- /dev/null +++ b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/agent-stream-leader.test.ts @@ -0,0 +1,55 @@ +/** + * @vitest-environment node + */ +import { describe, expect, it } from 'vitest' +import { Awareness } from 'y-protocols/awareness' +import * as Y from 'yjs' +import { + announceAgentApplying, + clearAgentApplying, + isAgentStreamLeader, +} from './agent-stream-leader' + +/** An Awareness with explicit peer states injected (self, if present, carries no `agentApplying`). */ +function awarenessWith(entries: Array<[number, Record]>): Awareness { + const aw = new Awareness(new Y.Doc()) + const states = aw.getStates() as Map> + for (const [clientId, state] of entries) states.set(clientId, state) + return aw +} + +describe('agent-stream leader election', () => { + it('a sole announcer is the leader', () => { + expect(isAgentStreamLeader(awarenessWith([[5, { agentApplying: true }]]), 5)).toBe(true) + }) + + it('the lowest clientID among announcers leads; higher announcers do not', () => { + const aw = awarenessWith([ + [7, { agentApplying: true }], + [3, { agentApplying: true }], + [9, { user: { name: 'someone else, not applying' } }], + ]) + expect(isAgentStreamLeader(aw, 3)).toBe(true) + expect(isAgentStreamLeader(aw, 7)).toBe(false) + }) + + it('a client that is not announcing is never the leader', () => { + expect(isAgentStreamLeader(awarenessWith([[3, { agentApplying: true }]]), 8)).toBe(false) + }) + + it('with no announcers, nobody leads', () => { + expect(isAgentStreamLeader(awarenessWith([[3, { user: {} }]]), 3)).toBe(false) + }) + + it('announce makes self the leader; clear relinquishes it', () => { + const doc = new Y.Doc() + const aw = new Awareness(doc) + announceAgentApplying(aw) + expect(aw.getLocalState()?.agentApplying).toBe(true) + expect(isAgentStreamLeader(aw, doc.clientID)).toBe(true) + + clearAgentApplying(aw) + expect(aw.getLocalState()?.agentApplying ?? null).toBeNull() + expect(isAgentStreamLeader(aw, doc.clientID)).toBe(false) + }) +}) diff --git a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/agent-stream-leader.ts b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/agent-stream-leader.ts new file mode 100644 index 00000000000..8cc79f790ac --- /dev/null +++ b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/agent-stream-leader.ts @@ -0,0 +1,37 @@ +import type { Awareness } from 'y-protocols/awareness' + +/** + * Awareness field a collaborative client sets on its OWN state while it is applying an agent stream into + * the shared doc. Read by every peer to run the single-writer election below. Coexists with the caret + * `user` field (`setLocalStateField` writes one field without clobbering others). + */ +const AGENT_APPLYING_FIELD = 'agentApplying' + +/** Announce that this client is applying an agent stream (candidate in the leader election). */ +export function announceAgentApplying(awareness: Awareness): void { + awareness.setLocalStateField(AGENT_APPLYING_FIELD, true) +} + +/** Stop announcing (this client is no longer applying an agent stream). */ +export function clearAgentApplying(awareness: Awareness): void { + awareness.setLocalStateField(AGENT_APPLYING_FIELD, null) +} + +/** + * Single-writer election: exactly one collaborative client applies a given agent stream into the shared + * doc, so N tabs/windows watching the same live copilot stream don't each insert it under a different + * Yjs clientID and duplicate the content. The leader is the MINIMUM clientID among all clients currently + * announcing (via {@link announceAgentApplying}) that they are applying — a deterministic tie-break that + * needs no coordinator. A brief startup race (before an announcement propagates to peers) is bounded to a + * frame or two — self-corrected the moment awareness converges, and reconciled anyway by the durable + * server write. In the common single-client case the caller is the only announcer, so it always leads. + */ +export function isAgentStreamLeader(awareness: Awareness, selfClientId: number): boolean { + let leader = Number.POSITIVE_INFINITY + awareness.getStates().forEach((state, clientId) => { + if ((state as Record | undefined)?.[AGENT_APPLYING_FIELD] === true) { + leader = Math.min(leader, clientId) + } + }) + return leader === selfClientId +} diff --git a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/apply-streamed-markdown.test.ts b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/apply-streamed-markdown.test.ts new file mode 100644 index 00000000000..8b4affc0843 --- /dev/null +++ b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/apply-streamed-markdown.test.ts @@ -0,0 +1,189 @@ +/** + * @vitest-environment jsdom + */ +import { Editor } from '@tiptap/core' +import { initProseMirrorDoc, updateYFragment, ySyncPluginKey } from '@tiptap/y-tiptap' +import { afterEach, beforeAll, describe, expect, it } from 'vitest' +import { Awareness } from 'y-protocols/awareness' +import * as Y from 'yjs' +import { createMarkdownEditorExtensions } from '../editor-extensions' +import { applyAgentStreamFrame, beginAgentStream, endAgentStream } from './apply-streamed-markdown' + +beforeAll(() => { + // jsdom does not implement elementFromPoint; the Placeholder extension's viewport tracking calls it + // on view mount. Returning null makes ProseMirror's posAtCoords fall back gracefully. + if (!document.elementFromPoint) { + document.elementFromPoint = () => null + } +}) + +/** A headless collaborative editor bound to a fresh Y.Doc — the same extension wiring the component uses. */ +function makeCollabEditor() { + const doc = new Y.Doc() + const awareness = new Awareness(doc) + const editor = new Editor({ + extensions: createMarkdownEditorExtensions({ + placeholder: '', + collaboration: { + doc, + awareness, + user: { name: 'Tester', color: '#ffffff', clientId: doc.clientID }, + }, + }), + content: '', + }) + return { editor, doc, awareness } +} + +const teardown: Array<() => void> = [] +afterEach(() => { + for (const fn of teardown.splice(0)) fn() +}) + +function track(t: { editor: Editor; doc: Y.Doc; awareness: Awareness }) { + teardown.push(() => { + t.editor.destroy() + t.awareness.destroy() + t.doc.destroy() + }) + return t +} + +describe('agent-stream applier', () => { + it('relies on y-tiptap internals that still exist (upgrade guardrail)', () => { + // beginAgentStream/applyAgentStreamFrame reach into y-tiptap internals (not public TipTap API): + // `ySyncPluginKey`, `updateYFragment`, `initProseMirrorDoc`. A y-tiptap bump that renames or drops + // any of them can pass typecheck yet break at runtime — assert their runtime shape here so an upgrade + // fails loudly at test time instead of in production. Pinned to an exact y-tiptap version in + // package.json; bump that pin and this guard together. + expect(typeof updateYFragment).toBe('function') + expect(typeof initProseMirrorDoc).toBe('function') + expect(ySyncPluginKey).toBeDefined() + expect(typeof ySyncPluginKey.getState).toBe('function') + }) + + it('streams agent content into the live collaborative doc and broadcasts it as Yjs ops', () => { + const { editor, doc } = track(makeCollabEditor()) + + const session = beginAgentStream(editor) + expect(session).not.toBeNull() + expect(applyAgentStreamFrame(editor, session!, '# Title\n\nHello world.')).toBe(true) + expect(editor.getText()).toContain('Hello world') + + // The write lands as ops on the shared doc, so any peer receives it (this is what makes a + // collaborator on /files see the stream without ever holding `streamingContent`). + const peer = new Y.Doc() + Y.applyUpdate(peer, Y.encodeStateAsUpdate(doc)) + expect(peer.getXmlFragment('default').toString()).toContain('Hello world') + peer.destroy() + endAgentStream(session!) + }) + + it('beginAgentStream returns null when the editor has no ySync binding', () => { + // A plain editor with no collaboration has no ySync binding, so a stream cannot start against it. + const editor = new Editor({ + extensions: createMarkdownEditorExtensions({ placeholder: '' }), + content: '', + }) + teardown.push(() => editor.destroy()) + expect(beginAgentStream(editor)).toBeNull() + }) + + it('keeps agent-streamed ops out of the undo stack while user edits stay undoable', () => { + const { editor } = track(makeCollabEditor()) + + const session = beginAgentStream(editor)! + applyAgentStreamFrame(editor, session, '# Streamed\n\nAgent wrote this.') + endAgentStream(session) + // The streamed op relayed under a non-`ySyncPluginKey` origin, which the Collaboration UndoManager + // does not track — so there is nothing to undo, and an undo must not revert the agent's content. + expect(editor.can().undo()).toBe(false) + editor.commands.undo() + expect(editor.getText()).toContain('Agent wrote this') + + // A genuine user edit IS captured (origin ySyncPluginKey) — proving the test isn't vacuous: + // undo works, and it reverts only the user edit, leaving the agent content intact. + editor.commands.focus('end') + editor.commands.insertContent(' USER-TYPED') + expect(editor.getText()).toContain('USER-TYPED') + expect(editor.can().undo()).toBe(true) + editor.commands.undo() + expect(editor.getText()).not.toContain('USER-TYPED') + 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. + const { editor, doc } = track(makeCollabEditor()) + + const session = beginAgentStream(editor)! + applyAgentStreamFrame(editor, session, 'Alpha paragraph.\n\nBeta paragraph.') + + // A peer edits the FIRST paragraph directly on the shared doc — the agent's later snapshot still + // carries the ORIGINAL first paragraph (it was built from the base, before this edit). + const peer = new Y.Doc() + Y.applyUpdate(peer, Y.encodeStateAsUpdate(doc)) + const peerFrag = peer.getXmlFragment('default') + peer.transact(() => { + const firstPara = peerFrag.get(0) as Y.XmlElement + const textNode = firstPara.get(0) as Y.XmlText + textNode.insert(textNode.toString().length, ' EDITED') + }) + Y.applyUpdate(doc, Y.encodeStateAsUpdate(peer, Y.encodeStateVector(doc))) + peer.destroy() + + // The agent appends a third paragraph. Its snapshot's first paragraph is the stale original, but the + // shadow-relayed delta only inserts the new paragraph — so the peer's " EDITED" must survive. + applyAgentStreamFrame( + editor, + session, + 'Alpha paragraph.\n\nBeta paragraph.\n\nGamma paragraph.' + ) + endAgentStream(session) + + const live = doc.getXmlFragment('default').toString() + expect(live).toContain('EDITED') + expect(live).toContain('Gamma paragraph') + }) +}) diff --git a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/apply-streamed-markdown.ts b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/apply-streamed-markdown.ts new file mode 100644 index 00000000000..f0a69769b2f --- /dev/null +++ b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/apply-streamed-markdown.ts @@ -0,0 +1,78 @@ +import type { Editor } from '@tiptap/core' +import { Node as PMNode } from '@tiptap/pm/model' +import { initProseMirrorDoc, updateYFragment, ySyncPluginKey } from '@tiptap/y-tiptap' +import * as Y from 'yjs' +import { parseMarkdownToDoc } from '../markdown-parse' + +/** The Yjs fragment name TipTap's Collaboration extension binds to (its default `field`). */ +const COLLAB_DOC_FIELD = 'default' + +/** + * Transaction origin for agent-streamed writes into a live collaborative doc. It is deliberately NOT + * 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. + */ +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 + * TRUE peer: only the agent's own delta reaches the shared doc, never a whole-document reconcile that + * would revert a collaborator's concurrent edit. Seeded from the live doc at stream start; it receives + * ONLY agent reconciles (never peer updates), so `shadow → nextTarget` yields exactly the agent's change. + */ +export interface AgentStreamSession { + shadow: Y.Doc + fragment: Y.XmlFragment +} + +/** + * Begin an agent stream by snapshotting the live doc into a private shadow replica. Returns `null` when + * the editor has no live ySync binding (e.g. a non-collaborative editor). + */ +export function beginAgentStream(editor: Editor): AgentStreamSession | null { + const binding = ySyncPluginKey.getState(editor.state)?.binding + if (!binding) return null + const shadow = new Y.Doc() + Y.applyUpdate(shadow, Y.encodeStateAsUpdate(binding.doc)) + return { shadow, fragment: shadow.getXmlFragment(COLLAB_DOC_FIELD) } +} + +/** + * Apply one streamed markdown body. Reconciles the shadow toward `body` with `updateYFragment` (the same + * minimal-diff primitive TipTap runs per keystroke), captures ONLY the resulting agent delta, and relays + * it into the live doc under {@link AGENT_STREAM_ORIGIN}. Because the shadow never sees peer updates, the + * delta touches only what the agent changed — so concurrent peer edits elsewhere in the live doc survive, + * the change renders locally (via the binding's observer, the remote-edit path), broadcasts to every + * peer, and stays out of the user's undo stack. Returns `false` when the editor has no live ySync binding. + */ +export function applyAgentStreamFrame( + editor: Editor, + session: AgentStreamSession, + body: string +): boolean { + const binding = ySyncPluginKey.getState(editor.state)?.binding + if (!binding) return false + const target = PMNode.fromJSON(editor.schema, parseMarkdownToDoc(body)) + let delta: Uint8Array | null = null + const capture = (update: Uint8Array, origin: unknown) => { + if (origin === AGENT_STREAM_ORIGIN) delta = update + } + session.shadow.on('update', capture) + try { + session.shadow.transact(() => { + // `updateYFragment` diffs against the fragment's CURRENT content, so it needs the fragment↔PM + // binding metadata; `initProseMirrorDoc` reconstructs it from the fragment's present state. + const { meta } = initProseMirrorDoc(session.fragment, editor.schema) + updateYFragment(session.shadow, session.fragment, target, meta) + }, AGENT_STREAM_ORIGIN) + } finally { + session.shadow.off('update', capture) + } + if (delta) Y.applyUpdate(binding.doc, delta, AGENT_STREAM_ORIGIN) + return true +} + +/** End an agent stream and free its shadow replica. */ +export function endAgentStream(session: AgentStreamSession): void { + session.shadow.destroy() +} diff --git a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/file-doc-provider.test.ts b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/file-doc-provider.test.ts index af4346011ff..689c2ee6c77 100644 --- a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/file-doc-provider.test.ts +++ b/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' }) diff --git a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/file-doc-provider.ts b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/file-doc-provider.ts index d23b3b21d49..9b55ac7b907 100644 --- a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/file-doc-provider.ts +++ b/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 { 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)) } diff --git a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/rich-markdown-editor.tsx b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/rich-markdown-editor.tsx index f2a1eaad79b..a5ce37498ad 100644 --- a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/rich-markdown-editor.tsx +++ b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/rich-markdown-editor.tsx @@ -20,6 +20,17 @@ import type { SaveStatus } from '@/hooks/use-autosave' import { useFileContentSource } from '@/hooks/use-file-content-source' import { PreviewLoadingFrame } from '../preview-shared' import { useEditableFileContent } from '../use-editable-file-content' +import { + announceAgentApplying, + clearAgentApplying, + isAgentStreamLeader, +} from './collaboration/agent-stream-leader' +import { + type AgentStreamSession, + applyAgentStreamFrame, + beginAgentStream, + endAgentStream, +} from './collaboration/apply-streamed-markdown' import { useFileDocCollaboration } from './collaboration/use-file-doc-collaboration' import { createMarkdownEditorExtensions } from './editor-extensions' import { findHeadingPos } from './heading-anchors' @@ -78,15 +89,22 @@ interface RichMarkdownEditorProps { * applied live; a rebuild is only revealed while it extends what's shown (see the streaming tick). */ streamIsIncremental?: boolean + /** + * The agent edit operation driving the stream, when known (`create`/`append`/`update`/`patch`). In the + * collaborative path it decides only whether to stream mid-flight: an `update` (from-scratch rewrite) is + * HELD until settle so the open doc doesn't collapse to a partial result, while `append`/`patch`/`create` + * apply each frame. + */ + streamOperation?: string disableStreamingAutoScroll?: boolean previewContextKey?: string /** Disable the `@` tag-insertion menu (existing tags still render). Defaults off — the file editor keeps tagging. */ disableTagging?: boolean /** - * Opt this surface into live collaborative editing. Set only by the Files page — - * the dedicated editing surface, which never streams agent output. The agent/Chat - * surface leaves it off, so collaboration and agent-streaming are disjoint by - * construction (they cannot both drive one editor and corrupt the shared doc). + * Opt this surface into live collaborative editing (Files page + the embedded chat file preview). + * Collaboration can coexist with agent streaming: while streaming, the growing content is applied to + * the shared Y.Doc as minimal CRDT diffs (see {@link applyAgentStreamFrame}) rather than a + * full-document `setContent`, so the stream stays smooth and every peer sees it live. */ collaborative?: boolean /** @@ -110,6 +128,7 @@ export const RichMarkdownEditor = memo(function RichMarkdownEditor({ streamingContent, isAgentEditing, streamIsIncremental, + streamOperation, disableStreamingAutoScroll = false, previewContextKey, disableTagging, @@ -180,6 +199,7 @@ export const RichMarkdownEditor = memo(function RichMarkdownEditor({ userName={userName} autoFocus={autoFocus} streamIsIncremental={streamIsIncremental} + streamOperation={streamOperation} disableStreamingAutoScroll={disableStreamingAutoScroll} disableTagging={disableTagging} collaborative={collaborative} @@ -205,6 +225,8 @@ interface LoadedRichMarkdownEditorProps { autoFocus?: boolean /** See {@link RichMarkdownEditorProps.streamIsIncremental}. */ streamIsIncremental?: boolean + /** See {@link RichMarkdownEditorProps.streamOperation}. */ + streamOperation?: string disableStreamingAutoScroll?: boolean disableTagging?: boolean /** See {@link RichMarkdownEditorProps.collaborative}. */ @@ -238,6 +260,7 @@ export function LoadedRichMarkdownEditor({ userName, autoFocus, streamIsIncremental, + streamOperation, disableStreamingAutoScroll, disableTagging, collaborative = false, @@ -257,9 +280,11 @@ export function LoadedRichMarkdownEditor({ /** * Collaboration is decided once at mount from synchronously-available inputs * (`settledRef` is set just above) via `useState`-init, and never changes — TipTap - * fixes the extension set at editor creation, so it cannot turn on later. Enabled - * only on a `collaborative` surface (the Files page, which never streams) for an - * editable, round-trip-safe, non-streaming workspace document with a known user. + * fixes the extension set at editor creation, so it cannot turn on later. Enabled on a + * `collaborative` surface (the Files page or the embedded chat file preview) for an editable, + * round-trip-safe workspace document with a known user, as long as it is not ALREADY streaming at + * mount (`!streamingAtMountRef.current`). An agent stream that begins AFTER mount is applied as CRDT + * diffs into the live doc, so collaboration and streaming coexist (see the streaming effect below). */ const [collaborationEnabled] = useState( () => @@ -318,6 +343,14 @@ export function LoadedRichMarkdownEditor({ const lastSyncedBodyRef = useRef( streamingAtMountRef.current ? null : splitFrontmatter(content).body ) + /** + * The body the AGENT last applied into the collaborative doc — a dedup guard for the collab streaming + * tick, so an unchanged frame skips a redundant shadow reconcile/reparse. Written ONLY by the streaming + * tick (never by `onUpdate`), and reset to `null` on settle for the next stream. It is NOT a string-prefix + * baseline: the mid-stream hold is decided by operation (`update` waits for settle), not by comparing the + * raw preview against the editor's canonical markdown. + */ + const lastStreamedBodyRef = useRef(null) const onChangeRef = useRef(onChange) onChangeRef.current = onChange const onSaveShortcutRef = useRef(onSaveShortcut) @@ -353,6 +386,12 @@ export function LoadedRichMarkdownEditor({ */ const streamIsIncrementalRef = useRef(streamIsIncremental) streamIsIncrementalRef.current = streamIsIncremental + const streamOperationRef = useRef(streamOperation) + streamOperationRef.current = streamOperation + /** The live agent-stream shadow replica, held for the current stream and freed on settle/unmount. */ + const agentStreamSessionRef = useRef(null) + /** True once this client has announced candidacy in the agent-stream election for the current stream. */ + const agentAnnouncedRef = useRef(false) const router = useRouter() const routerRef = useRef(router) routerRef.current = router @@ -751,9 +790,10 @@ export function LoadedRichMarkdownEditor({ }, [collaboration, editor, onCollabReadyChange, setCollabReady]) /** - * Owns editability for the collaborative lifecycle: `useEditor`'s `editable` is only - * the initial value, and the streaming/settle effect stays inert in collab mode — so - * re-apply here whenever collaboration readiness (synced + seeded) flips `isEditable`. + * Owns editability for the collaborative lifecycle: `useEditor`'s `editable` is only the initial + * value, and the streaming/settle effect only moves content in collab mode (never toggles + * editability) — so re-apply here whenever collaboration readiness (synced + seeded) or an agent + * stream flips `isEditable`. */ useEffect(() => { if (!editor || !collaborationEnabled) return @@ -800,13 +840,6 @@ export function LoadedRichMarkdownEditor({ const pendingCollapseRef = useRef(false) useEffect(() => { if (!editor) return - // Collaboration and agent-streaming are disjoint surfaces: collaboration is enabled - // only on the Files page, which never streams agent output, so in collab mode this - // reconcile loop stays fully inert. It must not run even defensively — its - // `setContent` would sync a full-document replace into the shared Y.Doc (the - // ySyncPlugin writes it regardless of `emitUpdate: false`), wiping peers' edits. - // Editability is owned by the reactive effect above. - if (collaborationEnabled) return const syncEditorBody = (body: string) => { if (body === lastSyncedBodyRef.current) return lastSyncedBodyRef.current = body @@ -831,6 +864,146 @@ export function LoadedRichMarkdownEditor({ mutate() }) } + // Collaborative surface: stream by applying a minimal CRDT diff into the live Y.Doc each frame + // (never `setContent`, which would replace the shared doc and wipe peers). Each diff renders + // locally and broadcasts to every peer, so the stream is smooth here and on other clients (e.g. + // the standalone Files page) alike. This branch moves content only — editability for the collab + // lifecycle is owned by the reactive effect above. + if (collaborationEnabled) { + if (isStreaming) { + wasStreamingRef.current = true + // Apply streamed diffs only after the shared doc has SEEDED (synced + seed flag, i.e. + // `collabReady`). Applying onto an unseeded (empty) doc would let the later seed CRDT-merge into + // it — transient garble. In the common case the doc is long seeded before an agent edit begins; + // in the rare stream-before-seed race we wait, and since `collabReady` is an effect dep this + // re-runs and applies once it lands (the read-only placeholder shows the base content meanwhile — + // see `showPlaceholder`). + if (!collabReady) return + // Announce candidacy in the single-writer election (see the tick) so only one tab/window applies + // this stream. The shadow is opened lazily in the tick, only when THIS client actually leads — so a + // non-leader builds none, and a client that takes leadership mid-stream (a handoff) seeds its shadow + // from the CURRENT doc, already carrying the prior leader's ops, never a stale base. + if (!agentAnnouncedRef.current) { + agentAnnouncedRef.current = true + if (collaboration) announceAgentApplying(collaboration.awareness) + } + const body = splitFrontmatter(content).body + if (body === lastStreamedBodyRef.current) return + pendingStreamBodyRef.current = body + if (streamRafRef.current !== null) return + const tick = () => { + const pending = pendingStreamBodyRef.current + if (pending === null || pending === lastStreamedBodyRef.current) { + streamRafRef.current = null + return + } + // Hold a from-scratch rewrite (`update`) until settle so the open doc doesn't collapse to a + // partial rewrite mid-stream (matching `main`). `append`/`patch`/`create` apply each frame — the + // shadow reconcile is peer-safe, and base-less `append` fragments no longer reach the client (the + // server fail-closes them), so there is nothing here to string-prefix or wipe-guard against. + if (streamOperationRef.current === 'update') { + streamRafRef.current = null + return + } + // Single-writer election: only the leader (min clientID among clients announcing they apply this + // stream) writes it into the shared doc, so multiple tabs/windows watching the same live copilot + // stream don't each insert it and duplicate content. A non-leader renders the leader's ops via + // Yjs; re-checked each frame, so a co-leader stops the moment awareness propagates. (The pick-up + // direction — a successor beginning to write after the leader tab closes — waits for the next + // content frame to run a tick; a stream that already delivered its last frame is covered by + // settle and the durable write, so at worst a brief end-of-stream display lag, never a loss.) + // Bounded residual (accepted): if two tabs start the SAME stream within the awareness-propagation + // window they briefly both lead and duplicate a frame or two — a rare, transient, never-persisted + // glitch (SYNC_NO_PERSIST keeps it out of storage; the durable edit_content write reconciles the + // final doc). Resumes are sequential (the second tab sees the first's announcement), so the common + // multi-tab case elects cleanly. + if ( + 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 + } + if ( + pending.length > STREAM_REPARSE_THROTTLE_THRESHOLD && + performance.now() - lastStreamParseAtRef.current < STREAM_REPARSE_THROTTLE_MS + ) { + streamRafRef.current = requestAnimationFrame(tick) + return + } + 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. 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)) { + streamRafRef.current = null + return + } + streamRafRef.current = null + lastStreamedBodyRef.current = pending + lastStreamParseAtRef.current = performance.now() + if (!disableStreamingAutoScroll && el && pinnedToBottom) el.scrollTop = el.scrollHeight + } + streamRafRef.current = requestAnimationFrame(tick) + return + } + if (streamRafRef.current !== null) { + cancelAnimationFrame(streamRafRef.current) + streamRafRef.current = null + } + // Settle: apply the FINAL body so the Y.Doc exactly equals the streamed result — but ONLY the + // elected writer applies it (the same min-clientID election the streaming tick uses). Without this, + // N tabs watching one run each open a fresh shadow and reconcile current→final; a non-leader's local + // settle microtask runs BEFORE the leader's final propagates (a server round-trip), so both insert + // the same tail and Yjs keeps both (it does not dedupe identical text from two clients) → a + // duplicated tail. The election is reliable here (unlike the bounded startup window): the stream ran + // for seconds, so awareness is long converged. Each tab reads leadership BEFORE clearing its own + // announcement — a remote clear is a network round-trip, always slower than these local microtasks, + // so every tab sees the same announcer set and agrees on one leader. The leader reuses its + // up-to-date shadow (catching a throttled last frame) or, if it never applied mid-stream (a held + // `update`, or a pre-seed stream), opens a FRESH shadow from the current doc; a non-leader applies + // nothing (the leader's final broadcasts to it) and frees any shadow it still held. The durable + // `edit_content` write then lands as a noop diff for everyone. + if (wasStreamingRef.current && collabReady) { + wasStreamingRef.current = false + agentAnnouncedRef.current = false + const isSettleWriter = + !collaboration || isAgentStreamLeader(collaboration.awareness, collaboration.doc.clientID) + if (collaboration) clearAgentApplying(collaboration.awareness) + lastStreamedBodyRef.current = null + const heldSession = agentStreamSessionRef.current + agentStreamSessionRef.current = null + if (isSettleWriter) { + const finalBody = splitFrontmatter(content).body + const session = heldSession ?? beginAgentStream(editor) + if (session) { + runOffRender(() => applyAgentStreamFrame(editor, session, finalBody)) + // Free the shadow with an UNGUARDED microtask (not `runOffRender`): a rapid follow-up stream + // can supersede the run token and drop the apply above, but the shadow must always be + // destroyed. Queued after the apply, so it frees the shadow only once that has had its chance. + queueMicrotask(() => endAgentStream(session)) + } + } else if (heldSession) { + // Non-leader: it never writes the final (the leader does + broadcasts it); free any shadow it held. + queueMicrotask(() => endAgentStream(heldSession)) + } + } + return + } if (isStreaming) { wasStreamingRef.current = true if (editor.isEditable) { @@ -931,14 +1104,21 @@ export function LoadedRichMarkdownEditor({ useEffect( () => () => { if (streamRafRef.current !== null) cancelAnimationFrame(streamRafRef.current) + if (agentStreamSessionRef.current) { + endAgentStream(agentStreamSessionRef.current) + agentStreamSessionRef.current = null + } + lastStreamedBodyRef.current = null + agentAnnouncedRef.current = false }, [] ) - // Show the read-only placeholder only for a plain cold open — never during an agent stream. A stream - // that begins before the doc has seeded fills the (hidden) editor via Yjs, so gating the placeholder - // off while streaming lets that live content show through instead of hiding it behind stale markdown. - const showPlaceholder = collaborationEnabled && !collabReady && !isStreaming + // Show the read-only placeholder (the already-fetched markdown) whenever a collaborative doc has not yet + // seeded — including during an agent stream that begins before the seed lands. Streamed diffs are held + // until `collabReady` (see the streaming effect), so before then the editor is empty; the placeholder + // shows the base content until the seed swaps it in, avoiding both a blank frame and a garbled merge. + const showPlaceholder = collaborationEnabled && !collabReady return (
@@ -663,6 +664,7 @@ interface EmbeddedFileProps { streamingContent?: string isAgentEditing?: boolean streamIsIncremental?: boolean + streamOperation?: string disableStreamingAutoScroll?: boolean previewContextKey?: string } @@ -675,6 +677,7 @@ function EmbeddedFile({ streamingContent, isAgentEditing, streamIsIncremental, + streamOperation, disableStreamingAutoScroll = false, previewContextKey, }: EmbeddedFileProps) { @@ -718,6 +721,7 @@ function EmbeddedFile({ streamingContent={streamingContent} isAgentEditing={isAgentEditing} streamIsIncremental={streamIsIncremental} + streamOperation={streamOperation} disableStreamingAutoScroll={disableStreamingAutoScroll} previewContextKey={previewContextKey} collaborative diff --git a/apps/sim/lib/copilot/request/go/file-preview-adapter.test.ts b/apps/sim/lib/copilot/request/go/file-preview-adapter.test.ts index 8915f2cd472..c9409958477 100644 --- a/apps/sim/lib/copilot/request/go/file-preview-adapter.test.ts +++ b/apps/sim/lib/copilot/request/go/file-preview-adapter.test.ts @@ -1,7 +1,7 @@ /** * @vitest-environment node */ -import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { beforeEach, describe, expect, it, vi } from 'vitest' import { MothershipStreamV1EventType, MothershipStreamV1ToolExecutor, @@ -9,27 +9,10 @@ import { MothershipStreamV1ToolPhase, } from '@/lib/copilot/generated/mothership-stream-v1' -const { mergeEditIntoLiveFileDocMock, isLiveDocMergeInFlightMock } = vi.hoisted(() => ({ - mergeEditIntoLiveFileDocMock: - vi.fn< - ( - fileId: string, - markdown: string, - order?: { version?: number; baseVersion?: number } - ) => Promise - >(), - isLiveDocMergeInFlightMock: vi.fn<(fileId: string) => boolean>(), -})) - const { peekFileIntentMock } = vi.hoisted(() => ({ peekFileIntentMock: vi.fn(), })) -vi.mock('@/lib/realtime/notify', () => ({ - mergeEditIntoLiveFileDoc: mergeEditIntoLiveFileDocMock, - isLiveDocMergeInFlight: isLiveDocMergeInFlightMock, -})) - vi.mock('@/lib/copilot/tools/server/files/file-intent-store', () => ({ peekFileIntent: peekFileIntentMock, })) @@ -46,7 +29,6 @@ import type { ActiveFileIntent, ExecutionContext, StreamEvent } from '@/lib/copi const STREAM_ID = 'stream-1' const EDIT_TOOL_CALL_ID = 'edit-content-1' const WORKSPACE_FILE_TOOL_CALL_ID = 'workspace-file-1' -/** The durable version (`contentUpdatedAt`, epoch ms) the streamed base content is at. */ const BASE_VERSION_MS = 900_000 /** One args_delta chunk of the streamed `edit_content` JSON, as a driveable StreamEvent. */ @@ -91,9 +73,14 @@ const flushMicrotasks = async () => { await Promise.resolve() } -describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { +/** + * The copilot preview adapter no longer merges the growing content into the file's live collaborative + * Y.Doc server-side — that is done client-side by the open editor as minimal CRDT diffs (see + * `applyStreamedMarkdownToLiveDoc`). These tests guard the surviving contract: the adapter still emits + * the growing `file_preview_content` events that drive the chat's inline preview. + */ +describe('processFilePreviewStreamEvent — preview content emission', () => { let state: FilePreviewAdapterState - let nowMs: number const execContext: ExecutionContext = { userId: 'user-1', workflowId: 'workflow-1', @@ -101,162 +88,55 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { chatId: 'chat-1', messageId: 'msg-1', } + const events: Array<{ payload: Record }> = [] beforeEach(() => { vi.clearAllMocks() - mergeEditIntoLiveFileDocMock.mockResolvedValue(undefined) - isLiveDocMergeInFlightMock.mockReturnValue(false) - // Default: an append/patch base is available (a non-empty file) at durable version BASE_VERSION_MS, - // so the base-present gate passes and the streaming merge carries that base version. + events.length = 0 + // An append base is available (a non-empty file) at durable version BASE_VERSION_MS, so the preview + // text is composed as base + streamed content. peekFileIntentMock.mockResolvedValue({ existingContent: 'Base.', fileRecord: { contentUpdatedAt: new Date(BASE_VERSION_MS) }, }) state = createFilePreviewAdapterState() - nowMs = 1_000_000 - vi.spyOn(Date, 'now').mockImplementation(() => nowMs) - }) - - afterEach(() => { - vi.restoreAllMocks() }) async function drive(streamEvent: StreamEvent, intent: ActiveFileIntent) { const context = createStreamingContext() - // channelId resolves to '' when the event carries no scope. context.activeFileIntents.set('', intent) await processFilePreviewStreamEvent({ streamId: STREAM_ID, streamEvent, context, execContext, - options: { onEvent: vi.fn() }, + options: { + onEvent: (event) => { + events.push(event as { payload: Record }) + }, + }, state, }) } - it('merges the growing full content (base version, no durable version) into the live doc as it streams', async () => { - const intent = makeIntent({ operation: 'append', fileId: 'file-grow', fileName: 'notes.md' }) - - await drive(editContentDelta('{"content":"Hello'), intent) - await flushMicrotasks() - - // Advance past the throttle window so the next delta is due for another merge. - nowMs += 300 - await drive(editContentDelta(' world'), intent) - await flushMicrotasks() - - expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledTimes(2) - const [first, second] = mergeEditIntoLiveFileDocMock.mock.calls - // A full-file snapshot (base + streamed), never a diff; it grows across deltas. Each streaming merge - // carries `baseVersion` (the durable version it was built from) to order it — never `version`, which - // rides the final edit_content write; so the relay drops it if a newer durable write has since landed - // but never records it as a durable checkpoint. - expect(first[0]).toBe('file-grow') - expect(first[1]).toContain('Base.') - expect(first[1]).toContain('Hello') - expect(first[2]?.baseVersion).toBe(BASE_VERSION_MS) - expect(first[2]?.version).toBeUndefined() - expect(second[1]).toContain('Hello world') - expect(second[1].length).toBeGreaterThan(first[1].length) - expect(second[2]?.baseVersion).toBe(BASE_VERSION_MS) - }) - - it('falls back to updatedAt for baseVersion when the file has no content version', async () => { - // A legacy file with no `contentUpdatedAt` — the base version must fall back to `updatedAt`, the SAME - // line the relay's synced version is on, so the snapshot is still ordered (not shipped unordered). - const UPDATED_AT_MS = 850_000 - peekFileIntentMock.mockResolvedValue({ - existingContent: 'Base.', - fileRecord: { contentUpdatedAt: null, updatedAt: new Date(UPDATED_AT_MS) }, - }) - const intent = makeIntent({ operation: 'append', fileId: 'file-legacy', fileName: 'notes.md' }) - - await drive(editContentDelta('{"content":"Hello'), intent) - await flushMicrotasks() - - expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledTimes(1) - expect(mergeEditIntoLiveFileDocMock.mock.calls[0][2]?.baseVersion).toBe(UPDATED_AT_MS) - }) - - it('throttles merges: two deltas within LIVE_DOC_MERGE_THROTTLE_MS yield one merge', async () => { - const intent = makeIntent({ - operation: 'append', - fileId: 'file-throttle', - fileName: 'notes.md', - }) - - await drive(editContentDelta('{"content":"Hel'), intent) - await flushMicrotasks() - - // 100ms < 250ms throttle → the second snapshot is dropped, not merged. - nowMs += 100 - await drive(editContentDelta('lo world'), intent) - await flushMicrotasks() - - expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledTimes(1) - }) - - it('does not merge for a non-markdown file (no collaborative room)', async () => { - const intent = makeIntent({ operation: 'append', fileId: 'file-txt', fileName: 'notes.txt' }) - - await drive(editContentDelta('{"content":"plain text body'), intent) - await flushMicrotasks() - - expect(mergeEditIntoLiveFileDocMock).not.toHaveBeenCalled() - }) - - it('does not merge when base content loads without a version (unordered-wipe guard)', async () => { - // Base text is available but the intent carries no file record → no baseVersion. The relay would - // treat a versionless snapshot as unordered (never stale), so it must be skipped fail-closed. - peekFileIntentMock.mockResolvedValue({ existingContent: 'Base.' }) - const intent = makeIntent({ operation: 'append', fileId: 'file-nover', fileName: 'notes.md' }) - - await drive(editContentDelta('{"content":"Hello'), intent) - await flushMicrotasks() - - expect(mergeEditIntoLiveFileDocMock).not.toHaveBeenCalled() - }) - - it('does not merge an append before base content loads (base-less-wipe guard)', async () => { - // No pending intent base is available yet → session.baseContent stays undefined. - peekFileIntentMock.mockResolvedValue(undefined) - const intent = makeIntent({ operation: 'append', fileId: 'file-append', fileName: 'notes.md' }) - - await drive(editContentDelta('{"content":"\\n- appended line'), intent) - await flushMicrotasks() - - // A base-less snapshot would diff to a delete-everything wipe of the seeded doc, so it must be skipped. - expect(mergeEditIntoLiveFileDocMock).not.toHaveBeenCalled() - }) - - it('does not stream an update (from-scratch rewrite) — it would blank the doc mid-stream', async () => { - const intent = makeIntent({ operation: 'update', fileId: 'file-update', fileName: 'notes.md' }) - - await drive(editContentDelta('{"content":"Rewritten intro'), intent) - await flushMicrotasks() - - // Update streams a partial rewrite; diffing the full doc toward it would delete most of the file - // until it grows back, so update applies atomically at the final durable write instead. - expect(mergeEditIntoLiveFileDocMock).not.toHaveBeenCalled() - }) + function previewContent(): string { + return events + .filter((e) => e.payload?.previewPhase === 'file_preview_content') + .map((e) => String(e.payload?.content ?? '')) + .join('') + } - it('skips the merge while one is already in flight for the file (does not backlog / advance throttle)', async () => { - isLiveDocMergeInFlightMock.mockReturnValue(true) - const intent = makeIntent({ operation: 'append', fileId: 'file-busy', fileName: 'notes.md' }) + it('emits the growing preview content (base + streamed) for an append stream', async () => { + const intent = makeIntent({ operation: 'append', fileId: 'file-grow', fileName: 'notes.md' }) await drive(editContentDelta('{"content":"Hello'), intent) await flushMicrotasks() - - expect(mergeEditIntoLiveFileDocMock).not.toHaveBeenCalled() - - // The dropped in-flight tick must NOT advance the throttle window, so once the in-flight merge - // clears the very next delta merges immediately — no wait for a fresh throttle interval. - isLiveDocMergeInFlightMock.mockReturnValue(false) await drive(editContentDelta(' world'), intent) await flushMicrotasks() - expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledTimes(1) - expect(mergeEditIntoLiveFileDocMock.mock.calls[0][0]).toBe('file-busy') + const combined = previewContent() + expect(combined).toContain('Base.') + expect(combined).toContain('Hello') + expect(combined).toContain('world') }) }) diff --git a/apps/sim/lib/copilot/request/go/file-preview-adapter.ts b/apps/sim/lib/copilot/request/go/file-preview-adapter.ts index ccc911bd08e..3d16cff4bb1 100644 --- a/apps/sim/lib/copilot/request/go/file-preview-adapter.ts +++ b/apps/sim/lib/copilot/request/go/file-preview-adapter.ts @@ -25,9 +25,7 @@ import { loadWorkspaceFileTextForPreview, type WorkspaceFilePreviewBase, } from '@/lib/copilot/tools/server/files/file-preview' -import { isLiveDocMergeInFlight, mergeEditIntoLiveFileDoc } from '@/lib/realtime/notify' import { resolveWorkspaceFileReference } from '@/lib/uploads/contexts/workspace/workspace-file-manager' -import { isMarkdownFile } from '@/lib/uploads/utils/file-utils' const logger = createLogger('CopilotFilePreviewAdapter') @@ -43,8 +41,6 @@ type FilePreviewStreamState = { session: FilePreviewSession lastEmittedPreviewText: string lastSnapshotAt: number - /** Epoch ms of the last merge of the growing content into the file's live collaborative Y.Doc. */ - lastLiveMergeAt: number } type ParsedWorkspaceFileArgs = { @@ -57,12 +53,6 @@ type ParsedWorkspaceFileArgs = { const PATCH_PREVIEW_SNAPSHOT_INTERVAL_MS = 80 const DELTA_PREVIEW_CHECKPOINT_INTERVAL_MS = 1000 -/** - * Throttle for merging the growing copilot content into the file's live collaborative Y.Doc as it - * streams. ~4 merges/sec reads as live while keeping CRDT diff churn and relay load bounded - * regardless of token rate; the final durable `edit_content` write is the stream-end flush. - */ -const LIVE_DOC_MERGE_THROTTLE_MS = 250 function asJsonRecord(value: unknown): JsonRecord | undefined { return value && typeof value === 'object' && !Array.isArray(value) @@ -273,7 +263,6 @@ function buildPreviewSessionFromIntent( operation: intent.operation, ...(intent.edit ? { edit: intent.edit } : {}), ...(typeof current?.baseContent === 'string' ? { baseContent: current.baseContent } : {}), - ...(typeof current?.baseVersion === 'number' ? { baseVersion: current.baseVersion } : {}), previewText: current?.previewText ?? '', previewVersion: current?.previewVersion ?? 0, status: current?.status ?? 'pending', @@ -412,16 +401,12 @@ export async function processFilePreviewStreamEvent(input: { session = { ...session, baseContent: previewBase.text, - ...(previewBase.baseVersion !== undefined - ? { baseVersion: previewBase.baseVersion } - : {}), } } filePreviewState.set(toolCallId, { session, lastEmittedPreviewText: '', lastSnapshotAt: 0, - lastLiveMergeAt: 0, }) await persistFilePreviewSession(session) @@ -487,16 +472,12 @@ export async function processFilePreviewStreamEvent(input: { session = { ...session, baseContent: previewBase.text, - ...(previewBase.baseVersion !== undefined - ? { baseVersion: previewBase.baseVersion } - : {}), } } filePreviewState.set(intent.toolCallId, { session, lastEmittedPreviewText: '', lastSnapshotAt: 0, - lastLiveMergeAt: 0, }) await persistFilePreviewSession(session) @@ -566,7 +547,6 @@ export async function processFilePreviewStreamEvent(input: { session: nextSession, lastEmittedPreviewText: previewText, lastSnapshotAt: Date.now(), - lastLiveMergeAt: 0, }) await persistFilePreviewSession(nextSession) await emitPreviewEvent(streamEvent, options, { @@ -600,7 +580,6 @@ export async function processFilePreviewStreamEvent(input: { session: buildPreviewSessionFromIntent(streamId, editIntent), lastEmittedPreviewText: '', lastSnapshotAt: 0, - lastLiveMergeAt: 0, } if ( @@ -619,15 +598,9 @@ export async function processFilePreviewStreamEvent(input: { } ) if (typeof intentBase?.existingContent === 'string') { - // Same version line as the seed/persist (`contentUpdatedAt ?? updatedAt`), so the stream's - // base is comparable to the relay's synced version even when the file has no content version. - const baseVersion = ( - intentBase.fileRecord?.contentUpdatedAt ?? intentBase.fileRecord?.updatedAt - )?.getTime() const seededSession: FilePreviewSession = { ...currentPreview.session, baseContent: intentBase.existingContent, - ...(baseVersion !== undefined ? { baseVersion } : {}), ...(intentBase.edit ? { edit: intentBase.edit } : {}), } currentPreview = { @@ -665,41 +638,12 @@ export async function processFilePreviewStreamEvent(input: { await persistFilePreviewSession(nextSession) - // Stream the growing content into the file's LIVE collaborative Y.Doc (when a room is open) - // so collaborators watching the file see the copilot write stream in via Yjs — the AI as a - // CRDT peer, applied by the relay as a minimal `updateYFragment` diff. Fire-and-forget so a - // slow relay never stalls the stream. Pass `baseVersion` (the durable version this snapshot is - // built from) so the relay drops it if a NEWER durable write has landed since — e.g. a - // concurrent human save — rather than diffing the live doc back toward stale content and - // clobbering that edit (which a later persist would then write over the durable file). It is - // never recorded as a checkpoint; the final `edit_content` write carries the real version and - // reconciles the durable file. No-op for `create` (never streams here) and for a file with no - // open room (the relay reports `applied: false`). - // - // Gates: markdown only (non-markdown has no collaborative room). Only `append`/`patch` stream - // — they build on the existing content, so they need the base loaded (a base-less snapshot - // would diff to a delete-everything wipe of the seeded doc). `update` is a from-scratch - // rewrite: streaming its partial content would diff the full doc toward a fragment and blank - // it mid-stream, so it applies atomically at the final durable write instead. Skip while a - // merge is in flight for this file — one at a time, and don't advance the throttle on a - // no-op — so a slow relay can't backlog stale snapshots or make the doc lag the stream. - // Require a numeric `baseVersion`: without it the relay can't order the snapshot and would - // treat it as unordered (never stale), so a rare base with no version (no file record) is - // fail-closed — skip the live merge rather than risk clobbering a concurrent durable write. - const dueForLiveMerge = - nextSession.fileId !== undefined && - isMarkdownFile({ type: editIntent.contentType, name: nextSession.fileName ?? '' }) && - (editIntent.operation === 'append' || editIntent.operation === 'patch') && - currentPreview.session.baseContent !== undefined && - nextSession.baseVersion !== undefined && - !isLiveDocMergeInFlight(nextSession.fileId) && - now - currentPreview.lastLiveMergeAt >= LIVE_DOC_MERGE_THROTTLE_MS - const nextLiveMergeAt = dueForLiveMerge ? now : currentPreview.lastLiveMergeAt - if (dueForLiveMerge && nextSession.fileId) { - void mergeEditIntoLiveFileDoc(nextSession.fileId, nextSession.previewText, { - baseVersion: nextSession.baseVersion, - }) - } + // The growing content is NOT merged into the live collaborative Y.Doc from here. When a + // collaborative editor for this file is open, that client applies the stream to the shared + // doc as minimal CRDT diffs (see `applyStreamedMarkdownToLiveDoc` in the editor), which + // renders smoothly locally AND broadcasts to every peer — so a server-side streaming merge + // would double-write the shared doc. The final `edit_content` durable write still reconciles + // the file and seeds any late joiner. if ( nextSession.operation === 'patch' && @@ -709,7 +653,6 @@ export async function processFilePreviewStreamEvent(input: { session: nextSession, lastEmittedPreviewText: currentPreview.lastEmittedPreviewText, lastSnapshotAt: currentPreview.lastSnapshotAt, - lastLiveMergeAt: nextLiveMergeAt, }) } else { const previewUpdate = buildPreviewContentUpdate( @@ -724,7 +667,6 @@ export async function processFilePreviewStreamEvent(input: { session: nextSession, lastEmittedPreviewText: nextSession.previewText, lastSnapshotAt: previewUpdate.lastSnapshotAt, - lastLiveMergeAt: nextLiveMergeAt, }) await emitPreviewEvent(streamEvent, options, { @@ -746,7 +688,6 @@ export async function processFilePreviewStreamEvent(input: { session: currentPreview.session, lastEmittedPreviewText: currentPreview.lastEmittedPreviewText, lastSnapshotAt: currentPreview.lastSnapshotAt, - lastLiveMergeAt: currentPreview.lastLiveMergeAt, }) } } @@ -780,7 +721,6 @@ export async function processFilePreviewStreamEvent(input: { session: currentPreview.session, lastEmittedPreviewText: currentPreview.session.previewText, lastSnapshotAt: Date.now(), - lastLiveMergeAt: currentPreview.lastLiveMergeAt, }) await emitPreviewEvent(streamEvent, options, { toolCallId: currentPreview.session.toolCallId, @@ -812,7 +752,6 @@ export async function processFilePreviewStreamEvent(input: { session: completedSession, lastEmittedPreviewText: completedSession.previewText, lastSnapshotAt: Date.now(), - lastLiveMergeAt: currentPreview.lastLiveMergeAt, }) await persistFilePreviewSession(completedSession) } diff --git a/apps/sim/lib/copilot/request/session/file-preview-session-contract.ts b/apps/sim/lib/copilot/request/session/file-preview-session-contract.ts index f29624f1562..a2e96208ba1 100644 --- a/apps/sim/lib/copilot/request/session/file-preview-session-contract.ts +++ b/apps/sim/lib/copilot/request/session/file-preview-session-contract.ts @@ -16,10 +16,6 @@ export interface FilePreviewSession { operation?: string edit?: Record baseContent?: string - /** The durable version (`contentUpdatedAt`, epoch ms) `baseContent` is at — the stream's causal base, - * passed to the relay so a snapshot is dropped if a newer durable write landed. Undefined for a - * legacy file with no recorded version, or a session with no loaded base. */ - baseVersion?: number previewText: string previewVersion: number updatedAt: string diff --git a/apps/sim/lib/copilot/request/session/file-preview-session.ts b/apps/sim/lib/copilot/request/session/file-preview-session.ts index c907ef020a1..df93b3a03ac 100644 --- a/apps/sim/lib/copilot/request/session/file-preview-session.ts +++ b/apps/sim/lib/copilot/request/session/file-preview-session.ts @@ -78,7 +78,6 @@ export function createFilePreviewSession(input: { operation?: string edit?: Record baseContent?: string - baseVersion?: number previewText?: string previewVersion?: number status?: FilePreviewStatus @@ -97,7 +96,6 @@ export function createFilePreviewSession(input: { ...(input.operation ? { operation: input.operation } : {}), ...(input.edit ? { edit: input.edit } : {}), ...(typeof input.baseContent === 'string' ? { baseContent: input.baseContent } : {}), - ...(typeof input.baseVersion === 'number' ? { baseVersion: input.baseVersion } : {}), previewText: input.previewText ?? '', previewVersion: input.previewVersion ?? 0, updatedAt: input.updatedAt ?? new Date().toISOString(), diff --git a/apps/sim/lib/copilot/tools/server/files/file-preview.test.ts b/apps/sim/lib/copilot/tools/server/files/file-preview.test.ts index 52a033c1b62..9e677b51377 100644 --- a/apps/sim/lib/copilot/tools/server/files/file-preview.test.ts +++ b/apps/sim/lib/copilot/tools/server/files/file-preview.test.ts @@ -31,6 +31,29 @@ describe('buildFilePreviewText', () => { ).toBe('line one\nline two') }) + it('fails closed (returns undefined) for an append when the base content has not loaded', () => { + // A base-less append preview is just the streamed fragment; a collaborative editor applying it as the + // full body would reconcile the seeded doc down to that fragment (a wipe). It must fail closed until + // the base loads, exactly like patch/update. + expect( + buildFilePreviewText({ + operation: 'append', + existingContent: undefined, + streamedContent: 'orphan fragment', + }) + ).toBeUndefined() + }) + + it('still previews an append into an EMPTY file (existingContent is "", not undefined)', () => { + expect( + buildFilePreviewText({ + operation: 'append', + existingContent: '', + streamedContent: 'first line', + }) + ).toBe('first line') + }) + it('applies anchored replace_between previews', () => { expect( buildFilePreviewText({ diff --git a/apps/sim/lib/copilot/tools/server/files/file-preview.ts b/apps/sim/lib/copilot/tools/server/files/file-preview.ts index 698d524d463..11b09b61c3a 100644 --- a/apps/sim/lib/copilot/tools/server/files/file-preview.ts +++ b/apps/sim/lib/copilot/tools/server/files/file-preview.ts @@ -141,16 +141,9 @@ function buildAppendPreview(existingContent: string, incomingContent: string): s * before Redis holds `existingContent`, which would make append previews look like * full-file replacement until the intent landed. */ -/** - * The base content a copilot edit is computed against, plus the durable version (epoch ms) that content - * is at. The version is the stream's causal base: the relay drops a streaming snapshot if a NEWER durable - * write landed than this, so a concurrent human edit is never clobbered. Derived as - * `contentUpdatedAt ?? updatedAt` — the SAME version line the seed/persist use — so it is directly - * comparable to the relay's recorded synced version. - */ +/** The base content a copilot append/patch edit is computed against, to compose the streamed preview. */ export interface WorkspaceFilePreviewBase { text: string - baseVersion: number } export async function loadWorkspaceFileTextForPreview( @@ -163,7 +156,6 @@ export async function loadWorkspaceFileTextForPreview( const buffer = await fetchWorkspaceFileBuffer(record) return { text: buffer.toString('utf-8'), - baseVersion: (record.contentUpdatedAt ?? record.updatedAt).getTime(), } } catch (error) { logger.warn('Failed to load workspace file text for preview', { @@ -190,10 +182,15 @@ export function buildFilePreviewText({ } if (operation === 'append') { - if (existingContent !== undefined) { - return buildAppendPreview(existingContent, streamedContent) + // Fail closed (like `patch`/`update` below) when the base file content has not loaded yet: a base-less + // `append` preview is just the streamed fragment, and a collaborative editor applying it as the full + // body would reconcile the seeded doc down to that fragment (a wipe). Skipping the preview until the + // base is available costs only a brief render delay; the final durable `edit_content` write is + // authoritative. An empty file has `existingContent === ''` (defined), so it is unaffected. + if (existingContent === undefined) { + return undefined } - return streamedContent + return buildAppendPreview(existingContent, streamedContent) } if (existingContent === undefined) { diff --git a/apps/sim/lib/realtime/notify.test.ts b/apps/sim/lib/realtime/notify.test.ts index 20689dd745d..98196492257 100644 --- a/apps/sim/lib/realtime/notify.test.ts +++ b/apps/sim/lib/realtime/notify.test.ts @@ -6,7 +6,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest' vi.mock('@/lib/core/utils/urls', () => ({ getSocketServerUrl: () => 'http://realtime' })) vi.mock('@/lib/core/config/env', () => ({ env: { INTERNAL_API_SECRET: 'secret' } })) -import { isLiveDocMergeInFlight, mergeEditIntoLiveFileDoc } from './notify' +import { mergeEditIntoLiveFileDoc } from './notify' describe('mergeEditIntoLiveFileDoc', () => { afterEach(() => { @@ -24,25 +24,12 @@ describe('mergeEditIntoLiveFileDoc', () => { expect.objectContaining({ method: 'POST', headers: expect.objectContaining({ 'x-api-key': 'secret' }), - // A durable write sends `version`; the undefined `baseVersion` is dropped by JSON.stringify. + // A durable write sends `version`; an unversioned (legacy) call would drop it via JSON.stringify. body: JSON.stringify({ fileId: 'file-1', markdown: '# hello', version: 42 }), }) ) }) - it('sends baseVersion (not version) for a streaming snapshot', async () => { - const fetchMock = vi.fn().mockResolvedValue({ ok: true }) - vi.stubGlobal('fetch', fetchMock) - - await mergeEditIntoLiveFileDoc('file-1', '# hello', { baseVersion: 1234 }) - - // The relay orders the snapshot by its causal baseVersion without recording it — durable version - // stays absent on the wire. - expect(fetchMock.mock.calls[0][1].body).toBe( - JSON.stringify({ fileId: 'file-1', markdown: '# hello', baseVersion: 1234 }) - ) - }) - it('never throws when the realtime call fails (best-effort)', async () => { vi.stubGlobal('fetch', vi.fn().mockRejectedValue(new Error('socket pod down'))) await expect( @@ -57,61 +44,42 @@ describe('mergeEditIntoLiveFileDoc', () => { ).resolves.toBeUndefined() }) - it('reports isLiveDocMergeInFlight while a merge runs and clears when it settles', async () => { - let resolveFetch: (value: { ok: boolean }) => void = () => {} - vi.stubGlobal( - 'fetch', - vi.fn(() => new Promise((resolve) => (resolveFetch = resolve))) - ) - - expect(isLiveDocMergeInFlight('file-flight')).toBe(false) - const run = mergeEditIntoLiveFileDoc('file-flight', 'v1') - await Promise.resolve() - // The streaming caller checks this to skip a redundant merge (and not advance its throttle) while - // one is in flight, so a slow relay can't backlog stale snapshots. - expect(isLiveDocMergeInFlight('file-flight')).toBe(true) - - resolveFetch({ ok: true }) - await run - expect(isLiveDocMergeInFlight('file-flight')).toBe(false) - }) - - it('a durable (versioned) merge waits for an in-flight streaming merge, then applies last', async () => { - let resolveStream: (value: { ok: boolean }) => void = () => {} + it('a later durable merge waits for an in-flight earlier one, then applies last', async () => { + let resolveFirst: (value: { ok: boolean }) => void = () => {} const fetchMock = vi .fn() - .mockImplementationOnce(() => new Promise((resolve) => (resolveStream = resolve))) + .mockImplementationOnce(() => new Promise((resolve) => (resolveFirst = resolve))) .mockResolvedValue({ ok: true }) vi.stubGlobal('fetch', fetchMock) - const stream = mergeEditIntoLiveFileDoc('file-durable', 'partial') // versionless, in flight + const first = mergeEditIntoLiveFileDoc('file-durable', 'earlier', { version: 99 }) // in flight await Promise.resolve() - const durable = mergeEditIntoLiveFileDoc('file-durable', 'final content', { version: 100 }) // versioned + const durable = mergeEditIntoLiveFileDoc('file-durable', 'final content', { version: 100 }) await Promise.resolve() await Promise.resolve() - // The durable write waits for the in-flight streaming merge → its fetch has not fired yet, so it - // cannot be reordered before a straggler and cannot be clobbered by one. + // The later write waits for the in-flight earlier one → its fetch has not fired yet, so it cannot be + // reordered before a straggler and cannot be clobbered by one. expect(fetchMock).toHaveBeenCalledTimes(1) - resolveStream({ ok: true }) - await stream + resolveFirst({ ok: true }) + await first await durable - // Only after the streaming merge completed does the durable (final) merge apply — always last. + // Only after the earlier merge completed does the later (final) merge apply — always last. expect(fetchMock).toHaveBeenCalledTimes(2) expect(fetchMock.mock.calls[1][1].body).toBe( JSON.stringify({ fileId: 'file-durable', markdown: 'final content', version: 100 }) ) }) - it('serializes concurrent durable writes behind a streaming merge, strictly in order', async () => { - const applied: Array = [] + it('serializes concurrent durable writes to a file strictly in order', async () => { + const applied: number[] = [] const resolvers: Array<() => void> = [] vi.stubGlobal( 'fetch', vi.fn((_url: string, init: { body: string }) => { - applied.push(JSON.parse(init.body).version ?? 'stream') + applied.push(JSON.parse(init.body).version) return new Promise<{ ok: boolean }>((resolve) => resolvers.push(() => resolve({ ok: true })) ) @@ -121,22 +89,22 @@ describe('mergeEditIntoLiveFileDoc', () => { for (let i = 0; i < 6; i++) await Promise.resolve() } - const s = mergeEditIntoLiveFileDoc('file-order', 's') // streaming, in flight + const s = mergeEditIntoLiveFileDoc('file-order', 's', { version: 0 }) // in flight await flush() - // Two durable writes arrive while the streaming merge is in flight — both must chain, not both + // Two later durable writes arrive while the first merge is in flight — both must chain, not both // resume-and-fire concurrently. const a = mergeEditIntoLiveFileDoc('file-order', 'a', { version: 1 }) const b = mergeEditIntoLiveFileDoc('file-order', 'b', { version: 2 }) await flush() - expect(applied).toEqual(['stream']) // A and B queued behind streaming + expect(applied).toEqual([0]) // A and B queued behind the in-flight first merge - resolvers[0]() // finish streaming → A applies next (not B) + resolvers[0]() // finish first → A applies next (not B) await flush() - expect(applied).toEqual(['stream', 1]) + expect(applied).toEqual([0, 1]) resolvers[1]() // finish A → B applies after A await flush() - expect(applied).toEqual(['stream', 1, 2]) + expect(applied).toEqual([0, 1, 2]) resolvers[2]() await Promise.all([s, a, b]) diff --git a/apps/sim/lib/realtime/notify.ts b/apps/sim/lib/realtime/notify.ts index a67f86401c4..b377635c49b 100644 --- a/apps/sim/lib/realtime/notify.ts +++ b/apps/sim/lib/realtime/notify.ts @@ -110,49 +110,39 @@ export async function notifyFolderResourceChanged( } /** - * How a live-doc merge is positioned on the file's monotonic version line. Pass one key or the other, - * never both; passing neither applies the merge without ordering it (legacy). + * How a durable live-doc merge is positioned on the file's monotonic version line. Omit `version` to + * apply the merge without ordering it (legacy). */ export interface LiveFileDocMergeOrder { /** A durable write's `contentUpdatedAt` (epoch ms): applied only if newer than the version the doc - * already incorporates, AND recorded as the synced version. */ + * already incorporates, AND recorded as the synced version (the persist If-Match guard). */ version?: number - /** A streaming snapshot's causal base — the durable `contentUpdatedAt` it was built from. The relay - * drops the snapshot if a NEWER durable version was recorded than this (a concurrent write landed), and - * never records it as a checkpoint. */ - baseVersion?: number } /** - * Best-effort: ask the realtime relay to merge a copilot edit into a file's LIVE collaborative - * document, so open editors see it stream in as a CRDT merge (Stage C) rather than the file changing - * underneath them. No-op when no doc is (or was recently) live (the relay reports `applied: false`). - * The file itself is written durably by the caller regardless — this only drives the live view. - * Never throws. + * Best-effort: ask the realtime relay to merge a durable copilot/file write into a file's LIVE + * collaborative document, so open editors reconcile to it as a CRDT merge rather than the file changing + * underneath them, and a late joiner is seeded from it. No-op when no doc is (or was recently) live (the + * relay reports `applied: false`). The file itself is written durably by the caller regardless — this + * only drives the live view. Never throws. * - * The former clobber gap — an open editor's autosave dropping this edit — is now closed: a - * collaborative editor no longer client-autosaves (the relay persists the shared doc to markdown - * server-side), and the relay applies this merge THROUGH the shared Redis stream, so it reaches the - * live doc on whichever task holds it and can't go stale relative to this direct write. + * (Streaming copilot output is NOT merged here: the open editor applies the stream client-side as minimal + * CRDT diffs — see `applyStreamedMarkdownToLiveDoc` — which renders smoothly and broadcasts to peers. This + * merge is the stream-end durable reconcile, and by then it is usually a noop diff.) * - * A durable caller awaits this (so the fetch dispatches before the route handler returns); the copilot - * streaming caller fires and forgets it. Bounded to {@link APPLY_EDIT_TIMEOUT_MS}, so it adds latency - * only when the socket pod is unreachable. + * The former clobber gap — an open editor's autosave dropping this edit — is closed: a collaborative + * editor no longer client-autosaves (the relay persists the shared doc to markdown server-side), and the + * relay applies this merge THROUGH the shared Redis stream, so it reaches the live doc on whichever task + * holds it and can't go stale relative to this direct write. * - * `order` ({@link LiveFileDocMergeOrder}) positions this merge so a stale write never regresses the doc. - * A durable `version` applies only if newer than the version the doc already incorporates, and is recorded - * as the synced version (the persist If-Match guard). A streaming `baseVersion` is the durable version the - * snapshot was built from: the relay drops the snapshot if a NEWER durable write has since landed — so a - * concurrent human edit is never clobbered (nor later persisted over the file) — and never records it, so - * the synced version stays pinned to the last durable write, which the copilot tool's final `edit_content` - * write carries, reconciling the durable file. + * The caller awaits this so the fetch dispatches before the route handler returns. Bounded to + * {@link APPLY_EDIT_TIMEOUT_MS}, so it adds latency only when the socket pod is unreachable. * - * Ordering is enforced at two scales. Within this process, merges for a file run on a single serialized - * chain — each chained after the current tail — so a durable write applies after any in-flight streaming - * merge and after every earlier durable write, never concurrently. Across processes, the per-process - * chain does not apply, so the relay orders merges by that monotonic version under a cluster-wide lock. - * The copilot streaming caller uses {@link isLiveDocMergeInFlight} to skip redundant snapshots while one - * is in flight, so a slow relay can't backlog stale snapshots. + * `order.version` positions the merge so a stale write never regresses the doc: it applies only if newer + * than the version the doc already incorporates, and is recorded as the synced version. Ordering is + * enforced at two scales: within this process, merges for a file run on a single serialized chain (each + * chained after the current tail) so writes never apply concurrently; across processes the relay orders + * by that monotonic version under a cluster-wide lock. */ export async function mergeEditIntoLiveFileDoc( fileId: string, @@ -173,16 +163,6 @@ export async function mergeEditIntoLiveFileDoc( * because {@link applyLiveFileDocMerge} never throws. Absent when the file's chain is idle. */ const liveDocMergeChain = new Map>() -/** - * Whether a live-doc merge is currently running or queued for the file. The copilot streaming caller - * checks this to skip a redundant snapshot (and to not advance its send throttle) while a merge is in - * flight — bounding the stream to one live merge per file at a time without backlogging stale - * snapshots behind a slow relay. - */ -export function isLiveDocMergeInFlight(fileId: string): boolean { - return liveDocMergeChain.has(fileId) -} - /** POST the merge to the relay. Never throws (a live-doc merge is best-effort). */ async function applyLiveFileDocMerge( fileId: string, @@ -194,13 +174,11 @@ async function applyLiveFileDocMerge( method: 'POST', headers: { 'Content-Type': 'application/json', 'x-api-key': env.INTERNAL_API_SECRET }, // `version` (durable `contentUpdatedAt`) records the synced version the live doc now incorporates - // (the persist If-Match guard); `baseVersion` is a streaming snapshot's causal base, checked (drop - // if a newer durable landed) but never recorded. JSON.stringify drops whichever is undefined. + // (the persist If-Match guard). JSON.stringify drops it when undefined (an unordered legacy merge). body: JSON.stringify({ fileId, markdown, version: order.version, - baseVersion: order.baseVersion, }), signal: AbortSignal.timeout(APPLY_EDIT_TIMEOUT_MS), }) diff --git a/packages/realtime-protocol/src/file-doc.ts b/packages/realtime-protocol/src/file-doc.ts index e58ceb477a3..6746e94e97c 100644 --- a/packages/realtime-protocol/src/file-doc.ts +++ b/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]