From 5614012acb55829a804bb2acb09246e68bca1b9f Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Sun, 3 May 2026 13:42:50 +0300 Subject: [PATCH] fix(ui): keep streaming deltas through pipeline --- .../src/sync/__tests__/event-pipeline.test.js | 14 ++++++----- packages/ui/src/sync/event-pipeline.ts | 25 +++---------------- .../ui/src/sync/reconnect-recovery.test.ts | 11 ++++++++ packages/ui/src/sync/reconnect-recovery.ts | 8 +++++- 4 files changed, 30 insertions(+), 28 deletions(-) diff --git a/packages/ui/src/sync/__tests__/event-pipeline.test.js b/packages/ui/src/sync/__tests__/event-pipeline.test.js index f01c724d..bd7a2369 100644 --- a/packages/ui/src/sync/__tests__/event-pipeline.test.js +++ b/packages/ui/src/sync/__tests__/event-pipeline.test.js @@ -275,7 +275,7 @@ describe('createEventPipeline', () => { expect(received[0].payload.type).toBe('server.connected'); }); - it('skips stale message.part.delta events after a newer message.part.updated for the same field', async () => { + it('keeps message.part.delta events when a newer message.part.updated is queued for the same field', async () => { installDomStubs(); let releaseStream; @@ -285,8 +285,8 @@ describe('createEventPipeline', () => { const received = []; - // Simulate: part.updated arrives, then delta, then a newer part.updated for the - // same part. The older queued delta becomes stale and must be skipped. + // The pipeline only routes/coalesces events. Whether this delta is already + // represented by the newer snapshot is reducer state, not queue state. const directory = '/test/dir'; const sdk = createSdkWithEvents([ // T0: message.part.updated for part-A @@ -299,7 +299,7 @@ describe('createEventPipeline', () => { }, }, }, - // T1: message.part.delta for part-A (should be dropped as stale) + // T1: message.part.delta for part-A { payload: { type: 'message.part.delta', @@ -329,7 +329,7 @@ describe('createEventPipeline', () => { sdk, onEvent: (dir, payload) => { received.push({ directory: dir, payload }); - if (received.length === 1) { + if (received.length === 2) { cleanup(); releaseStream(); resolve(); @@ -340,8 +340,10 @@ describe('createEventPipeline', () => { await delivered; - expect(received.length).toBe(1); + expect(received.length).toBe(2); expect(received[0].payload.type).toBe('message.part.updated'); + expect(received[1].payload.type).toBe('message.part.delta'); + expect(received[1].payload.properties.delta).toBe(' world'); }); it('keeps delta events for other fields on the same part', async () => { diff --git a/packages/ui/src/sync/event-pipeline.ts b/packages/ui/src/sync/event-pipeline.ts index f63e69d8..91346823 100644 --- a/packages/ui/src/sync/event-pipeline.ts +++ b/packages/ui/src/sync/event-pipeline.ts @@ -1,6 +1,10 @@ /** * Event Pipeline — transport connection, event coalescing, and batched flush. * + * This module must not make state-dependent decisions about event validity. + * For example, deciding whether a delta is already represented by a full part + * snapshot belongs in the reducer, which has access to the current state. + * * Plain closure API: * const { cleanup } = createEventPipeline({ sdk, onEvent }) * @@ -145,7 +149,6 @@ type DirectoryQueue = { queue: Event[] buffer: Event[] coalesced: Map - staleDeltas: Set timer: ReturnType | undefined last: number } @@ -185,7 +188,6 @@ export function createEventPipeline(input: EventPipelineInput) { queue: [], buffer: [], coalesced: new Map(), - staleDeltas: new Set(), timer: undefined, last: 0, } @@ -212,8 +214,6 @@ export function createEventPipeline(input: EventPipelineInput) { return undefined } - const deltaKey = (messageID: string, partID: string, field: string) => `${messageID}:${partID}:${field}` - const flushDir = (directory: string) => { const d = directories.get(directory) if (!d) return @@ -224,22 +224,14 @@ export function createEventPipeline(input: EventPipelineInput) { if (d.queue.length === 0) return const events = d.queue - const staleDeltas = d.staleDeltas.size > 0 ? new Set(d.staleDeltas) : undefined d.queue = d.buffer d.buffer = events d.queue.length = 0 d.coalesced.clear() - d.staleDeltas.clear() d.last = Date.now() syncDebug.pipeline.flush(events.length) for (const payload of events) { - if (staleDeltas && payload.type === "message.part.delta") { - const props = payload.properties as { messageID: string; partID: string; field: string } - if (staleDeltas.has(deltaKey(props.messageID, props.partID, props.field))) { - continue - } - } onEvent(directory, payload) } @@ -313,15 +305,6 @@ export function createEventPipeline(input: EventPipelineInput) { } as unknown as Event } else { d.queue[i] = normalizedPayload - if (normalizedPayload.type === "message.part.updated") { - const part = (normalizedPayload.properties as { part: Record & { messageID: string; id: string } }).part - for (const field of ["text", "output"] as const) { - const value = part[field] - if (typeof value === "string" && value.length > 0) { - d.staleDeltas.add(deltaKey(part.messageID, part.id, field)) - } - } - } } syncDebug.pipeline.coalesced(normalizedPayload.type, k) return diff --git a/packages/ui/src/sync/reconnect-recovery.test.ts b/packages/ui/src/sync/reconnect-recovery.test.ts index e4e8e016..fcede14b 100644 --- a/packages/ui/src/sync/reconnect-recovery.test.ts +++ b/packages/ui/src/sync/reconnect-recovery.test.ts @@ -54,6 +54,17 @@ describe("getReconnectCandidateSessionIds", () => { }).sort()).toContain("active") }) + test("includes completed assistant sessions when the latest assistant parts are missing", () => { + expect(getReconnectCandidateSessionIds({ + session: [createSession("blank")], + session_status: { blank: { type: "idle" } as SessionStatus }, + message: { + blank: [createAssistantMessage("m-1", "blank", 1)], + }, + part: {}, + })).toEqual(["blank"]) + }) + test("does not include a viewed session from another directory", () => { expect(getReconnectCandidateSessionIds({ session: [createSession("active")], diff --git a/packages/ui/src/sync/reconnect-recovery.ts b/packages/ui/src/sync/reconnect-recovery.ts index 7a4a44f0..e7e8c75f 100644 --- a/packages/ui/src/sync/reconnect-recovery.ts +++ b/packages/ui/src/sync/reconnect-recovery.ts @@ -1,10 +1,11 @@ -import type { SessionStatus, Message } from "@opencode-ai/sdk/v2/client" +import type { SessionStatus, Message, Part } from "@opencode-ai/sdk/v2/client" import type { Session } from "@opencode-ai/sdk/v2" type ReconnectRecoveryState = { session: Session[] session_status?: Record message?: Record + part?: Record } export type ViewedSessionRecoveryTarget = { @@ -26,12 +27,17 @@ export function getReconnectCandidateSessionIds(state: ReconnectRecoveryState, o for (const [sessionId, messages] of Object.entries(state.message ?? {})) { const lastMessage = messages[messages.length - 1] + const lastAssistantComplete = lastMessage + && lastMessage.role === "assistant" + && typeof (lastMessage as { time?: { completed?: number } }).time?.completed === "number" if ( lastMessage && lastMessage.role === "assistant" && typeof (lastMessage as { time?: { completed?: number } }).time?.completed !== "number" ) { ids.add(sessionId) + } else if (lastAssistantComplete && state.part && (state.part[lastMessage.id]?.length ?? 0) === 0) { + ids.add(sessionId) } }