Merge pull request #6021 from simstudioai/collab-doc-stage-c

feat(collab-doc): stream copilot edits into the live document (Stage C)

Author: waleedlatif1

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

@@ -0,0 +1,74 @@
+import { FILE_DOC_TIMEOUTS } from '@sim/realtime-protocol/file-doc'
+import { env, getBaseUrl } from '@/env'
+
+/**
+ * The relay's client for the app's internal file-doc endpoints. The app owns the markdown↔Yjs
+ * conversion engine (TipTap + jsdom) and blob/DB access; the relay owns the live document. So for any
+ * operation that needs conversion, the relay delegates here over the shared `x-api-key` channel. The
+ * timeouts (and their ordering vs. the app-side bounds) live in the shared `FILE_DOC_TIMEOUTS`.
+ */
+
+function postToApp(path: string, payload: unknown, timeoutMs: number): Promise<Response> {
+  return fetch(`${getBaseUrl()}${path}`, {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json', 'x-api-key': env.INTERNAL_API_SECRET },
+    body: JSON.stringify(payload),
+    signal: AbortSignal.timeout(timeoutMs),
+  })
+}
+
+/**
+ * Ask the app to build a server-authoritative seed (markdown → Yjs) for a file's collaborative
+ * document. Returns the Yjs update to apply, or `null` for a genuinely empty/missing file (an empty
+ * document is correct). THROWS on a transport failure (non-2xx / network / timeout / malformed body)
+ * so the caller can tell a real empty from a failure it should be allowed to retry.
+ */
+export async function fetchFileDocSeed(
+  workspaceId: string,
+  fileId: string
+): Promise<Uint8Array | null> {
+  const response = await postToApp(
+    '/api/internal/file-doc/seed',
+    { workspaceId, fileId },
+    FILE_DOC_TIMEOUTS.seedRequestMs
+  )
+  if (!response.ok) {
+    throw new Error(`Seed fetch failed for file ${fileId}: ${response.status}`)
+  }
+  const body = (await response.json()) as { update?: unknown }
+  const update = body?.update
+  // A well-formed response is `{ update: base64-string | null }`. Anything else is a contract
+  // violation, not a "genuinely empty file" — throw so the caller retries rather than silently
+  // treating a malformed body as empty and stranding the room unseeded.
+  if (update === null) return null
+  if (typeof update !== 'string') {
+    throw new Error(`Seed fetch for file ${fileId} returned a malformed body`)
+  }
+  return new Uint8Array(Buffer.from(update, 'base64'))
+}
+
+/**
+ * Ask the app to merge new markdown into the live document as a minimal Yjs diff — Stage C, so a
+ * copilot edit streams into open editors instead of the file changing underneath them. The relay
+ * ships the document's current state and applies the returned diff (which Yjs reconciles with any
+ * concurrent user edits). THROWS on a transport failure or malformed body.
+ */
+export async function fetchFileDocMerge(
+  fileId: string,
+  docState: Uint8Array,
+  markdown: string
+): Promise<Uint8Array> {
+  const response = await postToApp(
+    '/api/internal/file-doc/merge',
+    { fileId, docState: Buffer.from(docState).toString('base64'), markdown },
+    FILE_DOC_TIMEOUTS.mergeRequestMs
+  )
+  if (!response.ok) {
+    throw new Error(`Merge fetch failed for file ${fileId}: ${response.status}`)
+  }
+  const body = (await response.json()) as { update?: unknown }
+  if (typeof body?.update !== 'string') {
+    throw new Error(`Merge fetch for file ${fileId} returned a malformed body`)
+  }
+  return new Uint8Array(Buffer.from(body.update, 'base64'))
+}

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

@@ -14,20 +14,26 @@ import * as syncProtocol from 'y-protocols/sync'
 import * as Y from 'yjs'
 import type { IRoomManager } from '@/rooms'
 
-const { mockAuthorizeRoom, mockFetchFileDocSeed } = vi.hoisted(() => ({
+const { mockAuthorizeRoom, mockFetchFileDocSeed, mockFetchFileDocMerge } = vi.hoisted(() => ({
   mockAuthorizeRoom: vi.fn(),
   mockFetchFileDocSeed: vi.fn(),
+  mockFetchFileDocMerge: vi.fn(),
 }))
 
 vi.mock('@sim/platform-authz/rooms', () => ({
   authorizeRoom: mockAuthorizeRoom,
 }))
 
-vi.mock('@/handlers/file-doc-seed', () => ({
+vi.mock('@/handlers/file-doc-app', () => ({
   fetchFileDocSeed: mockFetchFileDocSeed,
+  fetchFileDocMerge: mockFetchFileDocMerge,
 }))
 
-import { cleanupFileDocForSocket, setupWorkspaceFileDocHandlers } from '@/handlers/file-doc'
+import {
+  applyMarkdownToLiveFileDoc,
+  cleanupFileDocForSocket,
+  setupWorkspaceFileDocHandlers,
+} from '@/handlers/file-doc'
 
 type Handler = (payload?: unknown) => Promise<void> | void
 
@@ -173,6 +179,9 @@ describe('setupWorkspaceFileDocHandlers', () => {
     // Default: the server seed builder returns no content (empty file). Tests that
     // exercise seeding override this per-case with an encoded Yjs update.
     mockFetchFileDocSeed.mockResolvedValue(null)
+    // Default: the merge builder returns a valid no-op (empty-doc) update. Tests exercising copilot
+    // merges override it.
+    mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc()))
   })
 
   afterEach(() => {
@@ -399,6 +408,65 @@ describe('setupWorkspaceFileDocHandlers', () => {
     expect(clientDoc.getText(FILE_DOC_FIELD).toString()).toContain('# Seeded')
   })
 
+  it('merges a copilot edit into a seeded live room and relays it to editors', async () => {
+    mockFetchFileDocSeed.mockResolvedValue(encodedSeedUpdate('# Original'))
+    const { io, sent } = createIo()
+    const { handlers } = setup('socket-1', io)
+    await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
+    await flushMicrotasks() // seed lands → room is seeded/live
+
+    // The app returns a diff (here, an update introducing text) for the relay to apply.
+    const diff = new Y.Doc()
+    diff.getText(FILE_DOC_FIELD).insert(0, 'copilot content')
+    mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(diff))
+    sent.length = 0
+
+    const result = await applyMarkdownToLiveFileDoc('file-1', '# Rewritten by copilot')
+
+    expect(result).toBe('applied')
+    expect(mockFetchFileDocMerge).toHaveBeenCalledWith(
+      'file-1',
+      expect.any(Uint8Array),
+      '# Rewritten by copilot'
+    )
+    // Applying the diff fires doc.on('update') → the merge is broadcast to the whole room.
+    expect(sent.some((m) => m.event === FILE_DOC_EVENTS.MESSAGE && m.target === ROOM_NAME)).toBe(
+      true
+    )
+  })
+
+  it('reports no-live-room (and does not call the app) when the file has no seeded room', async () => {
+    const result = await applyMarkdownToLiveFileDoc('file-1', '# anything')
+    expect(result).toBe('no-live-room')
+    expect(mockFetchFileDocMerge).not.toHaveBeenCalled()
+  })
+
+  it('serializes concurrent merges for the same file (second waits for the first)', async () => {
+    mockFetchFileDocSeed.mockResolvedValue(encodedSeedUpdate('# Original'))
+    const { io } = createIo()
+    const { handlers } = setup('socket-1', io)
+    await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
+    await flushMicrotasks()
+
+    // First merge is left in flight; the second must not start its own fetch until the first finishes.
+    const noOpUpdate = Y.encodeStateAsUpdate(new Y.Doc())
+    let resolveFirst: (v: Uint8Array) => void = () => {}
+    mockFetchFileDocMerge
+      .mockReturnValueOnce(new Promise((resolve) => (resolveFirst = resolve)))
+      .mockResolvedValueOnce(noOpUpdate)
+
+    const first = applyMarkdownToLiveFileDoc('file-1', '# One')
+    const second = applyMarkdownToLiveFileDoc('file-1', '# Two')
+    await flushMicrotasks()
+    expect(mockFetchFileDocMerge).toHaveBeenCalledTimes(1) // second is queued behind the first
+
+    resolveFirst(noOpUpdate)
+    await first
+    await second
+    // Only after the first resolved did the second run — and it snapshotted the post-first state.
+    expect(mockFetchFileDocMerge).toHaveBeenCalledTimes(2)
+  })
+
   it('relays a document update to the rest of the room, excluding the sender', async () => {
     const { io, sent } = createIo()
     const a = setup('socket-a', io)

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

@@ -36,7 +36,7 @@ import * as awarenessProtocol from 'y-protocols/awareness'
 import * as syncProtocol from 'y-protocols/sync'
 import * as Y from 'yjs'
 import { resolveAvatarUrl } from '@/handlers/avatar'
-import { fetchFileDocSeed } from '@/handlers/file-doc-seed'
+import { fetchFileDocMerge, fetchFileDocSeed } from '@/handlers/file-doc-app'
 import { resolveRoomJoinAuth } from '@/handlers/room-join-auth'
 import type { AuthenticatedSocket } from '@/middleware/auth'
 import type { IRoomManager } from '@/rooms'
@@ -205,6 +205,58 @@ async function ensureServerSeed(
   }
 }
 
+/** Serializes live merges per file so overlapping calls never race the same doc (see below). */
+const fileDocMergeChains = new Map<string, Promise<unknown>>()
+
+/**
+ * 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
+ * fires `doc.on('update')` and relays the merge to every connected editor, reconciled with any
+ * concurrent user edits — and reports whether it landed.
+ *
+ * Merges for the same file are SERIALIZED: each waits for any in-flight merge on that doc, so it
+ * snapshots the already-updated state and its diff is computed against current content — two
+ * overlapping full-document rewrites can never each diff the same stale snapshot and apply out of
+ * order. (The relay is single-writer per file — Helm pins one replica — so an in-process chain
+ * suffices.)
+ *
+ * Returns `'no-live-room'` when the file has no seeded, occupied room: an unseeded/empty doc has no
+ * authoritative content to merge against, so the caller (copilot) writes the file directly instead and
+ * the seed picks up the new content on the next open.
+ */
+export function applyMarkdownToLiveFileDoc(
+  fileId: string,
+  markdown: string
+): Promise<'applied' | 'no-live-room'> {
+  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))
+  fileDocMergeChains.set(
+    name,
+    run.finally(() => {
+      if (fileDocMergeChains.get(name) === run) fileDocMergeChains.delete(name)
+    })
+  )
+  return run
+}
+
+async function mergeMarkdownIntoRoom(
+  name: string,
+  fileId: string,
+  markdown: string
+): Promise<'applied' | 'no-live-room'> {
+  const room = fileDocRooms.get(name)
+  if (!room || room.owners.size === 0 || !isDocSeeded(room.doc)) return 'no-live-room'
+
+  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)
+  return 'applied'
+}
+
 /**
  * Get (or lazily create) the authoritative document for a room, wiring the two
  * relay handlers exactly once: document updates and awareness changes are

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

@@ -2,6 +2,7 @@ import type { IncomingMessage, ServerResponse } from 'http'
 import { ROOM_TYPES } from '@sim/realtime-protocol/rooms'
 import { safeCompare } from '@sim/security/compare'
 import { env } from '@/env'
+import { applyMarkdownToLiveFileDoc } from '@/handlers/file-doc'
 import { type IRoomManager, WorkflowRoomService } from '@/rooms'
 
 interface Logger {
@@ -183,6 +184,26 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) {
       return
     }
 
+    // Merge a copilot edit into a file's LIVE collaborative document so it streams into open editors
+    // (Stage C). Returns `{ applied }`: when false, no seeded live room exists and the caller writes
+    // the file directly instead. Any live user edits are preserved — the app builds a minimal CRDT diff.
+    if (req.method === 'POST' && req.url === '/api/file-doc/apply-edit') {
+      try {
+        const body = await readRequestBody(req)
+        const { fileId, markdown } = JSON.parse(body)
+        if (!isNonEmptyString(fileId) || typeof markdown !== 'string') {
+          return sendError(res, 'Invalid fileId or markdown', 400)
+        }
+        const result = await applyMarkdownToLiveFileDoc(fileId, markdown)
+        res.writeHead(200, { 'Content-Type': 'application/json' })
+        res.end(JSON.stringify({ applied: result === 'applied' }))
+      } catch (error) {
+        logger.error('Error applying copilot edit to live file-doc:', error)
+        sendError(res, 'Failed to apply edit to live document')
+      }
+      return
+    }
+
     res.writeHead(404, { 'Content-Type': 'application/json' })
     res.end(JSON.stringify({ error: 'Not found' }))
   }

apps/realtime/src/routes/http.ts

@@ -0,0 +1,37 @@
+import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
+import type { NextRequest } from 'next/server'
+import { NextResponse } from 'next/server'
+import { mergeFileDocContract } from '@/lib/api/contracts/file-doc'
+import { parseRequest } from '@/lib/api/server'
+import { buildFileDocMergeUpdate } from '@/lib/collab-doc/merge'
+import { checkInternalApiKey, createUnauthorizedResponse } from '@/lib/copilot/request/http'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+
+const logger = createLogger('FileDocMergeAPI')
+
+/**
+ * POST /api/internal/file-doc/merge — merge new markdown into a live collaborative document as a
+ * minimal Yjs diff (Stage C — copilot writing into an open doc). The realtime relay ships the current
+ * doc state; the app returns the diff to apply + relay. Internal only: gated on the shared
+ * `x-api-key: INTERNAL_API_SECRET` secret, matching the seed endpoint and the realtime relay.
+ */
+export const POST = withRouteHandler(async (request: NextRequest) => {
+  const auth = checkInternalApiKey(request)
+  if (!auth.success) return createUnauthorizedResponse()
+
+  const parsed = await parseRequest(mergeFileDocContract, request, {})
+  if (!parsed.success) return parsed.response
+  const { fileId, docState, markdown } = parsed.data.body
+
+  try {
+    const update = buildFileDocMergeUpdate(Buffer.from(docState, 'base64'), markdown)
+    return NextResponse.json({ update: Buffer.from(update).toString('base64') })
+  } catch (error) {
+    logger.error('Failed to merge markdown into file-doc', { fileId, error })
+    return NextResponse.json(
+      { error: getErrorMessage(error, 'Failed to merge document') },
+      { status: 500 }
+    )
+  }
+})

apps/sim/app/api/internal/file-doc/merge/route.ts

@@ -2,6 +2,7 @@ import {
   FILE_DOC_EVENTS,
   FILE_DOC_MESSAGE_TYPE,
   FILE_DOC_SEED,
+  FILE_DOC_TIMEOUTS,
   type JoinFileDocError,
   type JoinFileDocSuccess,
   toFileDocBytes,
@@ -34,9 +35,10 @@ interface FileDocProviderEvents {
  * On the deadline the provider latches fatal and surfaces a non-retryable `join-error` — the exact
  * path a fatal rejection uses — so the editor falls back to showing the file's stored content
  * read-only instead of a permanently blank pane. Generous enough to clear a slow connect + seed
- * round-trip; a healthy cold open reaches readiness well within it.
+ * round-trip; a healthy cold open reaches readiness well within it. Shared with (and must exceed) the
+ * relay's seed-fetch timeout — see `FILE_DOC_TIMEOUTS` and its ordering test.
  */
-const READINESS_DEADLINE_MS = 12_000
+const READINESS_DEADLINE_MS = FILE_DOC_TIMEOUTS.readinessDeadlineMs
 
 /**
  * The client half of the collaborative file-document protocol: a Yjs provider