From 9715b6c23cafdc1ae005d953d918f48f16e11ed3 Mon Sep 17 00:00:00 2001 From: Isaac Sanchez-Hawkins <266845420+isanchez404@users.noreply.github.com> Date: Fri, 8 May 2026 16:14:07 -0400 Subject: [PATCH] fix(sync): preserve part update ordering (#1167) * fix(sync): preserve part update ordering * test(sync): wait for pipeline delivery --------- Co-authored-by: Isaac Sanchez --- packages/ui/src/sync/event-pipeline.test.ts | 99 +++++++++++++++++++++ packages/ui/src/sync/event-pipeline.ts | 4 - 2 files changed, 99 insertions(+), 4 deletions(-) create mode 100644 packages/ui/src/sync/event-pipeline.test.ts diff --git a/packages/ui/src/sync/event-pipeline.test.ts b/packages/ui/src/sync/event-pipeline.test.ts new file mode 100644 index 00000000..9a4ea81f --- /dev/null +++ b/packages/ui/src/sync/event-pipeline.test.ts @@ -0,0 +1,99 @@ +import { describe, expect, test } from "bun:test" +import type { Event, OpencodeClient } from "@opencode-ai/sdk/v2/client" +import { createEventPipeline } from "./event-pipeline" + +const failAfter = (ms: number) => new Promise((_, reject) => { + setTimeout(() => reject(new Error("Timed out waiting for event pipeline flush")), ms) +}) + +function partUpdatedEvent(text: string): Event { + return { + type: "message.part.updated", + properties: { + part: { + id: "prt_1", + messageID: "msg_1", + sessionID: "ses_1", + type: "text", + text, + }, + }, + } as Event +} + +function deltaEvent(delta: string): Event { + return { + type: "message.part.delta", + properties: { + messageID: "msg_1", + partID: "prt_1", + field: "text", + delta, + }, + } as Event +} + +function createSdk(events: Event[], streamFinished: () => void): OpencodeClient { + return { + global: { + event: async ({ signal }: { signal: AbortSignal }) => ({ + stream: (async function* () { + for (const payload of events) { + yield { directory: "/repo", payload } + } + streamFinished() + await new Promise((resolve) => { + if (signal.aborted) { + resolve() + return + } + signal.addEventListener("abort", () => resolve(), { once: true }) + }) + })(), + }), + }, + } as unknown as OpencodeClient +} + +describe("createEventPipeline", () => { + test("preserves part update order around text deltas", async () => { + let resolveStreamFinished!: () => void + const streamFinished = new Promise((resolve) => { + resolveStreamFinished = resolve + }) + let resolveDelivered!: () => void + const deliveredAll = new Promise((resolve) => { + resolveDelivered = resolve + }) + const delivered: Event[] = [] + const pipeline = createEventPipeline({ + sdk: createSdk([ + partUpdatedEvent("a"), + deltaEvent("b"), + partUpdatedEvent("ab"), + ], resolveStreamFinished), + onEvent: (_directory, payload) => { + delivered.push(payload) + if (delivered.length === 3) { + resolveDelivered() + } + }, + transport: "sse", + heartbeatTimeoutMs: 1_000, + }) + + try { + await streamFinished + await Promise.race([deliveredAll, failAfter(500)]) + } finally { + pipeline.cleanup() + } + + expect(delivered.map((event) => { + if (event.type === "message.part.delta") { + return `delta:${(event.properties as { delta: string }).delta}` + } + return `updated:${((event.properties as { part: { text: string } }).part).text}` + })).toEqual(["updated:a", "delta:b", "updated:ab"]) + }) +}) diff --git a/packages/ui/src/sync/event-pipeline.ts b/packages/ui/src/sync/event-pipeline.ts index e680c536..7e8282a8 100644 --- a/packages/ui/src/sync/event-pipeline.ts +++ b/packages/ui/src/sync/event-pipeline.ts @@ -211,10 +211,6 @@ export function createEventPipeline(input: EventPipelineInput) { if (payload.type === "lsp.updated") { return "lsp.updated" } - if (payload.type === "message.part.updated") { - const part = (payload.properties as { part: { messageID: string; id: string } }).part - return `message.part.updated:${part.messageID}:${part.id}` - } if (payload.type === "message.part.delta") { const props = payload.properties as { messageID: string; partID: string; field: string } return `message.part.delta:${props.messageID}:${props.partID}:${props.field}`