From 9b6fce6cd42ae8b3c043bb3d7a7cfccc5e7513e1 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 15:39:35 -0700 Subject: [PATCH 01/14] feat(copilot): stream file edits into the live collaborative Y.Doc MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Copilot's file edits previously only reached the live doc once, at the final edit_content write, so a collaborative editor watching the file saw nothing until completion (streaming looked broken) and the client-side preview path was suppressed in collab mode. Make copilot a CRDT peer: as it streams append/update/patch content, merge the growing markdown into the file's live Y.Doc via the existing apply-edit path (a minimal updateYFragment diff, concurrent-edit-safe), throttled to ~250ms. version is omitted for these intermediate merges — they advance the live doc for viewers but are not durable checkpoints; the final edit_content write carries the real contentUpdatedAt and reconciles the durable file. Per the relay's persist gating, server-internal merges never schedule a persist, so a copilot-only stream produces zero intermediate file writes. - notify.ts: mergeEditIntoLiveFileDoc version is now optional (streaming omits it). - file-preview-adapter.ts: throttled live-doc merge at the edit_content stream hook. --- .../request/go/file-preview-adapter.ts | 33 +++++++++++++++++++ apps/sim/lib/realtime/notify.ts | 12 +++++-- 2 files changed, 42 insertions(+), 3 deletions(-) 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..4002b22e33a 100644 --- a/apps/sim/lib/copilot/request/go/file-preview-adapter.ts +++ b/apps/sim/lib/copilot/request/go/file-preview-adapter.ts @@ -24,6 +24,7 @@ import { buildFilePreviewText, loadWorkspaceFileTextForPreview, } from '@/lib/copilot/tools/server/files/file-preview' +import { mergeEditIntoLiveFileDoc } from '@/lib/realtime/notify' import { resolveWorkspaceFileReference } from '@/lib/uploads/contexts/workspace/workspace-file-manager' const logger = createLogger('CopilotFilePreviewAdapter') @@ -40,6 +41,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 +55,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) @@ -406,6 +415,7 @@ export async function processFilePreviewStreamEvent(input: { session, lastEmittedPreviewText: '', lastSnapshotAt: 0, + lastLiveMergeAt: 0, }) await persistFilePreviewSession(session) @@ -477,6 +487,7 @@ export async function processFilePreviewStreamEvent(input: { session, lastEmittedPreviewText: '', lastSnapshotAt: 0, + lastLiveMergeAt: 0, }) await persistFilePreviewSession(session) @@ -546,6 +557,7 @@ export async function processFilePreviewStreamEvent(input: { session: nextSession, lastEmittedPreviewText: previewText, lastSnapshotAt: Date.now(), + lastLiveMergeAt: 0, }) await persistFilePreviewSession(nextSession) await emitPreviewEvent(streamEvent, options, { @@ -579,6 +591,7 @@ export async function processFilePreviewStreamEvent(input: { session: buildPreviewSessionFromIntent(streamId, editIntent), lastEmittedPreviewText: '', lastSnapshotAt: 0, + lastLiveMergeAt: 0, } if ( @@ -637,6 +650,21 @@ 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 as a minimal `updateYFragment` diff. Throttled so we don't merge per + // token; fire-and-forget so a slow/unreachable relay never stalls the stream; version omitted + // because intermediate content is not a durable 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 (relay reports `applied: false`). + const dueForLiveMerge = + nextSession.fileId !== undefined && + now - currentPreview.lastLiveMergeAt >= LIVE_DOC_MERGE_THROTTLE_MS + const nextLiveMergeAt = dueForLiveMerge ? now : currentPreview.lastLiveMergeAt + if (dueForLiveMerge && nextSession.fileId) { + void mergeEditIntoLiveFileDoc(nextSession.fileId, nextSession.previewText) + } + if ( nextSession.operation === 'patch' && now - currentPreview.lastSnapshotAt < PATCH_PREVIEW_SNAPSHOT_INTERVAL_MS @@ -645,6 +673,7 @@ export async function processFilePreviewStreamEvent(input: { session: nextSession, lastEmittedPreviewText: currentPreview.lastEmittedPreviewText, lastSnapshotAt: currentPreview.lastSnapshotAt, + lastLiveMergeAt: nextLiveMergeAt, }) } else { const previewUpdate = buildPreviewContentUpdate( @@ -659,6 +688,7 @@ export async function processFilePreviewStreamEvent(input: { session: nextSession, lastEmittedPreviewText: nextSession.previewText, lastSnapshotAt: previewUpdate.lastSnapshotAt, + lastLiveMergeAt: nextLiveMergeAt, }) await emitPreviewEvent(streamEvent, options, { @@ -680,6 +710,7 @@ export async function processFilePreviewStreamEvent(input: { session: currentPreview.session, lastEmittedPreviewText: currentPreview.lastEmittedPreviewText, lastSnapshotAt: currentPreview.lastSnapshotAt, + lastLiveMergeAt: currentPreview.lastLiveMergeAt, }) } } @@ -713,6 +744,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 +776,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/realtime/notify.ts b/apps/sim/lib/realtime/notify.ts index 92a4eb2f920..aa9c8190963 100644 --- a/apps/sim/lib/realtime/notify.ts +++ b/apps/sim/lib/realtime/notify.ts @@ -123,18 +123,24 @@ export async function notifyFolderResourceChanged( * * 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. + * + * `version` is the durable `contentUpdatedAt` (epoch ms) this markdown was written with, for a durable + * write. Omit it for a STREAMING intermediate merge (the copilot stream mid-flight): intermediate + * content advances the live doc for viewers but is not a durable checkpoint, so the relay leaves its + * synced version pinned to the last durable write — which is exactly the copilot tool's final + * `edit_content` write, carrying the real version, that reconciles the durable file. */ export async function mergeEditIntoLiveFileDoc( fileId: string, markdown: string, - version: number + version?: number ): 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). + // A durable `version` (the durable `updatedAt` epoch ms) records the version the live doc now + // incorporates (the persist If-Match guard); omitted for a streaming intermediate merge. body: JSON.stringify({ fileId, markdown, version }), signal: AbortSignal.timeout(APPLY_EDIT_TIMEOUT_MS), }) From 83402901368024129d702ed23e07e654e0696276 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 15:51:20 -0700 Subject: [PATCH 02/14] fix(copilot): order + gate streaming live-doc merges; fast collab first render MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Harden the streaming merge (adversarial review): - Order + bound: dispatch through a per-file in-flight guard (drop-while-in-flight) so a stale out-of-order snapshot can never land after a newer one and regress the doc, and relay load is capped at one request per file regardless of rate. - No wipe: gate append/patch on the base file content having loaded — a base-less snapshot would diff to a delete-everything wipe of the seeded doc; update streams a full rewrite from scratch and needs no base. - Markdown-only gate: non-markdown files have no collaborative room, so skip the wasted relay round-trip. Fast collab first render (Issue 2): render the already-fetched markdown read-only via generateHTML while the collaborative doc seeds, with the editor mounted-but- hidden in the same layout box for a seamless swap on collabReady. Pure HTML — it never touches the Y.Doc (client seeding duplicates the doc), and generateHTML escapes text (raw-HTML snippets render escaped), so no XSS. --- .../rich-markdown-editor.tsx | 27 ++++++++++++-- .../request/go/file-preview-adapter.ts | 36 +++++++++++++++---- 2 files changed, 55 insertions(+), 8 deletions(-) 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..ee33df60c04 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 @@ -947,9 +959,20 @@ export function LoadedRichMarkdownEditor({ if (images.length > 0) void insertImagesRef.current(images, at) }} /> + {collaborationEnabled && !collabReady && 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/copilot/request/go/file-preview-adapter.ts b/apps/sim/lib/copilot/request/go/file-preview-adapter.ts index 4002b22e33a..5a4982a7959 100644 --- a/apps/sim/lib/copilot/request/go/file-preview-adapter.ts +++ b/apps/sim/lib/copilot/request/go/file-preview-adapter.ts @@ -26,6 +26,7 @@ import { } from '@/lib/copilot/tools/server/files/file-preview' import { 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') @@ -62,6 +63,24 @@ const DELTA_PREVIEW_CHECKPOINT_INTERVAL_MS = 1000 */ const LIVE_DOC_MERGE_THROTTLE_MS = 250 +/** Files with a live-doc merge currently in flight — one at a time per file. */ +const liveDocMergeInFlight = new Set() + +/** + * Merge the growing copilot content into a file's live collaborative Y.Doc, at most one in flight per + * file. Dropping a snapshot while the previous merge is still in flight (rather than queuing it) keeps + * merges strictly ordered — so a stale, out-of-order snapshot can never land after a newer one and + * regress the doc — and bounds relay load to one request per file regardless of stream rate. Any + * snapshot dropped this way is superseded by the next throttled tick, and the durable `edit_content` + * write is the final, authoritative merge. Fire-and-forget: {@link mergeEditIntoLiveFileDoc} never + * throws, so the stream is never stalled by a slow or unreachable relay. + */ +function streamMergeIntoLiveDoc(fileId: string, markdown: string): void { + if (liveDocMergeInFlight.has(fileId)) return + liveDocMergeInFlight.add(fileId) + void mergeEditIntoLiveFileDoc(fileId, markdown).finally(() => liveDocMergeInFlight.delete(fileId)) +} + function asJsonRecord(value: unknown): JsonRecord | undefined { return value && typeof value === 'object' && !Array.isArray(value) ? (value as JsonRecord) @@ -652,17 +671,22 @@ export async function processFilePreviewStreamEvent(input: { // 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 as a minimal `updateYFragment` diff. Throttled so we don't merge per - // token; fire-and-forget so a slow/unreachable relay never stalls the stream; version omitted - // because intermediate content is not a durable 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 (relay reports `applied: false`). + // CRDT peer, applied as a minimal `updateYFragment` diff. Throttled per file; intermediate + // content is not a durable 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; and for `append`/`patch`, only once the base file content has + // loaded — a base-less snapshot would diff to a delete-everything wipe of the seeded doc. + // `update` streams a full rewrite from scratch, so it needs no base. const dueForLiveMerge = nextSession.fileId !== undefined && + isMarkdownFile({ name: nextSession.fileName ?? '' }) && + (editIntent.operation === 'update' || + currentPreview.session.baseContent !== undefined) && now - currentPreview.lastLiveMergeAt >= LIVE_DOC_MERGE_THROTTLE_MS const nextLiveMergeAt = dueForLiveMerge ? now : currentPreview.lastLiveMergeAt if (dueForLiveMerge && nextSession.fileId) { - void mergeEditIntoLiveFileDoc(nextSession.fileId, nextSession.previewText) + streamMergeIntoLiveDoc(nextSession.fileId, nextSession.previewText) } if ( From 3dc137859643018eb0d8afad0e3633c467cccf50 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 15:57:51 -0700 Subject: [PATCH 03/14] test(copilot): cover streaming file edits into the live collaborative Y.Doc Drives edit_content args_delta stream events through processFilePreviewStreamEvent and asserts the live-doc merge: fires with the growing FULL previewText and no version arg; is throttled (~250ms per file); is skipped for non-markdown files and for a base-less append (the delete-everything wipe guard); and runs at most one-in-flight per file. Verified to fail if any gate/guard is removed. --- .../request/go/file-preview-adapter.test.ts | 211 ++++++++++++++++++ 1 file changed, 211 insertions(+) create mode 100644 apps/sim/lib/copilot/request/go/file-preview-adapter.test.ts 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..23efe2808c2 --- /dev/null +++ b/apps/sim/lib/copilot/request/go/file-preview-adapter.test.ts @@ -0,0 +1,211 @@ +/** + * @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 } = vi.hoisted(() => ({ + mergeEditIntoLiveFileDocMock: + vi.fn<(fileId: string, markdown: string, version?: number) => Promise>(), +})) + +const { peekFileIntentMock } = vi.hoisted(() => ({ + peekFileIntentMock: vi.fn(), +})) + +vi.mock('@/lib/realtime/notify', () => ({ + mergeEditIntoLiveFileDoc: mergeEditIntoLiveFileDocMock, +})) + +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' + +/** 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) + peekFileIntentMock.mockResolvedValue(undefined) + 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 (no version arg) into the live doc as it streams', async () => { + const intent = makeIntent({ operation: 'update', 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) + expect(mergeEditIntoLiveFileDocMock.mock.calls[0]).toEqual(['file-grow', 'Hello']) + // The second merge carries the GROWN full text, never a diff. + expect(mergeEditIntoLiveFileDocMock.mock.calls[1]).toEqual(['file-grow', 'Hello world']) + // No version arg on either streaming merge — the durable version rides the final edit_content write. + expect(mergeEditIntoLiveFileDocMock.mock.calls[0]).toHaveLength(2) + expect(mergeEditIntoLiveFileDocMock.mock.calls[1]).toHaveLength(2) + }) + + it('throttles merges: two deltas within LIVE_DOC_MERGE_THROTTLE_MS yield one merge', async () => { + const intent = makeIntent({ + operation: 'update', + 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) + expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledWith('file-throttle', 'Hel') + }) + + it('does not merge for a non-markdown file (no collaborative room)', async () => { + const intent = makeIntent({ operation: 'update', fileId: 'file-txt', fileName: 'notes.txt' }) + + await drive(editContentDelta('{"content":"plain text body'), 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('keeps at most one merge in-flight per file (second delta past the window is dropped while pending)', async () => { + let resolvePending: (() => void) | undefined + mergeEditIntoLiveFileDocMock.mockReturnValue( + new Promise((resolve) => { + resolvePending = () => resolve() + }) + ) + const intent = makeIntent({ + operation: 'update', + fileId: 'file-inflight', + fileName: 'notes.md', + }) + + await drive(editContentDelta('{"content":"first'), intent) + await flushMicrotasks() + expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledTimes(1) + + // Advance WELL past the throttle window so the throttle is not what blocks the second call — + // only the in-flight guard can. The first merge promise is still pending. + nowMs += 5000 + await drive(editContentDelta(' second'), intent) + await flushMicrotasks() + + expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledTimes(1) + + // Resolve so the in-flight slot clears and does not leak into other tests. + resolvePending?.() + await flushMicrotasks() + }) +}) From 42269878b42d0d6e12f13d5b5f94c5d8b5b0d34b Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 16:05:54 -0700 Subject: [PATCH 04/14] fix(collab-doc): coordinate live-doc merge ordering in one place; close durable-clobber race The second review found a residual: the durable edit_content write went through a different path than the adapter's in-flight guard, so a late straggler streaming merge could land after it and, via a persist, clobber the durable file's tail. Move the per-file coordination into mergeEditIntoLiveFileDoc (the one place both the streaming and durable paths call): a streaming (versionless) merge is dropped while one is in flight for the file; a durable (versioned) write instead WAITS for the in-flight streaming merge, so the final content is always the last merge applied and can't be regressed by a straggler. Simplifies the adapter (drops its Set + helper). Relocate the one-in-flight test to notify.test.ts (streaming-drops-while-busy + durable-waits-then-applies-last); the adapter test keeps throttle/gates/previewText. --- .../request/go/file-preview-adapter.test.ts | 30 ------------ .../request/go/file-preview-adapter.ts | 35 ++++---------- apps/sim/lib/realtime/notify.test.ts | 48 +++++++++++++++++++ apps/sim/lib/realtime/notify.ts | 28 +++++++++++ 4 files changed, 85 insertions(+), 56 deletions(-) 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 23efe2808c2..bee5fd63b2c 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 @@ -178,34 +178,4 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { // 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('keeps at most one merge in-flight per file (second delta past the window is dropped while pending)', async () => { - let resolvePending: (() => void) | undefined - mergeEditIntoLiveFileDocMock.mockReturnValue( - new Promise((resolve) => { - resolvePending = () => resolve() - }) - ) - const intent = makeIntent({ - operation: 'update', - fileId: 'file-inflight', - fileName: 'notes.md', - }) - - await drive(editContentDelta('{"content":"first'), intent) - await flushMicrotasks() - expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledTimes(1) - - // Advance WELL past the throttle window so the throttle is not what blocks the second call — - // only the in-flight guard can. The first merge promise is still pending. - nowMs += 5000 - await drive(editContentDelta(' second'), intent) - await flushMicrotasks() - - expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledTimes(1) - - // Resolve so the in-flight slot clears and does not leak into other tests. - resolvePending?.() - await flushMicrotasks() - }) }) 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 5a4982a7959..c8720d4126a 100644 --- a/apps/sim/lib/copilot/request/go/file-preview-adapter.ts +++ b/apps/sim/lib/copilot/request/go/file-preview-adapter.ts @@ -63,24 +63,6 @@ const DELTA_PREVIEW_CHECKPOINT_INTERVAL_MS = 1000 */ const LIVE_DOC_MERGE_THROTTLE_MS = 250 -/** Files with a live-doc merge currently in flight — one at a time per file. */ -const liveDocMergeInFlight = new Set() - -/** - * Merge the growing copilot content into a file's live collaborative Y.Doc, at most one in flight per - * file. Dropping a snapshot while the previous merge is still in flight (rather than queuing it) keeps - * merges strictly ordered — so a stale, out-of-order snapshot can never land after a newer one and - * regress the doc — and bounds relay load to one request per file regardless of stream rate. Any - * snapshot dropped this way is superseded by the next throttled tick, and the durable `edit_content` - * write is the final, authoritative merge. Fire-and-forget: {@link mergeEditIntoLiveFileDoc} never - * throws, so the stream is never stalled by a slow or unreachable relay. - */ -function streamMergeIntoLiveDoc(fileId: string, markdown: string): void { - if (liveDocMergeInFlight.has(fileId)) return - liveDocMergeInFlight.add(fileId) - void mergeEditIntoLiveFileDoc(fileId, markdown).finally(() => liveDocMergeInFlight.delete(fileId)) -} - function asJsonRecord(value: unknown): JsonRecord | undefined { return value && typeof value === 'object' && !Array.isArray(value) ? (value as JsonRecord) @@ -671,13 +653,14 @@ export async function processFilePreviewStreamEvent(input: { // 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 as a minimal `updateYFragment` diff. Throttled per file; intermediate - // content is not a durable 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; and for `append`/`patch`, only once the base file content has - // loaded — a base-less snapshot would diff to a delete-everything wipe of the seeded doc. - // `update` streams a full rewrite from scratch, so it needs no base. + // CRDT peer, applied by the relay as a minimal `updateYFragment` diff. Throttled per file; + // fire-and-forget so a slow relay never stalls the stream (`mergeEditIntoLiveFileDoc` never + // throws, coordinates ordering per file, and treats a versionless call as a non-durable + // preview merge). 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; and for `append`/`patch`, only once the base file content has loaded — + // a base-less snapshot would diff to a delete-everything wipe of the seeded doc. `update` + // streams a full rewrite from scratch, so it needs no base. const dueForLiveMerge = nextSession.fileId !== undefined && isMarkdownFile({ name: nextSession.fileName ?? '' }) && @@ -686,7 +669,7 @@ export async function processFilePreviewStreamEvent(input: { now - currentPreview.lastLiveMergeAt >= LIVE_DOC_MERGE_THROTTLE_MS const nextLiveMergeAt = dueForLiveMerge ? now : currentPreview.lastLiveMergeAt if (dueForLiveMerge && nextSession.fileId) { - streamMergeIntoLiveDoc(nextSession.fileId, nextSession.previewText) + void mergeEditIntoLiveFileDoc(nextSession.fileId, nextSession.previewText) } if ( diff --git a/apps/sim/lib/realtime/notify.test.ts b/apps/sim/lib/realtime/notify.test.ts index 005637ebe44..ecb4bfc8bdf 100644 --- a/apps/sim/lib/realtime/notify.test.ts +++ b/apps/sim/lib/realtime/notify.test.ts @@ -38,4 +38,52 @@ describe('mergeEditIntoLiveFileDoc', () => { vi.stubGlobal('fetch', vi.fn().mockResolvedValue({ ok: false, status: 503 })) await expect(mergeEditIntoLiveFileDoc('file-1', '# hello', 42)).resolves.toBeUndefined() }) + + it('drops a streaming (versionless) merge while one is already in flight for the same file', async () => { + let resolveFirst: (value: { ok: boolean }) => void = () => {} + const fetchMock = vi + .fn() + .mockImplementationOnce(() => new Promise((resolve) => (resolveFirst = resolve))) + .mockResolvedValue({ ok: true }) + vi.stubGlobal('fetch', fetchMock) + + const first = mergeEditIntoLiveFileDoc('file-inflight', 'v1') // versionless → in flight (pending) + await Promise.resolve() + // A second versionless merge while the first is pending is dropped, not queued — so a stale + // snapshot can never land after a newer one and regress the doc. + await mergeEditIntoLiveFileDoc('file-inflight', 'v2') + expect(fetchMock).toHaveBeenCalledTimes(1) + + resolveFirst({ ok: true }) + await first + }) + + 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', 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 }) + ) + }) }) diff --git a/apps/sim/lib/realtime/notify.ts b/apps/sim/lib/realtime/notify.ts index aa9c8190963..547bc6805e6 100644 --- a/apps/sim/lib/realtime/notify.ts +++ b/apps/sim/lib/realtime/notify.ts @@ -129,11 +129,39 @@ export async function notifyFolderResourceChanged( * content advances the live doc for viewers but is not a durable checkpoint, so the relay leaves its * synced version pinned to the last durable write — which is exactly the copilot tool's final * `edit_content` write, carrying the real version, that reconciles the durable file. + * + * Concurrency is coordinated per file so that ordering can never regress the doc: a STREAMING + * (versionless) merge is DROPPED while another is already in flight for the file — the next throttled + * caller sends the latest snapshot, and a stale snapshot can never land after a newer one; a DURABLE + * (versioned) write instead WAITS for the in-flight streaming merge, so the final content is always the + * last merge applied and cannot be clobbered by a late straggler. */ export async function mergeEditIntoLiveFileDoc( fileId: string, markdown: string, version?: number +): Promise { + const pending = liveDocMergeInFlight.get(fileId) + if (version === undefined && pending) return + if (pending) await pending + + const run = applyLiveFileDocMerge(fileId, markdown, version) + liveDocMergeInFlight.set(fileId, run) + try { + await run + } finally { + if (liveDocMergeInFlight.get(fileId) === run) liveDocMergeInFlight.delete(fileId) + } +} + +/** Files with a live-doc merge in flight → the running merge promise (never rejects). */ +const liveDocMergeInFlight = new Map>() + +/** POST the merge to the relay. Never throws (a live-doc merge is best-effort). */ +async function applyLiveFileDocMerge( + fileId: string, + markdown: string, + version?: number ): Promise { try { const response = await fetch(`${getSocketServerUrl()}/api/file-doc/apply-edit`, { From 68acab6cadd6f3bc4cbeee3331c0c91fc736d869 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 16:31:22 -0700 Subject: [PATCH 05/14] =?UTF-8?q?fix(copilot):=20address=20review=20?= =?UTF-8?q?=E2=80=94=20order=20merges,=20exclude=20update,=20gate=20thrott?= =?UTF-8?q?le,=20unhide=20stream?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round on #6108: - Greptile P1 (durable merges lose ordering): serialize ALL merges per file on one chain in mergeEditIntoLiveFileDoc (each chains after the current tail), so concurrent durable writes can't resume-and-fire out of order. notify now exposes isLiveDocMergeInFlight. - Cursor High (update stream blanks the doc): only append/patch stream — they build on the loaded base; update is a from-scratch rewrite whose partial snapshot would diff the full doc toward a fragment, so it applies atomically at the durable write. - Cursor Medium (throttle advances on a dropped merge): the adapter gates on !isLiveDocMergeInFlight, so the send throttle advances only on an actual dispatch — no lag, no backlog behind a slow relay. - Cursor Medium (placeholder hides a live stream): show the fast-render placeholder only when not streaming, so a stream that starts before the doc seeds shows through the editor. - Soften merge.ts/notify.ts comments per the lifecycle audit: only UNTOUCHED regions are preserved; a region the merge rewrites reconciles toward copilot's content. Tests updated: notify covers chain ordering + isLiveDocMergeInFlight; adapter covers append streaming, throttle, non-markdown/base-less/update skips, and the in-flight skip. --- .../rich-markdown-editor.tsx | 9 ++- apps/sim/lib/collab-doc/merge.ts | 10 +-- .../request/go/file-preview-adapter.test.ts | 52 ++++++++++---- .../request/go/file-preview-adapter.ts | 28 +++++--- apps/sim/lib/realtime/notify.test.ts | 67 ++++++++++++++----- apps/sim/lib/realtime/notify.ts | 42 +++++++----- 6 files changed, 148 insertions(+), 60 deletions(-) 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 ee33df60c04..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 @@ -935,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) }} /> - {collaborationEnabled && !collabReady && placeholderHtml && ( + {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 index bee5fd63b2c..2932635fa5a 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 @@ -9,9 +9,10 @@ import { MothershipStreamV1ToolPhase, } from '@/lib/copilot/generated/mothership-stream-v1' -const { mergeEditIntoLiveFileDocMock } = vi.hoisted(() => ({ +const { mergeEditIntoLiveFileDocMock, isLiveDocMergeInFlightMock } = vi.hoisted(() => ({ mergeEditIntoLiveFileDocMock: vi.fn<(fileId: string, markdown: string, version?: number) => Promise>(), + isLiveDocMergeInFlightMock: vi.fn<(fileId: string) => boolean>(), })) const { peekFileIntentMock } = vi.hoisted(() => ({ @@ -20,6 +21,7 @@ const { peekFileIntentMock } = vi.hoisted(() => ({ vi.mock('@/lib/realtime/notify', () => ({ mergeEditIntoLiveFileDoc: mergeEditIntoLiveFileDocMock, + isLiveDocMergeInFlight: isLiveDocMergeInFlightMock, })) vi.mock('@/lib/copilot/tools/server/files/file-intent-store', () => ({ @@ -95,7 +97,9 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { beforeEach(() => { vi.clearAllMocks() mergeEditIntoLiveFileDocMock.mockResolvedValue(undefined) - peekFileIntentMock.mockResolvedValue(undefined) + isLiveDocMergeInFlightMock.mockReturnValue(false) + // Default: an append/patch base is available (a non-empty file), so the base-present gate passes. + peekFileIntentMock.mockResolvedValue({ existingContent: 'Base.' }) state = createFilePreviewAdapterState() nowMs = 1_000_000 vi.spyOn(Date, 'now').mockImplementation(() => nowMs) @@ -120,7 +124,7 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { } it('merges the growing full content (no version arg) into the live doc as it streams', async () => { - const intent = makeIntent({ operation: 'update', fileId: 'file-grow', fileName: 'notes.md' }) + const intent = makeIntent({ operation: 'append', fileId: 'file-grow', fileName: 'notes.md' }) await drive(editContentDelta('{"content":"Hello'), intent) await flushMicrotasks() @@ -131,17 +135,21 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { await flushMicrotasks() expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledTimes(2) - expect(mergeEditIntoLiveFileDocMock.mock.calls[0]).toEqual(['file-grow', 'Hello']) - // The second merge carries the GROWN full text, never a diff. - expect(mergeEditIntoLiveFileDocMock.mock.calls[1]).toEqual(['file-grow', 'Hello world']) - // No version arg on either streaming merge — the durable version rides the final edit_content write. - expect(mergeEditIntoLiveFileDocMock.mock.calls[0]).toHaveLength(2) - expect(mergeEditIntoLiveFileDocMock.mock.calls[1]).toHaveLength(2) + const [first, second] = mergeEditIntoLiveFileDocMock.mock.calls + // A full-file snapshot (base + streamed), never a diff; it grows across deltas; no version arg on + // either streaming merge — the durable version rides the final edit_content write. + expect(first[0]).toBe('file-grow') + expect(first).toHaveLength(2) + expect(first[1]).toContain('Base.') + expect(first[1]).toContain('Hello') + expect(second).toHaveLength(2) + expect(second[1]).toContain('Hello world') + expect(second[1].length).toBeGreaterThan(first[1].length) }) it('throttles merges: two deltas within LIVE_DOC_MERGE_THROTTLE_MS yield one merge', async () => { const intent = makeIntent({ - operation: 'update', + operation: 'append', fileId: 'file-throttle', fileName: 'notes.md', }) @@ -155,11 +163,10 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { await flushMicrotasks() expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledTimes(1) - expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledWith('file-throttle', 'Hel') }) it('does not merge for a non-markdown file (no collaborative room)', async () => { - const intent = makeIntent({ operation: 'update', fileId: 'file-txt', fileName: 'notes.txt' }) + const intent = makeIntent({ operation: 'append', fileId: 'file-txt', fileName: 'notes.txt' }) await drive(editContentDelta('{"content":"plain text body'), intent) await flushMicrotasks() @@ -178,4 +185,25 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { // 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() + }) }) 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 c8720d4126a..0030cb2f6c2 100644 --- a/apps/sim/lib/copilot/request/go/file-preview-adapter.ts +++ b/apps/sim/lib/copilot/request/go/file-preview-adapter.ts @@ -24,7 +24,7 @@ import { buildFilePreviewText, loadWorkspaceFileTextForPreview, } from '@/lib/copilot/tools/server/files/file-preview' -import { mergeEditIntoLiveFileDoc } from '@/lib/realtime/notify' +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' @@ -653,19 +653,25 @@ export async function processFilePreviewStreamEvent(input: { // 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. Throttled per file; - // fire-and-forget so a slow relay never stalls the stream (`mergeEditIntoLiveFileDoc` never - // throws, coordinates ordering per file, and treats a versionless call as a non-durable - // preview merge). 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; and for `append`/`patch`, only once the base file content has loaded — - // a base-less snapshot would diff to a delete-everything wipe of the seeded doc. `update` - // streams a full rewrite from scratch, so it needs no base. + // CRDT peer, applied by the relay as a minimal `updateYFragment` diff. Fire-and-forget so a + // slow relay never stalls the stream; `mergeEditIntoLiveFileDoc` serializes ordering per file + // and treats a versionless call as a non-durable preview merge (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. const dueForLiveMerge = nextSession.fileId !== undefined && isMarkdownFile({ name: nextSession.fileName ?? '' }) && - (editIntent.operation === 'update' || - currentPreview.session.baseContent !== undefined) && + (editIntent.operation === 'append' || editIntent.operation === 'patch') && + currentPreview.session.baseContent !== undefined && + !isLiveDocMergeInFlight(nextSession.fileId) && now - currentPreview.lastLiveMergeAt >= LIVE_DOC_MERGE_THROTTLE_MS const nextLiveMergeAt = dueForLiveMerge ? now : currentPreview.lastLiveMergeAt if (dueForLiveMerge && nextSession.fileId) { diff --git a/apps/sim/lib/realtime/notify.test.ts b/apps/sim/lib/realtime/notify.test.ts index ecb4bfc8bdf..bc998b4e05c 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(() => { @@ -39,23 +39,23 @@ describe('mergeEditIntoLiveFileDoc', () => { await expect(mergeEditIntoLiveFileDoc('file-1', '# hello', 42)).resolves.toBeUndefined() }) - it('drops a streaming (versionless) merge while one is already in flight for the same file', async () => { - let resolveFirst: (value: { ok: boolean }) => void = () => {} - const fetchMock = vi - .fn() - .mockImplementationOnce(() => new Promise((resolve) => (resolveFirst = resolve))) - .mockResolvedValue({ ok: true }) - vi.stubGlobal('fetch', fetchMock) + 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))) + ) - const first = mergeEditIntoLiveFileDoc('file-inflight', 'v1') // versionless → in flight (pending) + expect(isLiveDocMergeInFlight('file-flight')).toBe(false) + const run = mergeEditIntoLiveFileDoc('file-flight', 'v1') await Promise.resolve() - // A second versionless merge while the first is pending is dropped, not queued — so a stale - // snapshot can never land after a newer one and regress the doc. - await mergeEditIntoLiveFileDoc('file-inflight', 'v2') - expect(fetchMock).toHaveBeenCalledTimes(1) + // 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) - resolveFirst({ ok: true }) - await first + 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 () => { @@ -86,4 +86,41 @@ describe('mergeEditIntoLiveFileDoc', () => { 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', 1) + const b = mergeEditIntoLiveFileDoc('file-order', 'b', 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 547bc6805e6..9517bfaa76e 100644 --- a/apps/sim/lib/realtime/notify.ts +++ b/apps/sim/lib/realtime/notify.ts @@ -121,8 +121,9 @@ 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. * * `version` is the durable `contentUpdatedAt` (epoch ms) this markdown was written with, for a durable * write. Omit it for a STREAMING intermediate merge (the copilot stream mid-flight): intermediate @@ -130,32 +131,41 @@ export async function notifyFolderResourceChanged( * synced version pinned to the last durable write — which is exactly the copilot tool's final * `edit_content` write, carrying the real version, that reconciles the durable file. * - * Concurrency is coordinated per file so that ordering can never regress the doc: a STREAMING - * (versionless) merge is DROPPED while another is already in flight for the file — the next throttled - * caller sends the latest snapshot, and a stale snapshot can never land after a newer one; a DURABLE - * (versioned) write instead WAITS for the in-flight streaming merge, so the final content is always the - * last merge applied and cannot be clobbered by a late straggler. + * Merges for a file run on a single serialized chain: each is chained after the current tail and + * applies strictly after it, so ordering can never regress the doc — a DURABLE (versioned) write + * always applies after any in-flight streaming merge AND after every earlier durable write, never + * concurrently. The final durable write is therefore always the last merge applied and cannot be + * clobbered by a late straggler. 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 ): Promise { - const pending = liveDocMergeInFlight.get(fileId) - if (version === undefined && pending) return - if (pending) await pending - - const run = applyLiveFileDocMerge(fileId, markdown, version) - liveDocMergeInFlight.set(fileId, run) + const tail = liveDocMergeChain.get(fileId) ?? Promise.resolve() + const run = tail.then(() => applyLiveFileDocMerge(fileId, markdown, version)) + liveDocMergeChain.set(fileId, run) try { await run } finally { - if (liveDocMergeInFlight.get(fileId) === run) liveDocMergeInFlight.delete(fileId) + if (liveDocMergeChain.get(fileId) === run) liveDocMergeChain.delete(fileId) } } -/** Files with a live-doc merge in flight → the running merge promise (never rejects). */ -const liveDocMergeInFlight = new Map>() +/** 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( From 63a4681de7a4665de0f5be9b3065767e16e8ae2e Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 16:45:13 -0700 Subject: [PATCH 06/14] fix(collab-doc): reject stale durable merges at the relay (cross-process ordering) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The in-process merge chain only orders merges within one apps/sim process. Two durable writes for the same file on DIFFERENT processes could reach the relay out of dispatch order; the relay recorded the version monotonically but still APPLIED the older markdown, regressing the live doc while the token stayed high (a later persist could then write the stale content back over the durable file). Enforce ordering at the relay — the single cross-process coordination point — using the existing Redis primitives: under the per-file Redis merge lock, read the cluster-wide synced version and SKIP a versioned merge that is not newer (a newer durable write already landed). Make recordVersion await setSyncedVersion so it is durable before the lock releases, so the next holder's staleness check reads a consistent value. Streaming (versionless) merges are unaffected — they carry no durable version and are ordered per-process by the caller. Adds a relay test asserting a stale/idempotent versioned merge returns 'stale' and never computes or publishes a diff. --- apps/realtime/src/handlers/file-doc.test.ts | 22 +++++++++++++ apps/realtime/src/handlers/file-doc.ts | 34 +++++++++++++++------ 2 files changed, 46 insertions(+), 10 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.test.ts b/apps/realtime/src/handlers/file-doc.test.ts index 4e0d9ffa2de..69b37047413 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -543,6 +543,28 @@ 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', 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', 50)).toBe('stale') + // The same version is idempotent — also skipped. + expect(await applyMarkdownToLiveFileDoc('file-1', '# same version', 100)).toBe('stale') + expect(mockFetchFileDocMerge).not.toHaveBeenCalled() + }) + 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..39da9434443 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -541,7 +541,7 @@ export function applyMarkdownToLiveFileDoc( fileId: string, markdown: string, version?: number -): Promise<'applied' | 'no-live-room' | 'merge-unavailable'> { +): 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. @@ -562,22 +562,30 @@ async function mergeMarkdownIntoRoom( fileId: string, markdown: string, version?: number -): Promise<'applied' | 'no-live-room' | 'merge-unavailable'> { +): 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. + 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) } + // A versioned (durable) merge that is not NEWER than the version the doc already incorporates is + // stale — a newer durable write already landed (possibly on another process, out of dispatch order). + // Diffing toward its older markdown would regress the live doc while the monotonic token stays high, + // so a later persist could write that stale content back over the durable file. Skip it. Streaming + // (versionless) merges carry no durable version and are ordered per-process by the caller. + const isStale = (current: number): boolean => version !== undefined && version <= current + if (store.enabled) { // Serialize merges to this file ACROSS tasks — the per-file chain above only covers this process. // Two copilot edits to the same file landing on different tasks must not diff the SAME shared base @@ -595,6 +603,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 +617,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 +627,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' } From 46212f3e40ad17c529a176a253d576823fff5b79 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 16:46:52 -0700 Subject: [PATCH 07/14] =?UTF-8?q?fix(copilot):=20match=20durable=20path=20?= =?UTF-8?q?=E2=80=94=20detect=20markdown=20by=20MIME=20type=20+=20name=20a?= =?UTF-8?q?t=20the=20stream=20gate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The streaming gate checked isMarkdownFile with only the filename, while the durable merge uses type + name — so a text/markdown file without a .md extension was skipped mid-stream (it self-corrected at the durable write). Pass editIntent.contentType so streaming detects the same set of markdown files as the durable path. --- apps/sim/lib/copilot/request/go/file-preview-adapter.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 0030cb2f6c2..d1f63bba6d1 100644 --- a/apps/sim/lib/copilot/request/go/file-preview-adapter.ts +++ b/apps/sim/lib/copilot/request/go/file-preview-adapter.ts @@ -668,7 +668,7 @@ export async function processFilePreviewStreamEvent(input: { // no-op — so a slow relay can't backlog stale snapshots or make the doc lag the stream. const dueForLiveMerge = nextSession.fileId !== undefined && - isMarkdownFile({ name: nextSession.fileName ?? '' }) && + isMarkdownFile({ type: editIntent.contentType, name: nextSession.fileName ?? '' }) && (editIntent.operation === 'append' || editIntent.operation === 'patch') && currentPreview.session.baseContent !== undefined && !isLiveDocMergeInFlight(nextSession.fileId) && From b0753add8bda352cab6667ee00a129f2453b53cc Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 16:57:26 -0700 Subject: [PATCH 08/14] test(copilot): assert throttle follow-through after an in-flight merge clears --- .../lib/copilot/request/go/file-preview-adapter.test.ts | 9 +++++++++ 1 file changed, 9 insertions(+) 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 2932635fa5a..8963b7b26d2 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 @@ -205,5 +205,14 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { 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') }) }) From e9622e9dbcd509bb3ce1bd9d32fec483f1123720 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 17:15:27 -0700 Subject: [PATCH 09/14] fix(collab-doc): order streaming merges by streamedAt so a late snapshot can't regress a newer durable write --- apps/realtime/src/handlers/file-doc.test.ts | 28 ++++++++++ apps/realtime/src/handlers/file-doc.ts | 28 ++++++---- apps/realtime/src/routes/http.ts | 6 ++- .../request/go/file-preview-adapter.test.ts | 18 +++++-- .../request/go/file-preview-adapter.ts | 14 +++-- apps/sim/lib/realtime/notify.test.ts | 29 +++++++--- apps/sim/lib/realtime/notify.ts | 54 +++++++++++++------ .../workspace/workspace-file-manager.ts | 8 ++- 8 files changed, 135 insertions(+), 50 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.test.ts b/apps/realtime/src/handlers/file-doc.test.ts index 69b37047413..338c3fc4267 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -565,6 +565,34 @@ describe('setupWorkspaceFileDocHandlers', () => { expect(mockFetchFileDocMerge).not.toHaveBeenCalled() }) + it('orders a streaming merge by streamedAt but never records it (stays behind the durable version)', 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', '# durable', 100)).toBe('applied') + + // A delayed streaming snapshot OLDER than that durable version — e.g. a throttled copilot snapshot + // from another process arriving late — is stale, so it can't regress the doc back toward its content. + // (`version` omitted, `streamedAt` passed as the 4th arg.) + expect(await applyMarkdownToLiveFileDoc('file-1', '# older stream', undefined, 50)).toBe( + 'stale' + ) + + // A streaming snapshot NEWER than the durable version applies (it advances the live view)... + expect(await applyMarkdownToLiveFileDoc('file-1', '# newer stream', undefined, 200)).toBe( + 'applied' + ) + // ...but it is never RECORDED as the synced version: a durable write between the two (150) still + // applies. Had the streaming 200 been recorded, 150 would have been rejected as stale. + expect(await applyMarkdownToLiveFileDoc('file-1', '# durable again', 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 39da9434443..1434832bbdb 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -540,14 +540,15 @@ const fileDocMergeChains = new Map>() export function applyMarkdownToLiveFileDoc( fileId: string, markdown: string, - version?: number + version?: number, + streamedAt?: number ): 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)) + .then(() => mergeMarkdownIntoRoom(name, fileId, markdown, version, streamedAt)) fileDocMergeChains.set( name, run.finally(() => { @@ -561,14 +562,17 @@ async function mergeMarkdownIntoRoom( name: string, fileId: string, markdown: string, - version?: number + version?: number, + streamedAt?: number ): 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. + // releases, so the next lock holder's staleness check (below) reads a consistent value. Only a durable + // `version` is recorded — a streaming `streamedAt` 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) @@ -579,12 +583,16 @@ async function mergeMarkdownIntoRoom( await store.setSyncedVersion(name, version) } - // A versioned (durable) merge that is not NEWER than the version the doc already incorporates is - // stale — a newer durable write already landed (possibly on another process, out of dispatch order). - // Diffing toward its older markdown would regress the live doc while the monotonic token stays high, - // so a later persist could write that stale content back over the durable file. Skip it. Streaming - // (versionless) merges carry no durable version and are ordered per-process by the caller. - const isStale = (current: number): boolean => version !== undefined && version <= current + // Position this merge on the file's monotonic version line: a durable write by its `version`, a + // streaming snapshot by its `streamedAt` production time. A merge not NEWER than the version the doc + // already incorporates is stale — a newer durable write already landed (possibly on another process, + // out of dispatch order). Diffing toward its older markdown would regress the live doc while the + // monotonic token stays high, so a later persist could write that stale content back over the durable + // file. Skip it. This is what stops a delayed streaming snapshot from clobbering a newer durable merge + // across processes (the per-process caller chain cannot). A merge with neither key is never stale. + const orderingVersion = version ?? streamedAt + const isStale = (current: number): boolean => + orderingVersion !== undefined && orderingVersion <= current if (store.enabled) { // Serialize merges to this file ACROSS tasks — the per-file chain above only covers this process. diff --git a/apps/realtime/src/routes/http.ts b/apps/realtime/src/routes/http.ts index fe824a70b86..a31f86e4b79 100644 --- a/apps/realtime/src/routes/http.ts +++ b/apps/realtime/src/routes/http.ts @@ -209,16 +209,18 @@ 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, streamedAt } = 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. + // `streamedAt` orders a streaming snapshot on the same version line without recording it. const result = await applyMarkdownToLiveFileDoc( fileId, markdown, - typeof version === 'number' ? version : undefined + typeof version === 'number' ? version : undefined, + typeof streamedAt === 'number' ? streamedAt : undefined ) res.writeHead(200, { 'Content-Type': 'application/json' }) res.end(JSON.stringify({ applied: result === 'applied' })) 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 8963b7b26d2..7519f61fe07 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 @@ -11,7 +11,13 @@ import { const { mergeEditIntoLiveFileDocMock, isLiveDocMergeInFlightMock } = vi.hoisted(() => ({ mergeEditIntoLiveFileDocMock: - vi.fn<(fileId: string, markdown: string, version?: number) => Promise>(), + vi.fn< + ( + fileId: string, + markdown: string, + order?: { version?: number; streamedAt?: number } + ) => Promise + >(), isLiveDocMergeInFlightMock: vi.fn<(fileId: string) => boolean>(), })) @@ -136,15 +142,17 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledTimes(2) const [first, second] = mergeEditIntoLiveFileDocMock.mock.calls - // A full-file snapshot (base + streamed), never a diff; it grows across deltas; no version arg on - // either streaming merge — the durable version rides the final edit_content write. + // A full-file snapshot (base + streamed), never a diff; it grows across deltas. Each streaming merge + // carries `streamedAt` (its wall-clock time) to order it — never `version`, which rides the final + // edit_content write; so the relay orders the snapshot without recording it as a durable checkpoint. expect(first[0]).toBe('file-grow') - expect(first).toHaveLength(2) expect(first[1]).toContain('Base.') expect(first[1]).toContain('Hello') - expect(second).toHaveLength(2) + expect(typeof first[2]?.streamedAt).toBe('number') + expect(first[2]?.version).toBeUndefined() expect(second[1]).toContain('Hello world') expect(second[1].length).toBeGreaterThan(first[1].length) + expect(typeof second[2]?.streamedAt).toBe('number') }) it('throttles merges: two deltas within LIVE_DOC_MERGE_THROTTLE_MS yield one merge', async () => { 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 d1f63bba6d1..f0daac5f828 100644 --- a/apps/sim/lib/copilot/request/go/file-preview-adapter.ts +++ b/apps/sim/lib/copilot/request/go/file-preview-adapter.ts @@ -654,10 +654,12 @@ export async function processFilePreviewStreamEvent(input: { // 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; `mergeEditIntoLiveFileDoc` serializes ordering per file - // and treats a versionless call as a non-durable preview merge (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`). + // slow relay never stalls the stream. Pass `streamedAt` (this snapshot's wall-clock time) so + // the relay orders it on the file's version line — a delayed snapshot older than a newer + // durable write, even from another process, is dropped rather than regressing the doc — but + // never records it as a durable 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 @@ -675,7 +677,9 @@ export async function processFilePreviewStreamEvent(input: { now - currentPreview.lastLiveMergeAt >= LIVE_DOC_MERGE_THROTTLE_MS const nextLiveMergeAt = dueForLiveMerge ? now : currentPreview.lastLiveMergeAt if (dueForLiveMerge && nextSession.fileId) { - void mergeEditIntoLiveFileDoc(nextSession.fileId, nextSession.previewText) + void mergeEditIntoLiveFileDoc(nextSession.fileId, nextSession.previewText, { + streamedAt: now, + }) } if ( diff --git a/apps/sim/lib/realtime/notify.test.ts b/apps/sim/lib/realtime/notify.test.ts index bc998b4e05c..67c4750b5a0 100644 --- a/apps/sim/lib/realtime/notify.test.ts +++ b/apps/sim/lib/realtime/notify.test.ts @@ -17,26 +17,43 @@ 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 `streamedAt` is dropped by JSON.stringify. body: JSON.stringify({ fileId: 'file-1', markdown: '# hello', version: 42 }), }) ) }) + it('sends streamedAt (not version) for a streaming snapshot', async () => { + const fetchMock = vi.fn().mockResolvedValue({ ok: true }) + vi.stubGlobal('fetch', fetchMock) + + await mergeEditIntoLiveFileDoc('file-1', '# hello', { streamedAt: 1234 }) + + // The relay orders the snapshot by streamedAt without recording it — version stays absent on the wire. + expect(fetchMock.mock.calls[0][1].body).toBe( + JSON.stringify({ fileId: 'file-1', markdown: '# hello', streamedAt: 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 () => { @@ -68,7 +85,7 @@ describe('mergeEditIntoLiveFileDoc', () => { const stream = mergeEditIntoLiveFileDoc('file-durable', 'partial') // versionless, in flight await Promise.resolve() - const durable = mergeEditIntoLiveFileDoc('file-durable', 'final content', 100) // versioned + const durable = mergeEditIntoLiveFileDoc('file-durable', 'final content', { version: 100 }) // versioned await Promise.resolve() await Promise.resolve() @@ -107,8 +124,8 @@ describe('mergeEditIntoLiveFileDoc', () => { 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', 1) - const b = mergeEditIntoLiveFileDoc('file-order', 'b', 2) + 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 diff --git a/apps/sim/lib/realtime/notify.ts b/apps/sim/lib/realtime/notify.ts index 9517bfaa76e..6ac8513eace 100644 --- a/apps/sim/lib/realtime/notify.ts +++ b/apps/sim/lib/realtime/notify.ts @@ -125,26 +125,40 @@ export async function notifyFolderResourceChanged( * streaming caller fires and forgets it. Bounded to {@link APPLY_EDIT_TIMEOUT_MS}, so it adds latency * only when the socket pod is unreachable. * - * `version` is the durable `contentUpdatedAt` (epoch ms) this markdown was written with, for a durable - * write. Omit it for a STREAMING intermediate merge (the copilot stream mid-flight): intermediate - * content advances the live doc for viewers but is not a durable checkpoint, so the relay leaves its - * synced version pinned to the last durable write — which is exactly the copilot tool's final - * `edit_content` write, carrying the real version, that reconciles the durable file. + * `order` positions this merge on the file's monotonic version line so a stale write never regresses + * the doc — the relay drops any merge not NEWER than the version the doc already incorporates: + * - `version` — a DURABLE write's `contentUpdatedAt` (epoch ms). Both orders the merge AND is recorded + * as the doc's synced version (the persist If-Match guard), so a later persist treats this write as + * synced rather than an out-of-band conflict. + * - `streamedAt` — the wall-clock time (epoch ms) a STREAMING snapshot was produced (the copilot stream + * mid-flight). Orders the merge so a delayed snapshot older than a newer durable write — possibly from + * another app process — is dropped, but is NEVER recorded: the synced version stays pinned to the last + * durable write, which is exactly the copilot tool's final `edit_content` write that reconciles the file. * - * Merges for a file run on a single serialized chain: each is chained after the current tail and - * applies strictly after it, so ordering can never regress the doc — a DURABLE (versioned) write - * always applies after any in-flight streaming merge AND after every earlier durable write, never - * concurrently. The final durable write is therefore always the last merge applied and cannot be - * clobbered by a late straggler. 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. + * Pass one or the other, never both. Passing neither applies the merge without ordering it (legacy). + * + * 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 the monotonic version above (durable version / + * streaming `streamedAt`) 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 interface LiveFileDocMergeOrder { + /** A durable write's `contentUpdatedAt` (epoch ms): orders the merge AND is recorded as the synced version. */ + version?: number + /** A streaming snapshot's production time (epoch ms): orders the merge only — never recorded as a checkpoint. */ + streamedAt?: number +} + 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, version)) + const run = tail.then(() => applyLiveFileDocMerge(fileId, markdown, order)) liveDocMergeChain.set(fileId, run) try { await run @@ -171,15 +185,21 @@ export function isLiveDocMergeInFlight(fileId: string): boolean { async function applyLiveFileDocMerge( fileId: string, markdown: string, - version?: number + 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 }, - // A durable `version` (the durable `updatedAt` epoch ms) records the version the live doc now - // incorporates (the persist If-Match guard); omitted for a streaming intermediate merge. - body: JSON.stringify({ fileId, markdown, version }), + // `version` (durable `contentUpdatedAt`) records the synced version the live doc now incorporates + // (the persist If-Match guard); `streamedAt` orders a streaming snapshot without recording it. + // JSON.stringify drops whichever is undefined, so the wire shape is unchanged for durable writes. + body: JSON.stringify({ + fileId, + markdown, + version: order.version, + streamedAt: order.streamedAt, + }), 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() From 6986b743557c1d5dbe358e4ff17aed78c5d539de Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 17:29:47 -0700 Subject: [PATCH 10/14] refactor(collab-doc): tidy merge-order docs + relay order object; cover multi-replica streaming stale-check --- .../handlers/file-doc.multireplica.test.ts | 91 +++++++++++++++++++ apps/realtime/src/handlers/file-doc.test.ts | 23 +++-- apps/realtime/src/handlers/file-doc.ts | 24 +++-- apps/realtime/src/routes/http.ts | 10 +- apps/sim/lib/realtime/notify.ts | 41 ++++----- 5 files changed, 146 insertions(+), 43 deletions(-) create mode 100644 apps/realtime/src/handlers/file-doc.multireplica.test.ts 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..878fd34536f --- /dev/null +++ b/apps/realtime/src/handlers/file-doc.multireplica.test.ts @@ -0,0 +1,91 @@ +/** + * @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('stale-checks against the SHARED synced version and never records a streaming merge', async () => { + // A durable write records the shared synced version cluster-wide. + expect(await applyMarkdownToLiveFileDoc('file-1', '# durable', { version: 100 })).toBe( + 'applied' + ) + expect(fakeStore.setSyncedVersion).toHaveBeenCalledWith(ROOM_NAME, 100) + mockFetchFileDocMerge.mockClear() + + // A delayed streaming snapshot older than the SHARED version (e.g. from another process) is stale — + // rejected under the lock before any diff is built, so the live doc never regresses. + expect(await applyMarkdownToLiveFileDoc('file-1', '# older stream', { streamedAt: 50 })).toBe( + 'stale' + ) + expect(mockFetchFileDocMerge).not.toHaveBeenCalled() + + // A streaming snapshot newer than the shared version applies (advances the live view)... + expect(await applyMarkdownToLiveFileDoc('file-1', '# newer stream', { streamedAt: 200 })).toBe( + 'applied' + ) + // ...but it is never recorded: a durable write between the two (150) still applies. Had 200 been + // recorded to the shared store, 150 would be rejected as stale. + 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 338c3fc4267..8c860d972a3 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -553,15 +553,19 @@ describe('setupWorkspaceFileDocHandlers', () => { 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', 100)).toBe('applied') + 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', 50)).toBe('stale') + 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', 100)).toBe('stale') + expect(await applyMarkdownToLiveFileDoc('file-1', '# same version', { version: 100 })).toBe( + 'stale' + ) expect(mockFetchFileDocMerge).not.toHaveBeenCalled() }) @@ -575,22 +579,25 @@ describe('setupWorkspaceFileDocHandlers', () => { mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc())) // A newer durable version lands and is recorded as the synced version. - expect(await applyMarkdownToLiveFileDoc('file-1', '# durable', 100)).toBe('applied') + expect(await applyMarkdownToLiveFileDoc('file-1', '# durable', { version: 100 })).toBe( + 'applied' + ) // A delayed streaming snapshot OLDER than that durable version — e.g. a throttled copilot snapshot // from another process arriving late — is stale, so it can't regress the doc back toward its content. - // (`version` omitted, `streamedAt` passed as the 4th arg.) - expect(await applyMarkdownToLiveFileDoc('file-1', '# older stream', undefined, 50)).toBe( + expect(await applyMarkdownToLiveFileDoc('file-1', '# older stream', { streamedAt: 50 })).toBe( 'stale' ) // A streaming snapshot NEWER than the durable version applies (it advances the live view)... - expect(await applyMarkdownToLiveFileDoc('file-1', '# newer stream', undefined, 200)).toBe( + expect(await applyMarkdownToLiveFileDoc('file-1', '# newer stream', { streamedAt: 200 })).toBe( 'applied' ) // ...but it is never RECORDED as the synced version: a durable write between the two (150) still // applies. Had the streaming 200 been recorded, 150 would have been rejected as stale. - expect(await applyMarkdownToLiveFileDoc('file-1', '# durable again', 150)).toBe('applied') + 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 () => { diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 1434832bbdb..4a1c0540c93 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -515,6 +515,15 @@ 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 `streamedAt` is checked only. + */ +interface MergeOrder { + version?: number + streamedAt?: 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,15 +549,12 @@ const fileDocMergeChains = new Map>() export function applyMarkdownToLiveFileDoc( fileId: string, markdown: string, - version?: number, - streamedAt?: number + 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, streamedAt)) + const run = prior.catch(() => {}).then(() => mergeMarkdownIntoRoom(name, fileId, markdown, order)) fileDocMergeChains.set( name, run.finally(() => { @@ -562,8 +568,7 @@ async function mergeMarkdownIntoRoom( name: string, fileId: string, markdown: string, - version?: number, - streamedAt?: number + { version, streamedAt }: MergeOrder ): Promise<'applied' | 'no-live-room' | 'merge-unavailable' | 'stale'> { const store = getFileDocStore() @@ -590,6 +595,11 @@ async function mergeMarkdownIntoRoom( // monotonic token stays high, so a later persist could write that stale content back over the durable // file. Skip it. This is what stops a delayed streaming snapshot from clobbering a newer durable merge // across processes (the per-process caller chain cannot). A merge with neither key is never stale. + // + // Only durable `version` (DB-monotonic `contentUpdatedAt`) is ever recorded, so the durable line is + // immune to clock skew. `streamedAt` is best-effort wall-clock used ONLY to order transient streaming + // snapshots; correctness never depends on it — a skewed clock at worst drops or briefly regresses the + // live preview, which the next durable write reconciles. const orderingVersion = version ?? streamedAt const isStale = (current: number): boolean => orderingVersion !== undefined && orderingVersion <= current diff --git a/apps/realtime/src/routes/http.ts b/apps/realtime/src/routes/http.ts index a31f86e4b79..b64e18e6248 100644 --- a/apps/realtime/src/routes/http.ts +++ b/apps/realtime/src/routes/http.ts @@ -216,12 +216,10 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) { // `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. // `streamedAt` orders a streaming snapshot on the same version line without recording it. - const result = await applyMarkdownToLiveFileDoc( - fileId, - markdown, - typeof version === 'number' ? version : undefined, - typeof streamedAt === 'number' ? streamedAt : undefined - ) + const result = await applyMarkdownToLiveFileDoc(fileId, markdown, { + version: typeof version === 'number' ? version : undefined, + streamedAt: typeof streamedAt === 'number' ? streamedAt : undefined, + }) res.writeHead(200, { 'Content-Type': 'application/json' }) res.end(JSON.stringify({ applied: result === 'applied' })) } catch (error) { diff --git a/apps/sim/lib/realtime/notify.ts b/apps/sim/lib/realtime/notify.ts index 6ac8513eace..936eb80aca5 100644 --- a/apps/sim/lib/realtime/notify.ts +++ b/apps/sim/lib/realtime/notify.ts @@ -109,6 +109,17 @@ 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): orders the merge AND is recorded as the synced version. */ + version?: number + /** A streaming snapshot's production time (epoch ms): orders the merge only — never recorded as a checkpoint. */ + streamedAt?: 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 @@ -125,33 +136,19 @@ export async function notifyFolderResourceChanged( * 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` positions this merge on the file's monotonic version line so a stale write never regresses - * the doc — the relay drops any merge not NEWER than the version the doc already incorporates: - * - `version` — a DURABLE write's `contentUpdatedAt` (epoch ms). Both orders the merge AND is recorded - * as the doc's synced version (the persist If-Match guard), so a later persist treats this write as - * synced rather than an out-of-band conflict. - * - `streamedAt` — the wall-clock time (epoch ms) a STREAMING snapshot was produced (the copilot stream - * mid-flight). Orders the merge so a delayed snapshot older than a newer durable write — possibly from - * another app process — is dropped, but is NEVER recorded: the synced version stays pinned to the last - * durable write, which is exactly the copilot tool's final `edit_content` write that reconciles the file. - * - * Pass one or the other, never both. Passing neither applies the merge without ordering it (legacy). + * `order` ({@link LiveFileDocMergeOrder}) positions this merge so a stale write never regresses the doc: + * the relay drops any merge not NEWER than the version the doc already incorporates. A durable `version` + * is recorded as the synced version (the persist If-Match guard); a streaming `streamedAt` is checked but + * never recorded, 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 the monotonic version above (durable version / - * streaming `streamedAt`) 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. + * 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 interface LiveFileDocMergeOrder { - /** A durable write's `contentUpdatedAt` (epoch ms): orders the merge AND is recorded as the synced version. */ - version?: number - /** A streaming snapshot's production time (epoch ms): orders the merge only — never recorded as a checkpoint. */ - streamedAt?: number -} - export async function mergeEditIntoLiveFileDoc( fileId: string, markdown: string, From 8a7bca439205e1cdc9330405b9061a8867092246 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 17:54:54 -0700 Subject: [PATCH 11/14] fix(collab-doc): order streaming merges by causal base version, not wall-clock A streaming snapshot now carries baseVersion (the durable contentUpdatedAt it was built from) instead of a wall-clock streamedAt. The relay drops the snapshot when a newer durable write landed since that base, so a concurrent human save can no longer be clobbered in the live doc and then persisted over the durable file. Skew-immune: both keys are DB-monotonic contentUpdatedAt values. --- .../handlers/file-doc.multireplica.test.ts | 25 +++++---- apps/realtime/src/handlers/file-doc.test.ts | 28 +++++----- apps/realtime/src/handlers/file-doc.ts | 42 ++++++++------- apps/realtime/src/routes/http.ts | 6 +-- .../request/go/file-preview-adapter.test.ts | 23 +++++--- .../request/go/file-preview-adapter.ts | 53 +++++++++++-------- .../session/file-preview-session-contract.ts | 4 ++ .../request/session/file-preview-session.ts | 2 + .../tools/server/files/file-preview.ts | 15 +++++- apps/sim/lib/realtime/notify.test.ts | 11 ++-- apps/sim/lib/realtime/notify.ts | 27 ++++++---- 11 files changed, 141 insertions(+), 95 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.multireplica.test.ts b/apps/realtime/src/handlers/file-doc.multireplica.test.ts index 878fd34536f..17c17ffe371 100644 --- a/apps/realtime/src/handlers/file-doc.multireplica.test.ts +++ b/apps/realtime/src/handlers/file-doc.multireplica.test.ts @@ -60,27 +60,26 @@ describe('applyMarkdownToLiveFileDoc — multi-replica (store-enabled) ordering' mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc())) }) - it('stale-checks against the SHARED synced version and never records a streaming merge', async () => { - // A durable write records the shared synced version cluster-wide. + 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 delayed streaming snapshot older than the SHARED version (e.g. from another process) is stale — - // rejected under the lock before any diff is built, so the live doc never regresses. - expect(await applyMarkdownToLiveFileDoc('file-1', '# older stream', { streamedAt: 50 })).toBe( - 'stale' - ) + // 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 newer than the shared version applies (advances the live view)... - expect(await applyMarkdownToLiveFileDoc('file-1', '# newer stream', { streamedAt: 200 })).toBe( - 'applied' - ) - // ...but it is never recorded: a durable write between the two (150) still applies. Had 200 been - // recorded to the shared store, 150 would be rejected as stale. + // 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' ) diff --git a/apps/realtime/src/handlers/file-doc.test.ts b/apps/realtime/src/handlers/file-doc.test.ts index 8c860d972a3..7ccbe3d2254 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -569,7 +569,7 @@ describe('setupWorkspaceFileDocHandlers', () => { expect(mockFetchFileDocMerge).not.toHaveBeenCalled() }) - it('orders a streaming merge by streamedAt but never records it (stays behind the durable version)', async () => { + 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) @@ -578,23 +578,25 @@ describe('setupWorkspaceFileDocHandlers', () => { mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc())) - // A newer durable version lands and is recorded as the synced version. + // 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 delayed streaming snapshot OLDER than that durable version — e.g. a throttled copilot snapshot - // from another process arriving late — is stale, so it can't regress the doc back toward its content. - expect(await applyMarkdownToLiveFileDoc('file-1', '# older stream', { streamedAt: 50 })).toBe( - 'stale' - ) + // 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 NEWER than the durable version applies (it advances the live view)... - expect(await applyMarkdownToLiveFileDoc('file-1', '# newer stream', { streamedAt: 200 })).toBe( - 'applied' - ) - // ...but it is never RECORDED as the synced version: a durable write between the two (150) still - // applies. Had the streaming 200 been recorded, 150 would have been rejected as 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' ) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 4a1c0540c93..52fb0a69a84 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -517,11 +517,12 @@ 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 `streamedAt` is checked only. + * 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 - streamedAt?: number + baseVersion?: number } /** @@ -568,7 +569,7 @@ async function mergeMarkdownIntoRoom( name: string, fileId: string, markdown: string, - { version, streamedAt }: MergeOrder + { version, baseVersion }: MergeOrder ): Promise<'applied' | 'no-live-room' | 'merge-unavailable' | 'stale'> { const store = getFileDocStore() @@ -576,7 +577,7 @@ async function mergeMarkdownIntoRoom( // 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 `streamedAt` orders the merge (below) but is never a checkpoint, + // `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 @@ -588,21 +589,24 @@ async function mergeMarkdownIntoRoom( await store.setSyncedVersion(name, version) } - // Position this merge on the file's monotonic version line: a durable write by its `version`, a - // streaming snapshot by its `streamedAt` production time. A merge not NEWER than the version the doc - // already incorporates is stale — a newer durable write already landed (possibly on another process, - // out of dispatch order). Diffing toward its older markdown would regress the live doc while the - // monotonic token stays high, so a later persist could write that stale content back over the durable - // file. Skip it. This is what stops a delayed streaming snapshot from clobbering a newer durable merge - // across processes (the per-process caller chain cannot). A merge with neither key is never stale. - // - // Only durable `version` (DB-monotonic `contentUpdatedAt`) is ever recorded, so the durable line is - // immune to clock skew. `streamedAt` is best-effort wall-clock used ONLY to order transient streaming - // snapshots; correctness never depends on it — a skewed clock at worst drops or briefly regresses the - // live preview, which the next durable write reconciles. - const orderingVersion = version ?? streamedAt - const isStale = (current: number): boolean => - orderingVersion !== undefined && orderingVersion <= current + // 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. + const isStale = (current: number): boolean => { + if (version !== undefined) return version <= current + if (baseVersion !== undefined) return current > baseVersion + return false + } if (store.enabled) { // Serialize merges to this file ACROSS tasks — the per-file chain above only covers this process. diff --git a/apps/realtime/src/routes/http.ts b/apps/realtime/src/routes/http.ts index b64e18e6248..b37e9061107 100644 --- a/apps/realtime/src/routes/http.ts +++ b/apps/realtime/src/routes/http.ts @@ -209,16 +209,16 @@ 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, streamedAt } = 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. - // `streamedAt` orders a streaming snapshot on the same version line without recording it. + // `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, - streamedAt: typeof streamedAt === 'number' ? streamedAt : 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/lib/copilot/request/go/file-preview-adapter.test.ts b/apps/sim/lib/copilot/request/go/file-preview-adapter.test.ts index 7519f61fe07..4fffebab85e 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 @@ -15,7 +15,7 @@ const { mergeEditIntoLiveFileDocMock, isLiveDocMergeInFlightMock } = vi.hoisted( ( fileId: string, markdown: string, - order?: { version?: number; streamedAt?: number } + order?: { version?: number; baseVersion?: number } ) => Promise >(), isLiveDocMergeInFlightMock: vi.fn<(fileId: string) => boolean>(), @@ -46,6 +46,8 @@ 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. */ function editContentDelta(argumentsDelta: string): StreamEvent { @@ -104,8 +106,12 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { vi.clearAllMocks() mergeEditIntoLiveFileDocMock.mockResolvedValue(undefined) isLiveDocMergeInFlightMock.mockReturnValue(false) - // Default: an append/patch base is available (a non-empty file), so the base-present gate passes. - peekFileIntentMock.mockResolvedValue({ existingContent: 'Base.' }) + // 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) @@ -129,7 +135,7 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { }) } - it('merges the growing full content (no version arg) into the live doc as it streams', async () => { + 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) @@ -143,16 +149,17 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { 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 `streamedAt` (its wall-clock time) to order it — never `version`, which rides the final - // edit_content write; so the relay orders the snapshot without recording it as a durable checkpoint. + // 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(typeof first[2]?.streamedAt).toBe('number') + 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(typeof second[2]?.streamedAt).toBe('number') + expect(second[2]?.baseVersion).toBe(BASE_VERSION_MS) }) it('throttles merges: two deltas within LIVE_DOC_MERGE_THROTTLE_MS yield one merge', async () => { 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 f0daac5f828..432d0db58ca 100644 --- a/apps/sim/lib/copilot/request/go/file-preview-adapter.ts +++ b/apps/sim/lib/copilot/request/go/file-preview-adapter.ts @@ -23,6 +23,7 @@ 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' @@ -272,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', @@ -396,21 +398,24 @@ 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, @@ -469,20 +474,23 @@ 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, @@ -611,9 +619,11 @@ export async function processFilePreviewStreamEvent(input: { } ) if (typeof intentBase?.existingContent === 'string') { + const baseVersion = intentBase.fileRecord?.contentUpdatedAt?.getTime() const seededSession: FilePreviewSession = { ...currentPreview.session, baseContent: intentBase.existingContent, + ...(baseVersion !== undefined ? { baseVersion } : {}), ...(intentBase.edit ? { edit: intentBase.edit } : {}), } currentPreview = { @@ -654,12 +664,13 @@ export async function processFilePreviewStreamEvent(input: { // 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 `streamedAt` (this snapshot's wall-clock time) so - // the relay orders it on the file's version line — a delayed snapshot older than a newer - // durable write, even from another process, is dropped rather than regressing the doc — but - // never records it as a durable 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`). + // 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 @@ -678,7 +689,7 @@ export async function processFilePreviewStreamEvent(input: { const nextLiveMergeAt = dueForLiveMerge ? now : currentPreview.lastLiveMergeAt if (dueForLiveMerge && nextSession.fileId) { void mergeEditIntoLiveFileDoc(nextSession.fileId, nextSession.previewText, { - streamedAt: now, + baseVersion: nextSession.baseVersion, }) } 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..11ba48d8cdb 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,26 @@ 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 (`contentUpdatedAt`, + * 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. + * `baseVersion` is undefined only for a legacy file with no recorded `contentUpdatedAt`. + */ +export interface WorkspaceFilePreviewBase { + text: string + baseVersion: number | undefined +} + 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?.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 67c4750b5a0..20689dd745d 100644 --- a/apps/sim/lib/realtime/notify.test.ts +++ b/apps/sim/lib/realtime/notify.test.ts @@ -24,21 +24,22 @@ describe('mergeEditIntoLiveFileDoc', () => { expect.objectContaining({ method: 'POST', headers: expect.objectContaining({ 'x-api-key': 'secret' }), - // A durable write sends `version`; the undefined `streamedAt` is dropped by JSON.stringify. + // 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 streamedAt (not version) for a streaming snapshot', async () => { + 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', { streamedAt: 1234 }) + await mergeEditIntoLiveFileDoc('file-1', '# hello', { baseVersion: 1234 }) - // The relay orders the snapshot by streamedAt without recording it — version stays absent on the wire. + // 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', streamedAt: 1234 }) + JSON.stringify({ fileId: 'file-1', markdown: '# hello', baseVersion: 1234 }) ) }) diff --git a/apps/sim/lib/realtime/notify.ts b/apps/sim/lib/realtime/notify.ts index 936eb80aca5..a67f86401c4 100644 --- a/apps/sim/lib/realtime/notify.ts +++ b/apps/sim/lib/realtime/notify.ts @@ -114,10 +114,13 @@ export async function notifyFolderResourceChanged( * never both; passing neither applies the merge without ordering it (legacy). */ export interface LiveFileDocMergeOrder { - /** A durable write's `contentUpdatedAt` (epoch ms): orders the merge AND is recorded as the synced version. */ + /** 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 production time (epoch ms): orders the merge only — never recorded as a checkpoint. */ - streamedAt?: 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 } /** @@ -136,11 +139,13 @@ export interface LiveFileDocMergeOrder { * 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: - * the relay drops any merge not NEWER than the version the doc already incorporates. A durable `version` - * is recorded as the synced version (the persist If-Match guard); a streaming `streamedAt` is checked but - * never recorded, 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. + * `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 @@ -189,13 +194,13 @@ 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); `streamedAt` orders a streaming snapshot without recording it. - // JSON.stringify drops whichever is undefined, so the wire shape is unchanged for durable writes. + // (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, - streamedAt: order.streamedAt, + baseVersion: order.baseVersion, }), signal: AbortSignal.timeout(APPLY_EDIT_TIMEOUT_MS), }) From d95070f3e2b43956a3ed7cc72f463b6e6d155b49 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 18:06:03 -0700 Subject: [PATCH 12/14] fix(collab-doc): derive streaming baseVersion as contentUpdatedAt ?? updatedAt Match the version line the seed/persist use so a legacy file with no content version still ships an ordered streaming snapshot instead of an unordered one. --- .../request/go/file-preview-adapter.test.ts | 17 +++++++++++++++++ .../copilot/request/go/file-preview-adapter.ts | 6 +++++- .../copilot/tools/server/files/file-preview.ts | 16 ++++++++++------ 3 files changed, 32 insertions(+), 7 deletions(-) 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 4fffebab85e..d3a876588d1 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 @@ -162,6 +162,23 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { 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', 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 432d0db58ca..c80679df162 100644 --- a/apps/sim/lib/copilot/request/go/file-preview-adapter.ts +++ b/apps/sim/lib/copilot/request/go/file-preview-adapter.ts @@ -619,7 +619,11 @@ export async function processFilePreviewStreamEvent(input: { } ) if (typeof intentBase?.existingContent === 'string') { - const baseVersion = intentBase.fileRecord?.contentUpdatedAt?.getTime() + // 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, 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 11ba48d8cdb..698d524d463 100644 --- a/apps/sim/lib/copilot/tools/server/files/file-preview.ts +++ b/apps/sim/lib/copilot/tools/server/files/file-preview.ts @@ -142,14 +142,15 @@ function buildAppendPreview(existingContent: string, incomingContent: string): s * full-file replacement until the intent landed. */ /** - * The base content a copilot edit is computed against, plus the durable version (`contentUpdatedAt`, - * 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. - * `baseVersion` is undefined only for a legacy file with no recorded `contentUpdatedAt`. + * 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 | undefined + baseVersion: number } export async function loadWorkspaceFileTextForPreview( @@ -160,7 +161,10 @@ export async function loadWorkspaceFileTextForPreview( const record = await getWorkspaceFile(workspaceId, fileId) if (!record) return undefined const buffer = await fetchWorkspaceFileBuffer(record) - return { text: buffer.toString('utf-8'), baseVersion: record.contentUpdatedAt?.getTime() } + 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, From 3a5ffbd7ce98436530c832a13dc8261017f155dd Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 18:19:25 -0700 Subject: [PATCH 13/14] fix(collab-doc): fail-closed on a streaming snapshot with no baseVersion The live-merge gate now requires a numeric baseVersion, not just loaded base content. A rare base with no file record (hence no version) would otherwise ship an unordered snapshot the relay can't stale-check, risking a clobber of a concurrent durable write. Skip the live merge instead; the durable write reconciles. --- .../copilot/request/go/file-preview-adapter.test.ts | 12 ++++++++++++ .../lib/copilot/request/go/file-preview-adapter.ts | 4 ++++ 2 files changed, 16 insertions(+) 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 d3a876588d1..8915f2cd472 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 @@ -206,6 +206,18 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => { 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) 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 c80679df162..ccc911bd08e 100644 --- a/apps/sim/lib/copilot/request/go/file-preview-adapter.ts +++ b/apps/sim/lib/copilot/request/go/file-preview-adapter.ts @@ -683,11 +683,15 @@ export async function processFilePreviewStreamEvent(input: { // 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 From 16fa747e32f022e94403fc48a9f36a6a0e2b81f7 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 18:29:07 -0700 Subject: [PATCH 14/14] docs(collab-doc): document the accepted concurrent-independent-streams limitation --- apps/realtime/src/handlers/file-doc.ts | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 52fb0a69a84..039e2ab6661 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -602,6 +602,14 @@ async function mergeMarkdownIntoRoom( // 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