Skip to content

Commit ae8c7a5

Browse files
committed
feat(files): stream copilot edits into the collaborative doc smoothly
- apply the agent stream client-side into the live Yjs binding as minimal updateYFragment diffs (like main's setContent, but incremental) so it renders smoothly AND broadcasts to every peer via CRDT — a collaborator on /files sees the stream for free - gate the apply on collabReady so diffs never land on an unseeded doc; keep the read-only placeholder visible until the seed swaps in - run streamed ops under a dedicated tx origin so they stay out of the user's undo stack - delete the throttled server-side streaming merge and the baseVersion ordering machinery it needed (relay + notify + session contract); the durable final write still reconciles open editors and seeds late joiners
1 parent ae8f662 commit ae8c7a5

14 files changed

Lines changed: 353 additions & 429 deletions

File tree

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

Lines changed: 8 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -60,31 +60,27 @@ describe('applyMarkdownToLiveFileDoc — multi-replica (store-enabled) ordering'
6060
mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc()))
6161
})
6262

63-
it('drops a stale-base streaming snapshot against the SHARED synced version and never records it', async () => {
63+
it('drops a stale durable write against the SHARED synced version', async () => {
6464
// A durable write (e.g. a concurrent human save on another process) records the shared synced version.
6565
expect(await applyMarkdownToLiveFileDoc('file-1', '# durable', { version: 100 })).toBe(
6666
'applied'
6767
)
6868
expect(fakeStore.setSyncedVersion).toHaveBeenCalledWith(ROOM_NAME, 100)
6969
mockFetchFileDocMerge.mockClear()
7070

71-
// A streaming snapshot built from an older base (50) than the SHARED synced version is stale —
72-
// rejected under the lock before any diff is built, so it can't clobber the durable write.
73-
expect(
74-
await applyMarkdownToLiveFileDoc('file-1', '# stale-base stream', { baseVersion: 50 })
75-
).toBe('stale')
71+
// A durable write with an OLDER version than the SHARED synced version is stale — rejected under the
72+
// lock before any diff is built, so it can't regress the doc across replicas.
73+
expect(await applyMarkdownToLiveFileDoc('file-1', '# older durable', { version: 50 })).toBe(
74+
'stale'
75+
)
7676
expect(mockFetchFileDocMerge).not.toHaveBeenCalled()
7777

78-
// A streaming snapshot whose base is the current shared version applies (nothing newer to clobber)...
79-
expect(
80-
await applyMarkdownToLiveFileDoc('file-1', '# current-base stream', { baseVersion: 100 })
81-
).toBe('applied')
82-
// ...but is never recorded: a later durable write at 150 still applies.
78+
// A newer durable write applies and advances the shared synced version.
8379
expect(await applyMarkdownToLiveFileDoc('file-1', '# durable again', { version: 150 })).toBe(
8480
'applied'
8581
)
8682
expect(fakeStore.setSyncedVersion).toHaveBeenCalledWith(ROOM_NAME, 150)
87-
// setSyncedVersion fired only for the two durable writes, never for a streaming snapshot.
83+
// setSyncedVersion fired only for the two applied durable writes, never for the stale one.
8884
expect(fakeStore.setSyncedVersion).toHaveBeenCalledTimes(2)
8985
})
9086
})

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

Lines changed: 0 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -569,39 +569,6 @@ describe('setupWorkspaceFileDocHandlers', () => {
569569
expect(mockFetchFileDocMerge).not.toHaveBeenCalled()
570570
})
571571

572-
it('drops a streaming snapshot whose base predates a newer durable write, but never records it', async () => {
573-
mockFetchFileDocSeed.mockResolvedValue(seedResult('# Original')) // seed version 1
574-
const { io } = createIo()
575-
const { handlers } = setup('socket-1', io)
576-
await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
577-
await flushMicrotasks()
578-
579-
mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc()))
580-
581-
// A durable write (e.g. a concurrent human save) lands and is recorded as the synced version.
582-
expect(await applyMarkdownToLiveFileDoc('file-1', '# durable', { version: 100 })).toBe(
583-
'applied'
584-
)
585-
586-
// A streaming snapshot built from an OLDER base (50) — copilot loaded the file before that durable
587-
// write — is stale: applying it would diff the live doc back toward the copilot content and clobber
588-
// the durable write, which a later persist would then write over the file.
589-
expect(
590-
await applyMarkdownToLiveFileDoc('file-1', '# stale-base stream', { baseVersion: 50 })
591-
).toBe('stale')
592-
593-
// A streaming snapshot whose base IS the current durable version applies — nothing newer to clobber.
594-
expect(
595-
await applyMarkdownToLiveFileDoc('file-1', '# current-base stream', { baseVersion: 100 })
596-
).toBe('applied')
597-
598-
// ...and a streaming merge is never recorded as the synced version: a later durable write at 150 still
599-
// applies (only durable writes move the synced version; the final edit_content write reconciles).
600-
expect(await applyMarkdownToLiveFileDoc('file-1', '# durable again', { version: 150 })).toBe(
601-
'applied'
602-
)
603-
})
604-
605572
it('serializes concurrent merges for the same file (second waits for the first)', async () => {
606573
mockFetchFileDocSeed.mockResolvedValue(seedResult('# Original'))
607574
const { io } = createIo()

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

Lines changed: 9 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -517,12 +517,11 @@ const fileDocMergeChains = new Map<string, Promise<unknown>>()
517517

518518
/**
519519
* How a merge is positioned on the file's version line — mirrors the sim-side `LiveFileDocMergeOrder`
520-
* wire fields. A durable `version` is checked AND recorded; a streaming `baseVersion` (the durable version
521-
* the snapshot was built from) is checked only — dropped if a newer durable write has since landed.
520+
* wire field. A durable `version` is checked (applied only if newer than the doc's current version) AND
521+
* recorded as the synced version.
522522
*/
523523
interface MergeOrder {
524524
version?: number
525-
baseVersion?: number
526525
}
527526

528527
/**
@@ -569,16 +568,14 @@ async function mergeMarkdownIntoRoom(
569568
name: string,
570569
fileId: string,
571570
markdown: string,
572-
{ version, baseVersion }: MergeOrder
571+
{ version }: MergeOrder
573572
): Promise<'applied' | 'no-live-room' | 'merge-unavailable' | 'stale'> {
574573
const store = getFileDocStore()
575574

576575
// The durable version this merge carries is now incorporated in the live doc — record it (cluster-wide
577576
// in Redis for multi-task, plus this task's room) so the persist If-Match guard treats this write as
578577
// synced rather than an out-of-band conflict. AWAITED so the version is durable before the merge lock
579-
// releases, so the next lock holder's staleness check (below) reads a consistent value. Only a durable
580-
// `version` is recorded — a streaming `baseVersion` orders the merge (below) but is never a checkpoint,
581-
// so the synced version stays pinned to the last durable write.
578+
// releases, so the next lock holder's staleness check (below) reads a consistent value.
582579
const recordVersion = async () => {
583580
if (version === undefined) return
584581
const room = fileDocRooms.get(name)
@@ -590,29 +587,13 @@ async function mergeMarkdownIntoRoom(
590587
}
591588

592589
// Order this merge on the file's version line, where `current` is the durable version the doc already
593-
// incorporates. Both keys are DB-monotonic `contentUpdatedAt` values (no wall-clock), so ordering is
594-
// immune to clock skew:
595-
// - A durable `version` is stale if it is NOT strictly newer than `current` — a newer durable write
596-
// already landed (possibly on another process, out of dispatch order); applying its older markdown
597-
// would regress the doc while the monotonic token stays high.
598-
// - A streaming `baseVersion` (the durable version the snapshot was built from) is stale if `current`
599-
// has moved PAST it — a newer durable write landed since the snapshot's base, so diffing the live
600-
// doc back toward the snapshot would clobber that write's content (which a later persist, still
601-
// holding the current If-Match token, would then write over the durable file). This is what stops a
602-
// concurrent human edit from being silently lost; the per-process caller chain cannot see it.
603-
// A merge with neither key is never stale (legacy, unordered). Only a durable `version` is recorded,
604-
// so a streaming snapshot never advances the synced version — the final `edit_content` write does.
605-
//
606-
// Known, accepted limitation: two INDEPENDENT copilot streams editing the SAME file at once share one
607-
// base version, so neither is stale relative to the other and their snapshots can interleave in the live
608-
// doc. This is transient only — each stream's final durable write is version-ordered and reconciles the
609-
// doc, so the steady state is deterministic (last durable wins) and the durable file is never corrupted.
610-
// Ordering two independent snapshot streams would need a shared sequence they don't have; the fully
611-
// robust form (a per-file streaming lease, or embedding the version in each stream entry) is a scoped
612-
// follow-up, not a durability fix owed here.
590+
// incorporates. `version` is a DB-monotonic `contentUpdatedAt` value (no wall-clock), so ordering is
591+
// immune to clock skew: a durable `version` is stale if it is NOT strictly newer than `current` — a
592+
// newer durable write already landed (possibly on another process, out of dispatch order); applying its
593+
// older markdown would regress the doc while the monotonic token stays high. A merge with no `version`
594+
// is never stale (legacy, unordered).
613595
const isStale = (current: number): boolean => {
614596
if (version !== undefined) return version <= current
615-
if (baseVersion !== undefined) return current > baseVersion
616597
return false
617598
}
618599

‎apps/realtime/src/routes/http.ts‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -203,22 +203,22 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) {
203203
return
204204
}
205205

206-
// Merge a copilot edit into a file's LIVE collaborative document so it streams into open editors
207-
// (Stage C). Returns `{ applied }`: when false, no seeded live room exists and the caller writes
208-
// the file directly instead. Any live user edits are preserved — the app builds a minimal CRDT diff.
206+
// Merge a durable file write into a file's LIVE collaborative document so open editors reconcile to
207+
// it (Stage C) — this is the stream-end/durable reconcile, not token-by-token streaming (that is now
208+
// applied client-side by the open editor). Returns `{ applied }`: when false, no seeded live room
209+
// exists and the caller writes the file directly instead. Live user edits are preserved — the app
210+
// builds a minimal CRDT diff.
209211
if (req.method === 'POST' && req.url === '/api/file-doc/apply-edit') {
210212
try {
211213
const body = await readRequestBody(req)
212-
const { fileId, markdown, version, baseVersion } = JSON.parse(body)
214+
const { fileId, markdown, version } = JSON.parse(body)
213215
if (!isNonEmptyString(fileId) || typeof markdown !== 'string') {
214216
return sendError(res, 'Invalid fileId or markdown', 400)
215217
}
216218
// `version` (the durable updatedAt this markdown was written with) records that the live doc now
217219
// incorporates that durable version, so the persist If-Match guard won't flag it as a conflict.
218-
// `baseVersion` is a streaming snapshot's causal base: dropped if a newer durable write landed.
219220
const result = await applyMarkdownToLiveFileDoc(fileId, markdown, {
220221
version: typeof version === 'number' ? version : undefined,
221-
baseVersion: typeof baseVersion === 'number' ? baseVersion : undefined,
222222
})
223223
res.writeHead(200, { 'Content-Type': 'application/json' })
224224
res.end(JSON.stringify({ applied: result === 'applied' }))
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
/**
2+
* @vitest-environment jsdom
3+
*/
4+
import { Editor } from '@tiptap/core'
5+
import { afterEach, beforeAll, describe, expect, it } from 'vitest'
6+
import { Awareness } from 'y-protocols/awareness'
7+
import * as Y from 'yjs'
8+
import { createMarkdownEditorExtensions } from '../editor-extensions'
9+
import { applyStreamedMarkdownToLiveDoc } from './apply-streamed-markdown'
10+
11+
beforeAll(() => {
12+
// jsdom does not implement elementFromPoint; the Placeholder extension's viewport tracking calls it
13+
// on view mount. Returning null makes ProseMirror's posAtCoords fall back gracefully.
14+
if (!document.elementFromPoint) {
15+
document.elementFromPoint = () => null
16+
}
17+
})
18+
19+
/** A headless collaborative editor bound to a fresh Y.Doc — the same extension wiring the component uses. */
20+
function makeCollabEditor() {
21+
const doc = new Y.Doc()
22+
const awareness = new Awareness(doc)
23+
const editor = new Editor({
24+
extensions: createMarkdownEditorExtensions({
25+
placeholder: '',
26+
collaboration: {
27+
doc,
28+
awareness,
29+
user: { name: 'Tester', color: '#ffffff', clientId: doc.clientID },
30+
},
31+
}),
32+
content: '',
33+
})
34+
return { editor, doc, awareness }
35+
}
36+
37+
const teardown: Array<() => void> = []
38+
afterEach(() => {
39+
for (const fn of teardown.splice(0)) fn()
40+
})
41+
42+
function track(t: { editor: Editor; doc: Y.Doc; awareness: Awareness }) {
43+
teardown.push(() => {
44+
t.editor.destroy()
45+
t.awareness.destroy()
46+
t.doc.destroy()
47+
})
48+
return t
49+
}
50+
51+
describe('applyStreamedMarkdownToLiveDoc', () => {
52+
it('streams agent content into the live collaborative doc and broadcasts it as Yjs ops', () => {
53+
const { editor, doc } = track(makeCollabEditor())
54+
55+
expect(applyStreamedMarkdownToLiveDoc(editor, '# Title\n\nHello world.')).toBe(true)
56+
expect(editor.getText()).toContain('Hello world')
57+
58+
// The write lands as ops on the shared doc, so any peer receives it (this is what makes a
59+
// collaborator on /files see the stream without ever holding `streamingContent`).
60+
const peer = new Y.Doc()
61+
Y.applyUpdate(peer, Y.encodeStateAsUpdate(doc))
62+
expect(peer.getXmlFragment('default').toString()).toContain('Hello world')
63+
peer.destroy()
64+
})
65+
66+
it('returns false when the doc is not yet bound (no ySync binding)', () => {
67+
// A plain editor with no collaboration has no ySync binding, so the applier reports "not ready"
68+
// rather than throwing — the streaming tick re-arms until the doc is seeded.
69+
const editor = new Editor({
70+
extensions: createMarkdownEditorExtensions({ placeholder: '' }),
71+
content: '',
72+
})
73+
teardown.push(() => editor.destroy())
74+
expect(applyStreamedMarkdownToLiveDoc(editor, '# Nope')).toBe(false)
75+
})
76+
77+
it('keeps agent-streamed ops out of the undo stack while user edits stay undoable', () => {
78+
const { editor } = track(makeCollabEditor())
79+
80+
applyStreamedMarkdownToLiveDoc(editor, '# Streamed\n\nAgent wrote this.')
81+
// The streamed op used AGENT_STREAM_ORIGIN, which the Collaboration UndoManager does not track —
82+
// so there is nothing to undo, and an undo must not revert the agent's content.
83+
expect(editor.can().undo()).toBe(false)
84+
editor.commands.undo()
85+
expect(editor.getText()).toContain('Agent wrote this')
86+
87+
// A genuine user edit IS captured (origin ySyncPluginKey) — proving the test isn't vacuous:
88+
// undo works, and it reverts only the user edit, leaving the agent content intact.
89+
editor.commands.focus('end')
90+
editor.commands.insertContent(' USER-TYPED')
91+
expect(editor.getText()).toContain('USER-TYPED')
92+
expect(editor.can().undo()).toBe(true)
93+
editor.commands.undo()
94+
expect(editor.getText()).not.toContain('USER-TYPED')
95+
expect(editor.getText()).toContain('Agent wrote this')
96+
})
97+
98+
it('merges an agent write with a concurrent peer edit (minimal diff, no clobber)', () => {
99+
const { editor, doc } = track(makeCollabEditor())
100+
applyStreamedMarkdownToLiveDoc(editor, 'Alpha paragraph.\n\nBeta paragraph.')
101+
102+
// A peer forks the current state and edits the FIRST paragraph directly on the shared type…
103+
const remote = new Y.Doc()
104+
Y.applyUpdate(remote, Y.encodeStateAsUpdate(doc))
105+
const remoteFrag = remote.getXmlFragment('default')
106+
remote.transact(() => {
107+
const firstPara = remoteFrag.get(0) as Y.XmlElement
108+
const textNode = firstPara.get(0) as Y.XmlText
109+
textNode.insert(textNode.toString().length, ' EDITED')
110+
})
111+
112+
// …while the agent rewrites the SECOND paragraph through the live editor binding.
113+
applyStreamedMarkdownToLiveDoc(editor, 'Alpha paragraph.\n\nBeta paragraph, expanded.')
114+
115+
// Exchange updates both ways (as the relay would); a full-document replace would have clobbered
116+
// the peer's concurrent edit — a minimal `updateYFragment` diff preserves both.
117+
Y.applyUpdate(doc, Y.encodeStateAsUpdate(remote, Y.encodeStateVector(doc)))
118+
Y.applyUpdate(remote, Y.encodeStateAsUpdate(doc, Y.encodeStateVector(remote)))
119+
120+
const merged = doc.getXmlFragment('default').toString()
121+
expect(merged).toContain('EDITED')
122+
expect(merged).toContain('expanded')
123+
remote.destroy()
124+
})
125+
})
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
import type { Editor } from '@tiptap/core'
2+
import { Node as PMNode } from '@tiptap/pm/model'
3+
import { updateYFragment, ySyncPluginKey } from '@tiptap/y-tiptap'
4+
import { parseMarkdownToDoc } from '../markdown-parse'
5+
6+
/**
7+
* Transaction origin for agent-streamed writes into a live collaborative doc. It is deliberately NOT
8+
* the `ySyncPluginKey` origin that local user edits use, so the Collaboration UndoManager — which
9+
* tracks only `ySyncPluginKey` — excludes streamed ops from the user's undo stack.
10+
*/
11+
const AGENT_STREAM_ORIGIN = Symbol('agent-stream')
12+
13+
/**
14+
* Apply a streamed markdown body into the editor's live collaborative Y.Doc as a minimal CRDT diff.
15+
*
16+
* Uses the running `ySyncPlugin` binding's {@link updateYFragment} — the same primitive TipTap runs on
17+
* every keystroke — so only the delta between the doc's current content and `body` is written, never a
18+
* full-document replace that would wipe collaborators. Each diff is a small Yjs op that renders locally
19+
* (via the binding's observer, the remote-edit render path) AND broadcasts to every peer, so the stream
20+
* is smooth here and on other clients alike. Runs under {@link AGENT_STREAM_ORIGIN} so the streamed ops
21+
* stay out of the user's undo stack. Returns `false` when the editor has no live ySync binding (e.g. a
22+
* non-collaborative editor); the caller gates seed-readiness separately via `collabReady`.
23+
*/
24+
export function applyStreamedMarkdownToLiveDoc(editor: Editor, body: string): boolean {
25+
const binding = ySyncPluginKey.getState(editor.state)?.binding
26+
if (!binding) return false
27+
const target = PMNode.fromJSON(editor.schema, parseMarkdownToDoc(body))
28+
binding.doc.transact(() => {
29+
updateYFragment(binding.doc, binding.type, target, binding)
30+
}, AGENT_STREAM_ORIGIN)
31+
return true
32+
}

0 commit comments

Comments
 (0)