diff --git a/apps/realtime/src/handlers/file-doc.multireplica.test.ts b/apps/realtime/src/handlers/file-doc.multireplica.test.ts new file mode 100644 index 00000000000..17c17ffe371 --- /dev/null +++ b/apps/realtime/src/handlers/file-doc.multireplica.test.ts @@ -0,0 +1,90 @@ +/** + * @vitest-environment node + * + * Multi-replica (store-enabled) coverage for the copilot live-merge stale-check. The main + * `file-doc.test.ts` runs with the store DISABLED (single-replica fallback); this file mocks an ENABLED + * store so the cross-process branch of `mergeMarkdownIntoRoom` — staleness against the SHARED synced + * version under the merge lock, and `recordVersion` writing `setSyncedVersion` — is exercised directly. + * The enabled merge path reads its base from the shared store (not an in-memory room), so no JOIN/seed + * is needed: calling `applyMarkdownToLiveFileDoc` against the fake store drives the branch on its own. + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import * as Y from 'yjs' + +const { mockFetchFileDocMerge } = vi.hoisted(() => ({ + mockFetchFileDocMerge: vi.fn(), +})) + +/** + * A minimal ENABLED store: in-memory monotonic synced version (mirrors SET_VERSION_IF_NEWER_SCRIPT), a + * non-null stream state so the merge has a base, and no-op locks/publish. Only the surface the + * store-enabled merge path touches is implemented. + */ +const fakeStore = { + enabled: true, + versions: new Map(), + acquireMergeSlot: vi.fn(async () => 'token'), + releaseMergeSlot: vi.fn(async () => {}), + getStreamState: vi.fn(async () => new Uint8Array([1])), + publishAndWait: vi.fn(async () => {}), + getSyncedVersion: vi.fn(async (name: string) => fakeStore.versions.get(name) ?? null), + setSyncedVersion: vi.fn(async (name: string, version: number) => { + fakeStore.versions.set(name, Math.max(fakeStore.versions.get(name) ?? 0, version)) + }), +} + +vi.mock('@sim/platform-authz/rooms', () => ({ authorizeRoom: vi.fn() })) + +vi.mock('@/handlers/file-doc-app', () => ({ + fetchFileDocSeed: vi.fn(), + fetchFileDocMerge: mockFetchFileDocMerge, + fetchFileDocPersist: vi.fn(), +})) + +vi.mock('@/handlers/file-doc-store', () => ({ + getFileDocStore: () => fakeStore, + REDIS_ORIGIN: Symbol('redis'), + REDIS_SNAPSHOT_ORIGIN: Symbol('redis-snapshot'), +})) + +import { applyMarkdownToLiveFileDoc } from '@/handlers/file-doc' + +const ROOM_NAME = 'workspace-file-doc:file-1' + +describe('applyMarkdownToLiveFileDoc — multi-replica (store-enabled) ordering', () => { + beforeEach(() => { + vi.clearAllMocks() + fakeStore.versions.clear() + fakeStore.acquireMergeSlot.mockResolvedValue('token') + fakeStore.getStreamState.mockResolvedValue(new Uint8Array([1])) + mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc())) + }) + + it('drops a stale-base streaming snapshot against the SHARED synced version and never records it', 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' + ) + 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') + 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. + 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. + 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 4e0d9ffa2de..7ccbe3d2254 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -543,6 +543,65 @@ describe('setupWorkspaceFileDocHandlers', () => { expect(mockFetchFileDocMerge).not.toHaveBeenCalled() }) + it('rejects a stale versioned merge (not newer than the synced version) without regressing the doc', 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 newer durable version lands and is recorded as the synced version. + expect(await applyMarkdownToLiveFileDoc('file-1', '# newer', { version: 100 })).toBe('applied') + mockFetchFileDocMerge.mockClear() + + // An older durable version arriving out of order (e.g. a concurrent write on another process) is + // stale: skipped before any diff is computed, so the live doc never regresses to older content and + // no diff is published that a later persist could write back. + expect(await applyMarkdownToLiveFileDoc('file-1', '# older, stale', { version: 50 })).toBe( + 'stale' + ) + // The same version is idempotent — also skipped. + expect(await applyMarkdownToLiveFileDoc('file-1', '# same version', { version: 100 })).toBe( + 'stale' + ) + 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() diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 194943153a6..039e2ab6661 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -515,6 +515,16 @@ function emptySeedUpdate(): Uint8Array { /** Serializes live merges per file so overlapping calls never race the same doc (see below). */ 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. + */ +interface MergeOrder { + version?: number + baseVersion?: number +} + /** * Apply new markdown into a file's LIVE collaborative document (Stage C — copilot writing into an open * doc). Ships the document's current state to the app to build a minimal Yjs diff, applies it — which @@ -540,14 +550,12 @@ const fileDocMergeChains = new Map>() export function applyMarkdownToLiveFileDoc( fileId: string, markdown: string, - version?: number -): Promise<'applied' | 'no-live-room' | 'merge-unavailable'> { + order: MergeOrder = {} +): Promise<'applied' | 'no-live-room' | 'merge-unavailable' | 'stale'> { const name = roomName(fileDocRoom(fileId)) const prior = fileDocMergeChains.get(name) ?? Promise.resolve() // `.catch` so a failed prior merge doesn't reject this one — each merge is independent. - const run = prior - .catch(() => {}) - .then(() => mergeMarkdownIntoRoom(name, fileId, markdown, version)) + const run = prior.catch(() => {}).then(() => mergeMarkdownIntoRoom(name, fileId, markdown, order)) fileDocMergeChains.set( name, run.finally(() => { @@ -561,21 +569,51 @@ async function mergeMarkdownIntoRoom( name: string, fileId: string, markdown: string, - version?: number -): Promise<'applied' | 'no-live-room' | 'merge-unavailable'> { + { version, baseVersion }: 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. Set on success below. - const recordVersion = () => { + // 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. + const recordVersion = async () => { if (version === undefined) return const room = fileDocRooms.get(name) - // Never regress the token: merges/seeds/persists all write it (locally and via fire-and-forget - // Redis), so a lower value arriving out of order must not shadow a higher one the doc already - // incorporates (the Redis side is guarded identically by SET_VERSION_IF_NEWER_SCRIPT). + // Never regress the token: merges/seeds/persists all write it, so a lower value arriving out of + // order must not shadow a higher one the doc already incorporates (the Redis side is guarded + // identically by SET_VERSION_IF_NEWER_SCRIPT). if (room) room.syncedVersion = Math.max(room.syncedVersion ?? 0, version) - void store.setSyncedVersion(name, version) + await store.setSyncedVersion(name, version) + } + + // 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. + const isStale = (current: number): boolean => { + if (version !== undefined) return version <= current + if (baseVersion !== undefined) return current > baseVersion + return false } if (store.enabled) { @@ -595,6 +633,11 @@ async function mergeMarkdownIntoRoom( return 'merge-unavailable' } try { + // Staleness is checked under the lock against the cluster-wide synced version, so a durable merge + // that lost the race to a newer one (on any process) is dropped rather than regressing the doc. + const shared = await store.getSyncedVersion(name) + const current = Math.max(shared ?? 0, fileDocRooms.get(name)?.syncedVersion ?? 0) + if (isStale(current)) return 'stale' // Compute the diff against the committed SHARED state and PUBLISH it — every task with the doc // live (including this one, via its own tailer) applies it and fans it out to its clients, so the // merge reaches the live doc no matter which task the apply-edit call landed on. An empty stream @@ -604,7 +647,7 @@ async function mergeMarkdownIntoRoom( if (!base) return 'no-live-room' const diff = await fetchFileDocMerge(fileId, base, markdown) await store.publishAndWait(name, diff) - recordVersion() + await recordVersion() return 'applied' } finally { await store.releaseMergeSlot(name, token) @@ -614,12 +657,13 @@ async function mergeMarkdownIntoRoom( // Single-replica fallback: apply straight to the local authoritative doc. const room = fileDocRooms.get(name) if (!room || room.owners.size === 0 || !isDocSeeded(room.doc)) return 'no-live-room' + if (isStale(room.syncedVersion ?? 0)) return 'stale' const update = await fetchFileDocMerge(fileId, Y.encodeStateAsUpdate(room.doc), markdown) // The room may have been dropped while the diff was being built; never touch a destroyed doc. if (fileDocRooms.get(name) !== room) return 'no-live-room' // No transaction origin → `doc.on('update')` relays to the WHOLE room (every editor sees copilot). Y.applyUpdate(room.doc, update) - recordVersion() + await recordVersion() return 'applied' } diff --git a/apps/realtime/src/routes/http.ts b/apps/realtime/src/routes/http.ts index fe824a70b86..b37e9061107 100644 --- a/apps/realtime/src/routes/http.ts +++ b/apps/realtime/src/routes/http.ts @@ -209,17 +209,17 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) { if (req.method === 'POST' && req.url === '/api/file-doc/apply-edit') { try { const body = await readRequestBody(req) - const { fileId, markdown, version } = JSON.parse(body) + const { fileId, markdown, version, baseVersion } = 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. - const result = await applyMarkdownToLiveFileDoc( - fileId, - markdown, - typeof version === 'number' ? version : undefined - ) + // `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' })) } catch (error) { 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 4290dd696fe..f2a1eaad79b 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 @@ -3,7 +3,7 @@ import { memo, useCallback, useEffect, useRef, useState } from 'react' import { cn, toast } from '@sim/emcn' import { FILE_DOC_SEED, type JoinFileDocError } from '@sim/realtime-protocol/file-doc' -import type { Extensions, JSONContent } from '@tiptap/core' +import { type Extensions, generateHTML, type JSONContent } from '@tiptap/core' import { isChangeOrigin } from '@tiptap/extension-collaboration' import { Fragment, Slice } from '@tiptap/pm/model' import { NodeSelection } from '@tiptap/pm/state' @@ -297,6 +297,18 @@ export function LoadedRichMarkdownEditor({ ? '' : parseMarkdownToDoc(splitFrontmatter(content).body) ) + /** + * A read-only placeholder rendered from the already-fetched markdown while a collaborative doc waits + * for its server seed, so the pane shows content instantly instead of blocking blank on the socket + * round-trip (the seed IS the same markdown, so the swap on {@link collabReady} is seamless). Static + * HTML — it holds no editor, doc, or awareness, so it structurally cannot write to the Y.Doc, which + * is the invariant that keeps seeding out of the client (a client seed duplicates the doc). + */ + const [placeholderHtml] = useState(() => + collaborationEnabled + ? generateHTML(parseMarkdownToDoc(splitFrontmatter(content).body), EXTENSIONS) + : null + ) /** * The body currently shown in the editor: seeded from a settled mount, updated on local edits (via * onUpdate) and on each streamed sync. Incremental edits (append/patch) stream complete snapshots and @@ -923,6 +935,11 @@ export function LoadedRichMarkdownEditor({ [] ) + // 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 + return (
0) void insertImagesRef.current(images, at) }} /> + {showPlaceholder && placeholderHtml && ( + // Instant read-only content while the collaborative doc seeds; the editor stays mounted-but- + // hidden below so it renders the seeded doc before the swap. Same layout box → no reflow. +
+ )}
) diff --git a/apps/sim/lib/collab-doc/merge.ts b/apps/sim/lib/collab-doc/merge.ts index 3db9c73365e..85e5bee9733 100644 --- a/apps/sim/lib/collab-doc/merge.ts +++ b/apps/sim/lib/collab-doc/merge.ts @@ -10,10 +10,12 @@ import { applyMarkdownToYDoc } from './converter' * and applies the returned diff, which Yjs merges with any concurrent user edits before relaying it to * every connected editor. * - * `applyMarkdownToYDoc` performs a real `updateYFragment` diff (not a replace), so unrelated - * paragraphs the user is editing are preserved. The returned update is relative to the document's - * state at call time (`Y.encodeStateAsUpdate(doc, before)`), so it is exactly the change to apply — and - * is empty (a no-op update) when `markdown` already matches the document. + * `applyMarkdownToYDoc` performs a real `updateYFragment` diff (not a replace), so paragraphs the diff + * does not touch are preserved even while the user edits them. A region the incoming `markdown` DOES + * change is reconciled toward that markdown — a concurrent user edit inside such a region is diffed + * away, since `markdown` is built from a base snapshot, not the user's in-flight text. The returned + * update is relative to the document's state at call time (`Y.encodeStateAsUpdate(doc, before)`), so it + * is exactly the change to apply — and is empty (a no-op update) when `markdown` already matches. */ export function buildFileDocMergeUpdate(docState: Uint8Array, markdown: string): Uint8Array { const doc = new Y.Doc() 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 new file mode 100644 index 00000000000..8915f2cd472 --- /dev/null +++ b/apps/sim/lib/copilot/request/go/file-preview-adapter.test.ts @@ -0,0 +1,262 @@ +/** + * @vitest-environment node + */ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + MothershipStreamV1EventType, + MothershipStreamV1ToolExecutor, + MothershipStreamV1ToolMode, + 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, +})) + +import { createStreamingContext } from '@/lib/copilot/request/context/request-context' +import { + createFilePreviewAdapterState, + type FilePreviewAdapterState, + processFilePreviewStreamEvent, +} from '@/lib/copilot/request/go/file-preview-adapter' +import { createEvent, eventToStreamEvent } from '@/lib/copilot/request/session' +import type { ActiveFileIntent, ExecutionContext, StreamEvent } from '@/lib/copilot/request/types' + +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. */ +function editContentDelta(argumentsDelta: string): StreamEvent { + return eventToStreamEvent( + createEvent({ + streamId: STREAM_ID, + cursor: '1', + seq: 1, + requestId: 'req-1', + type: MothershipStreamV1EventType.tool, + payload: { + toolCallId: EDIT_TOOL_CALL_ID, + toolName: 'edit_content', + executor: MothershipStreamV1ToolExecutor.sim, + mode: MothershipStreamV1ToolMode.async, + phase: MothershipStreamV1ToolPhase.args_delta, + argumentsDelta, + }, + }) + ) +} + +function makeIntent(overrides: { + operation: string + fileId?: string + fileName: string +}): ActiveFileIntent { + return { + toolCallId: WORKSPACE_FILE_TOOL_CALL_ID, + operation: overrides.operation, + target: { + kind: 'file_id', + ...(overrides.fileId ? { fileId: overrides.fileId } : {}), + fileName: overrides.fileName, + }, + } +} + +const flushMicrotasks = async () => { + await Promise.resolve() + await Promise.resolve() +} + +describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { + let state: FilePreviewAdapterState + let nowMs: number + const execContext: ExecutionContext = { + userId: 'user-1', + workflowId: 'workflow-1', + workspaceId: 'workspace-1', + chatId: 'chat-1', + messageId: 'msg-1', + } + + 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. + 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() }, + 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() + }) + + 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' }) + + 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') + }) +}) 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 581dd8c8f93..ccc911bd08e 100644 --- a/apps/sim/lib/copilot/request/go/file-preview-adapter.ts +++ b/apps/sim/lib/copilot/request/go/file-preview-adapter.ts @@ -23,8 +23,11 @@ import { peekFileIntent } from '@/lib/copilot/tools/server/files/file-intent-sto import { buildFilePreviewText, 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') @@ -40,6 +43,8 @@ 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 = { @@ -52,6 +57,12 @@ 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) @@ -262,6 +273,7 @@ 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', @@ -386,26 +398,30 @@ export async function processFilePreviewStreamEvent(input: { setIntent(intent) if (isContentOp && previewTargetKind) { - let previewBaseContent: string | undefined + let previewBase: WorkspaceFilePreviewBase | undefined if ( execContext.workspaceId && fileId && (operation === 'append' || operation === 'patch') ) { - previewBaseContent = await loadWorkspaceFileTextForPreview( - execContext.workspaceId, - fileId - ) + previewBase = await loadWorkspaceFileTextForPreview(execContext.workspaceId, fileId) } let session = buildPreviewSessionFromIntent(streamId, intent) - if (previewBaseContent !== undefined) { - session = { ...session, baseContent: previewBaseContent } + if (previewBase !== undefined) { + session = { + ...session, + baseContent: previewBase.text, + ...(previewBase.baseVersion !== undefined + ? { baseVersion: previewBase.baseVersion } + : {}), + } } filePreviewState.set(toolCallId, { session, lastEmittedPreviewText: '', lastSnapshotAt: 0, + lastLiveMergeAt: 0, }) await persistFilePreviewSession(session) @@ -458,25 +474,29 @@ export async function processFilePreviewStreamEvent(input: { } setIntent(intent) - let previewBaseContent: string | undefined + let previewBase: WorkspaceFilePreviewBase | undefined if ( execContext.workspaceId && (intent.operation === 'append' || intent.operation === 'patch') ) { - previewBaseContent = await loadWorkspaceFileTextForPreview( - execContext.workspaceId, - result.fileId - ) + previewBase = await loadWorkspaceFileTextForPreview(execContext.workspaceId, result.fileId) } let session = buildPreviewSessionFromIntent(streamId, intent) - if (previewBaseContent !== undefined) { - session = { ...session, baseContent: previewBaseContent } + if (previewBase !== undefined) { + session = { + ...session, + baseContent: previewBase.text, + ...(previewBase.baseVersion !== undefined + ? { baseVersion: previewBase.baseVersion } + : {}), + } } filePreviewState.set(intent.toolCallId, { session, lastEmittedPreviewText: '', lastSnapshotAt: 0, + lastLiveMergeAt: 0, }) await persistFilePreviewSession(session) @@ -546,6 +566,7 @@ export async function processFilePreviewStreamEvent(input: { session: nextSession, lastEmittedPreviewText: previewText, lastSnapshotAt: Date.now(), + lastLiveMergeAt: 0, }) await persistFilePreviewSession(nextSession) await emitPreviewEvent(streamEvent, options, { @@ -579,6 +600,7 @@ export async function processFilePreviewStreamEvent(input: { session: buildPreviewSessionFromIntent(streamId, editIntent), lastEmittedPreviewText: '', lastSnapshotAt: 0, + lastLiveMergeAt: 0, } if ( @@ -597,9 +619,15 @@ 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 = { @@ -637,6 +665,42 @@ 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, + }) + } + if ( nextSession.operation === 'patch' && now - currentPreview.lastSnapshotAt < PATCH_PREVIEW_SNAPSHOT_INTERVAL_MS @@ -645,6 +709,7 @@ export async function processFilePreviewStreamEvent(input: { session: nextSession, lastEmittedPreviewText: currentPreview.lastEmittedPreviewText, lastSnapshotAt: currentPreview.lastSnapshotAt, + lastLiveMergeAt: nextLiveMergeAt, }) } else { const previewUpdate = buildPreviewContentUpdate( @@ -659,6 +724,7 @@ export async function processFilePreviewStreamEvent(input: { session: nextSession, lastEmittedPreviewText: nextSession.previewText, lastSnapshotAt: previewUpdate.lastSnapshotAt, + lastLiveMergeAt: nextLiveMergeAt, }) await emitPreviewEvent(streamEvent, options, { @@ -680,6 +746,7 @@ export async function processFilePreviewStreamEvent(input: { session: currentPreview.session, lastEmittedPreviewText: currentPreview.lastEmittedPreviewText, lastSnapshotAt: currentPreview.lastSnapshotAt, + lastLiveMergeAt: currentPreview.lastLiveMergeAt, }) } } @@ -713,6 +780,7 @@ 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, @@ -744,6 +812,7 @@ 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 a2e96208ba1..f29624f1562 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,6 +16,10 @@ 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 df93b3a03ac..c907ef020a1 100644 --- a/apps/sim/lib/copilot/request/session/file-preview-session.ts +++ b/apps/sim/lib/copilot/request/session/file-preview-session.ts @@ -78,6 +78,7 @@ export function createFilePreviewSession(input: { operation?: string edit?: Record baseContent?: string + baseVersion?: number previewText?: string previewVersion?: number status?: FilePreviewStatus @@ -96,6 +97,7 @@ 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.ts b/apps/sim/lib/copilot/tools/server/files/file-preview.ts index ecb9fdf08e2..698d524d463 100644 --- a/apps/sim/lib/copilot/tools/server/files/file-preview.ts +++ b/apps/sim/lib/copilot/tools/server/files/file-preview.ts @@ -141,15 +141,30 @@ 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. + */ +export interface WorkspaceFilePreviewBase { + text: string + baseVersion: number +} + export async function loadWorkspaceFileTextForPreview( workspaceId: string, fileId: string -): Promise { +): Promise { try { const record = await getWorkspaceFile(workspaceId, fileId) if (!record) return undefined const buffer = await fetchWorkspaceFileBuffer(record) - return buffer.toString('utf-8') + return { + text: buffer.toString('utf-8'), + baseVersion: (record.contentUpdatedAt ?? record.updatedAt).getTime(), + } } catch (error) { logger.warn('Failed to load workspace file text for preview', { workspaceId, diff --git a/apps/sim/lib/realtime/notify.test.ts b/apps/sim/lib/realtime/notify.test.ts index 005637ebe44..20689dd745d 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 { mergeEditIntoLiveFileDoc } from './notify' +import { isLiveDocMergeInFlight, mergeEditIntoLiveFileDoc } from './notify' describe('mergeEditIntoLiveFileDoc', () => { afterEach(() => { @@ -17,25 +17,128 @@ describe('mergeEditIntoLiveFileDoc', () => { const fetchMock = vi.fn().mockResolvedValue({ ok: true }) vi.stubGlobal('fetch', fetchMock) - await mergeEditIntoLiveFileDoc('file-1', '# hello', 42) + await mergeEditIntoLiveFileDoc('file-1', '# hello', { version: 42 }) expect(fetchMock).toHaveBeenCalledWith( 'http://realtime/api/file-doc/apply-edit', expect.objectContaining({ method: 'POST', headers: expect.objectContaining({ 'x-api-key': 'secret' }), + // A durable write sends `version`; the undefined `baseVersion` is dropped by 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(mergeEditIntoLiveFileDoc('file-1', '# hello', 42)).resolves.toBeUndefined() + await expect( + mergeEditIntoLiveFileDoc('file-1', '# hello', { version: 42 }) + ).resolves.toBeUndefined() }) it('never throws on a non-2xx response', async () => { vi.stubGlobal('fetch', vi.fn().mockResolvedValue({ ok: false, status: 503 })) - await expect(mergeEditIntoLiveFileDoc('file-1', '# hello', 42)).resolves.toBeUndefined() + await expect( + mergeEditIntoLiveFileDoc('file-1', '# hello', { version: 42 }) + ).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 = () => {} + const fetchMock = vi + .fn() + .mockImplementationOnce(() => new Promise((resolve) => (resolveStream = resolve))) + .mockResolvedValue({ ok: true }) + vi.stubGlobal('fetch', fetchMock) + + const stream = mergeEditIntoLiveFileDoc('file-durable', 'partial') // versionless, in flight + await Promise.resolve() + const durable = mergeEditIntoLiveFileDoc('file-durable', 'final content', { version: 100 }) // versioned + 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. + expect(fetchMock).toHaveBeenCalledTimes(1) + + resolveStream({ ok: true }) + await stream + await durable + + // Only after the streaming merge completed does the durable (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 = [] + const resolvers: Array<() => void> = [] + vi.stubGlobal( + 'fetch', + vi.fn((_url: string, init: { body: string }) => { + applied.push(JSON.parse(init.body).version ?? 'stream') + return new Promise<{ ok: boolean }>((resolve) => + resolvers.push(() => resolve({ ok: true })) + ) + }) + ) + const flush = async () => { + for (let i = 0; i < 6; i++) await Promise.resolve() + } + + const s = mergeEditIntoLiveFileDoc('file-order', 's') // streaming, in flight + await flush() + // Two durable writes arrive while the streaming 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 + + resolvers[0]() // finish streaming → A applies next (not B) + await flush() + expect(applied).toEqual(['stream', 1]) + + resolvers[1]() // finish A → B applies after A + await flush() + expect(applied).toEqual(['stream', 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 92a4eb2f920..a67f86401c4 100644 --- a/apps/sim/lib/realtime/notify.ts +++ b/apps/sim/lib/realtime/notify.ts @@ -109,6 +109,20 @@ export async function notifyFolderResourceChanged( await FOLDER_RESOURCE_NOTIFIERS[resourceType]?.(workspaceId) } +/** + * 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). + */ +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. */ + 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 @@ -121,21 +135,73 @@ export async function notifyFolderResourceChanged( * 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. * - * Awaited (not fire-and-forget) 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. + * 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. + * + * `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. + * + * 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. */ export async function mergeEditIntoLiveFileDoc( fileId: string, markdown: string, - version: number + order: LiveFileDocMergeOrder = {} +): Promise { + const tail = liveDocMergeChain.get(fileId) ?? Promise.resolve() + const run = tail.then(() => applyLiveFileDocMerge(fileId, markdown, order)) + liveDocMergeChain.set(fileId, run) + try { + await run + } finally { + if (liveDocMergeChain.get(fileId) === run) liveDocMergeChain.delete(fileId) + } +} + +/** Per file, the tail of the serialized merge chain (each merge applies after it); never rejects + * 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, + markdown: string, + order: LiveFileDocMergeOrder ): Promise { try { const response = await fetch(`${getSocketServerUrl()}/api/file-doc/apply-edit`, { method: 'POST', headers: { 'Content-Type': 'application/json', 'x-api-key': env.INTERNAL_API_SECRET }, - // `version` is the durable `updatedAt` (epoch ms) this markdown was written with — the relay - // records it as the version its live doc now incorporates (see the persist If-Match guard). - body: JSON.stringify({ fileId, markdown, version }), + // `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. + body: JSON.stringify({ + fileId, + markdown, + version: order.version, + baseVersion: order.baseVersion, + }), signal: AbortSignal.timeout(APPLY_EDIT_TIMEOUT_MS), }) if (!response.ok) { diff --git a/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts b/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts index becee921859..c0dbbbecce9 100644 --- a/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts +++ b/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts @@ -1223,11 +1223,9 @@ export async function updateWorkspaceFileContent( // incorporates this durable version — the collab persist's optimistic-concurrency guard then won't // treat this (already-merged) write as an out-of-band conflict. Must be the SAME field the CAS // guards on (`contentUpdatedAt`), not `updatedAt`, or the relay's token wouldn't match the CAS. - await mergeEditIntoLiveFileDoc( - fileId, - content.toString('utf-8'), - finalized.file.contentUpdatedAt.getTime() - ) + await mergeEditIntoLiveFileDoc(fileId, content.toString('utf-8'), { + version: finalized.file.contentUpdatedAt.getTime(), + }) } const pathPrefix = getServePathPrefix()