From 2a4def500410d66396b3d4c061661d634bfca07a Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Tue, 27 Oct 2026 03:58:51 +0100 Subject: [PATCH] fix(server): publish OpenCode snapshots before live suffixes Commit snapshot cursors and publish each event together. Holding those events until attachment IO completed let live deltas overtake text that had already advanced the cursor. Require snapshot publishers and exercise text and reasoning interleaving against real image materialization. --- .../opencode/event-translator.test.ts | 27 ++++--- .../test-utils/v2-streamed-text-memory.ts | 3 +- .../opencode/test-utils/v2-timeline.ts | 15 ++++ .../agent/providers/opencode/v2/children.ts | 4 +- .../agent/providers/opencode/v2/session.ts | 10 ++- .../providers/opencode/v2/streaming.test.ts | 78 +++++++++++++++++-- .../providers/opencode/v2/timeline.test.ts | 7 +- .../agent/providers/opencode/v2/timeline.ts | 18 +++-- 8 files changed, 128 insertions(+), 34 deletions(-) create mode 100644 packages/server/src/server/agent/providers/opencode/test-utils/v2-timeline.ts diff --git a/packages/server/src/server/agent/providers/opencode/event-translator.test.ts b/packages/server/src/server/agent/providers/opencode/event-translator.test.ts index a6828ad666..6dc5bd5a2e 100644 --- a/packages/server/src/server/agent/providers/opencode/event-translator.test.ts +++ b/packages/server/src/server/agent/providers/opencode/event-translator.test.ts @@ -1,3 +1,4 @@ +import { collectTimelineSnapshot } from "./test-utils/v2-timeline.js"; import type { SessionMessageAssistant, SessionMessageUser } from "@opencode/client"; import { V2Timeline } from "./v2/timeline.js"; import { describe, expect, it } from "vitest"; @@ -1568,13 +1569,13 @@ describe("OpenCode v2 history reconciliation", () => { time: { created: 1 }, metadata: { paseoClientMessageId: "client-message" }, }; - expect(await new V2Timeline().messages([user])).toMatchObject([ + expect(await collectTimelineSnapshot(new V2Timeline(), [user])).toMatchObject([ { item: { clientMessageId: "client-message", messageId: "native-message" } }, ]); - expect(await new V2Timeline(false).messages([user])).toMatchObject([ + expect(await collectTimelineSnapshot(new V2Timeline(false), [user])).toMatchObject([ { item: { type: "user_message", messageId: "native-message" } }, ]); - const [event] = await new V2Timeline(false).messages([user]); + const [event] = await collectTimelineSnapshot(new V2Timeline(false), [user]); expect(event && "item" in event && event.item).not.toHaveProperty("clientMessageId"); }); function assistant(text: string): SessionMessageAssistant { @@ -1589,23 +1590,25 @@ describe("OpenCode v2 history reconciliation", () => { } it("emits only the missing suffix after reconnect, even if an intermediate snapshot lags", async () => { const timeline = new V2Timeline(); - const text = (messages: Awaited>) => + const text = (messages: Awaited>) => messages.flatMap((event) => event.type === "timeline" && event.item.type === "assistant_message" ? [event.item.text] : [], ); - expect(text(await timeline.messages([assistant("hello")]))).toEqual(["hello"]); - expect(await timeline.messages([assistant("hel")])).toEqual([]); - expect(text(await timeline.messages([assistant("hello world")]))).toEqual([" world"]); - expect(await timeline.messages([assistant("hello world")])).toEqual([]); + expect(text(await collectTimelineSnapshot(timeline, [assistant("hello")]))).toEqual(["hello"]); + expect(await collectTimelineSnapshot(timeline, [assistant("hel")])).toEqual([]); + expect(text(await collectTimelineSnapshot(timeline, [assistant("hello world")]))).toEqual([ + " world", + ]); + expect(await collectTimelineSnapshot(timeline, [assistant("hello world")])).toEqual([]); }); it("keeps reasoning separate from assistant output", async () => { const timeline = new V2Timeline(); const message = assistant("answer"); message.content.unshift({ type: "reasoning", text: "thinking" }); expect( - (await timeline.messages([message])).map((event) => + (await collectTimelineSnapshot(timeline, [message])).map((event) => event.type === "timeline" ? event.item : null, ), ).toEqual([ @@ -1625,7 +1628,7 @@ describe("OpenCode v2 history reconciliation", () => { state: { status: "running", input: { command: "pwd" }, metadata: {} }, }, ]; - expect(await timeline.messages([message])).toMatchObject([ + expect(await collectTimelineSnapshot(timeline, [message])).toMatchObject([ { item: { type: "tool_call", callId: "call", status: "running" } }, ]); message.content = [ @@ -1641,9 +1644,9 @@ describe("OpenCode v2 history reconciliation", () => { }, }, ]; - expect(await timeline.messages([message])).toMatchObject([ + expect(await collectTimelineSnapshot(timeline, [message])).toMatchObject([ { item: { type: "tool_call", callId: "call", status: "completed" } }, ]); - expect(await timeline.messages([message])).toEqual([]); + expect(await collectTimelineSnapshot(timeline, [message])).toEqual([]); }); }); diff --git a/packages/server/src/server/agent/providers/opencode/test-utils/v2-streamed-text-memory.ts b/packages/server/src/server/agent/providers/opencode/test-utils/v2-streamed-text-memory.ts index e855a8b595..98cd79526a 100644 --- a/packages/server/src/server/agent/providers/opencode/test-utils/v2-streamed-text-memory.ts +++ b/packages/server/src/server/agent/providers/opencode/test-utils/v2-streamed-text-memory.ts @@ -1,3 +1,4 @@ +import { collectTimelineSnapshot } from "./v2-timeline.js"; import { TimelineProjection } from "../../../timeline-projection.js"; import { V2Timeline } from "../v2/timeline.js"; @@ -22,7 +23,7 @@ for (let index = 0; index < chunks; index++) { mode === "delta" ? timeline.delta({ ...part, delta }) : ( - await timeline.messages([ + await collectTimelineSnapshot(timeline, [ { id: "answer", type: "assistant", diff --git a/packages/server/src/server/agent/providers/opencode/test-utils/v2-timeline.ts b/packages/server/src/server/agent/providers/opencode/test-utils/v2-timeline.ts new file mode 100644 index 0000000000..e7aa552479 --- /dev/null +++ b/packages/server/src/server/agent/providers/opencode/test-utils/v2-timeline.ts @@ -0,0 +1,15 @@ +import type { SessionMessageInfo } from "@opencode/client"; +import type { AgentStreamEvent } from "../../../agent-sdk-types.js"; +import type { V2Timeline } from "../v2/timeline.js"; + +// Snapshot assertions without a live stream can collect the publisher's events. +export async function collectTimelineSnapshot( + timeline: V2Timeline, + messages: SessionMessageInfo[], +): Promise { + const events: AgentStreamEvent[] = []; + await timeline.messages(messages, (event) => { + events.push(event); + }); + return events; +} diff --git a/packages/server/src/server/agent/providers/opencode/v2/children.ts b/packages/server/src/server/agent/providers/opencode/v2/children.ts index 1ec3bf01f8..08af88dcb0 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/children.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/children.ts @@ -92,13 +92,13 @@ export class SessionChildren { this.childStates.set(info.id, signature); this.options.emit({ type: "provider_subagent", provider: "opencode", event: presentation }); } - for (const event of await timeline.messages(await messages(this.options.client(), info.id))) { + await timeline.messages(await messages(this.options.client(), info.id), (event) => { if (event.type === "timeline") this.options.emit({ type: "provider_subagent", provider: "opencode", event: { type: "timeline", id: info.id, item: event.item, timestamp: event.timestamp }, }); - } + }); } } diff --git a/packages/server/src/server/agent/providers/opencode/v2/session.ts b/packages/server/src/server/agent/providers/opencode/v2/session.ts index 865c576e52..f173ccb301 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/session.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/session.ts @@ -131,7 +131,7 @@ export class OpenCodeV2Session implements AgentSession { await this.configureConnection(); const location = { directory: this.config.cwd }; this.modes = modesFromV2((await this.client.agent.list({ location })).data); - await this.timeline.messages(await messages(this.client, this.id)); + await this.timeline.messages(await messages(this.client, this.id), () => {}); await this.startStream(); await this.children.reconcile(this.id); } @@ -261,7 +261,11 @@ export class OpenCodeV2Session implements AgentSession { async *streamHistory() { await this.reconnectIfExited(); const history = new V2Timeline(false); - yield* await history.messages(await messages(this.client, this.id)); + const events: AgentStreamEvent[] = []; + await history.messages(await messages(this.client, this.id), (event) => { + events.push(event); + }); + yield* events; } async getRuntimeInfo() { await this.reconnectIfExited(); @@ -417,7 +421,7 @@ export class OpenCodeV2Session implements AgentSession { ]); this.info = info; this.history = history; - for (const event of await this.timeline.messages(history)) this.emitTimeline(event); + await this.timeline.messages(history, (event) => this.emitTimeline(event)); await this.permissions.reconcile(this.id); } private async reconcile() { diff --git a/packages/server/src/server/agent/providers/opencode/v2/streaming.test.ts b/packages/server/src/server/agent/providers/opencode/v2/streaming.test.ts index ef30689255..b8c118f037 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/streaming.test.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/streaming.test.ts @@ -1,3 +1,4 @@ +import { collectTimelineSnapshot } from "../test-utils/v2-timeline.js"; import { V2Timeline } from "./timeline.js"; import type { SessionMessageAssistant } from "@opencode/client"; import { describe, expect, test } from "vitest"; @@ -99,11 +100,13 @@ describe("OpenCode v2 token streaming", () => { }, { type: "text", text: "Hello" }, ]); - expect(await timeline.messages([snapshot])).toMatchObject([{ item: { text: "lo" } }]); + expect(await collectTimelineSnapshot(timeline, [snapshot])).toMatchObject([ + { item: { text: "lo" } }, + ]); expect(timeline.delta({ ...text, delta: "lo" })).toBeNull(); - expect(await timeline.messages([snapshot])).toEqual([]); + expect(await collectTimelineSnapshot(timeline, [snapshot])).toEqual([]); expect(timeline.delta({ ...text, delta: "!" })).toMatchObject({ item: { text: "!" } }); - expect(await timeline.messages([snapshot])).toEqual([]); + expect(await collectTimelineSnapshot(timeline, [snapshot])).toEqual([]); }); test("recovers missed fragments from snapshots after reconnect", async () => { @@ -113,10 +116,12 @@ describe("OpenCode v2 token streaming", () => { expect(timeline.delta({ ...part, delta: "Hel" })).toMatchObject({ item: { text: "Hel" } }); timeline.resetStreams(); expect(timeline.delta({ ...part, delta: "!" })).toBeNull(); - expect(await timeline.messages([assistant([{ type: "text", text: "Hello!" }])])).toMatchObject([ - { item: { text: "lo!" } }, - ]); - expect(await timeline.messages([assistant([{ type: "text", text: "Hello!" }])])).toEqual([]); + expect( + await collectTimelineSnapshot(timeline, [assistant([{ type: "text", text: "Hello!" }])]), + ).toMatchObject([{ item: { text: "lo!" } }]); + expect( + await collectTimelineSnapshot(timeline, [assistant([{ type: "text", text: "Hello!" }])]), + ).toEqual([]); }); test("withholds streamed prose during a structured-output turn", async () => { @@ -203,3 +208,62 @@ describe("OpenCode v2 token streaming", () => { } }); }); + +test.each(["text", "reasoning"] as const)( + "publishes snapshot %s before a live suffix during image materialization", + async (type) => { + const timeline = new V2Timeline(); + const part = { assistantMessageID: "answer", type, ordinal: 0 }; + timeline.startPart(part); + const emitted: AgentStreamEvent[] = []; + let settled = false; + const snapshot = timeline + .messages( + [ + assistant([ + { type, text: "Hello" }, + { + type: "tool", + id: "screenshot", + name: "paseo_browser_screenshot", + executed: false, + state: { + status: "completed", + input: {}, + content: [ + { type: "file", uri: "data:image/png;base64,aGVsbG8=", mime: "image/png" }, + ], + metadata: {}, + }, + time: { created: 2, ran: 2, completed: 2 }, + }, + ]), + ], + (event) => { + emitted.push(event); + }, + ) + .then(() => { + settled = true; + return undefined; + }); + await new Promise((resolve) => setImmediate(resolve)); + expect(settled).toBe(false); + const delta = timeline.delta({ ...part, delta: "Hello world" }); + if (delta) emitted.push(delta); + await snapshot; + const text = emitted.flatMap((event) => { + if (event.type !== "timeline") return []; + if ( + type === "text" && + event.item.type === "assistant_message" && + !event.item.text.includes("![Image]") + ) + return [event.item.text]; + if (type === "reasoning" && event.item.type === "reasoning") return [event.item.text]; + return []; + }); + expect(text).toEqual(["Hello", " world"]); + expect(text.join("")).toBe("Hello world"); + }, +); diff --git a/packages/server/src/server/agent/providers/opencode/v2/timeline.test.ts b/packages/server/src/server/agent/providers/opencode/v2/timeline.test.ts index e5c23bb25a..6e9b886c09 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/timeline.test.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/timeline.test.ts @@ -1,3 +1,4 @@ +import { collectTimelineSnapshot } from "../test-utils/v2-timeline.js"; import type { SessionMessageAssistant } from "@opencode/client"; import { execFileSync } from "node:child_process"; import { fileURLToPath } from "node:url"; @@ -45,7 +46,7 @@ test("a completed edit carries the replaced and replacement text", async () => { }, ]); - expect(await new V2Timeline().messages([edit])).toMatchObject([ + expect(await collectTimelineSnapshot(new V2Timeline(), [edit])).toMatchObject([ { item: { type: "tool_call", @@ -80,7 +81,7 @@ test("a tool's image output is shown as an image, not as base64 in the tool outp ]); const timeline = new V2Timeline(); - const events = await timeline.messages([screenshot]); + const events = await collectTimelineSnapshot(timeline, [screenshot]); expect(JSON.stringify(events)).not.toContain(png); expect(events.map((event) => event.type === "timeline" && event.item)).toMatchObject([ @@ -95,7 +96,7 @@ test("a tool's image output is shown as an image, not as base64 in the tool outp text: expect.stringMatching(/^\n\n!\[Image\]\(file:\/\/.*paseo-attachments.*\.png\)\n\n$/), }, ]); - expect(await timeline.messages([screenshot])).toEqual([]); + expect(await collectTimelineSnapshot(timeline, [screenshot])).toEqual([]); }); test("a surrogate pair split across deltas streams back intact", () => { diff --git a/packages/server/src/server/agent/providers/opencode/v2/timeline.ts b/packages/server/src/server/agent/providers/opencode/v2/timeline.ts index 944699e22e..5596443bbb 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/timeline.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/timeline.ts @@ -57,24 +57,30 @@ export class V2Timeline { private pendingMessages: Promise = Promise.resolve(); - messages(messages: SessionMessageInfo[]): Promise { - const result = this.pendingMessages.then(() => this.projectMessages(messages)); + // Publish each cursor advance in the same synchronous step. Buffering snapshot + // events until image IO finishes lets live deltas overtake already committed text. + messages( + messages: SessionMessageInfo[], + publish: (event: AgentStreamEvent) => void, + ): Promise { + const result = this.pendingMessages.then(() => this.publishMessages(messages, publish)); this.pendingMessages = result.catch(() => {}); return result; } - private async projectMessages(messages: SessionMessageInfo[]): Promise { - const events: AgentStreamEvent[] = []; + private async publishMessages( + messages: SessionMessageInfo[], + publish: (event: AgentStreamEvent) => void, + ): Promise { const state: TimelineState = { structured: false, accepted: false }; for (const message of messages) { const timestamp = new Date(message.time.created).toISOString(); const push = (item: AgentTimelineItem) => - events.push({ type: "timeline", provider: "opencode", item, timestamp }); + publish({ type: "timeline", provider: "opencode", item, timestamp }); if (message.type === "user") this.userMessage(message, state, push); else if (message.type === "assistant") await this.assistantMessage(message, state, push); else if (message.type === "compaction") this.compactionMessage(message, push); } - return events; } private userMessage(