fix(ui): throttle SSE part deltas and prevent stale delta text duplication (#536)

## Problem

Streaming responses send many `message.part.delta` events per second.
Each event triggers a Solid store mutation + re-render, causing:

1. **Mobile freezes** — KaTeX/highlight.js re-renders on every delta
block the main thread
2. **Text duplication** — When `message.part.updated` replaces a part
with the server's complete state, previously enqueued deltas would flush
AFTER the replacement, concatenating stale text onto the fresh part

## Solution

### 1. Delta throttling (50ms buffer)
Accumulate deltas in a buffer and flush every 50ms instead of updating
on every event. This batches Solid mutations and cuts main-thread
blocking during streaming.

### 2. Stale delta cleanup
When a full `message.part.updated` arrives, clear any pending deltas for
that part before applying the update. The update already contains the
complete server state, so accumulated deltas are stale.

## Changes

- `packages/ui/src/stores/session-events.ts`: Added `enqueueDelta`,
`flushDeltas`, `clearPendingDeltasForPart`; updated
`handleMessagePartDelta` and `handleMessageUpdate`

## Verification

- Build passes: `npm run build:ui`
- Tested on mobile — eliminates visible freezes during long streaming
responses
- No duplicate text observed at end of assistant messages after part
updates

---------
This commit is contained in:
Dark 2026-06-13 10:42:17 -04:00 committed by GitHub
parent 17547a15d8
commit 14dfd3e371
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 219 additions and 1 deletions

View file

@ -0,0 +1,102 @@
import assert from "node:assert/strict"
import { afterEach, beforeEach, describe, it } from "node:test"
import { setTimeout as delay } from "node:timers/promises"
import {
clearPendingDeltasForPart,
enqueueDelta,
flushPendingDeltasForMessage,
resetDeltaBufferForTests,
setFlushCallback,
} from "./delta-buffer.ts"
type DeltaBatch = Array<{ instanceId: string; messageId: string; partId: string; field: string; delta: string }>
describe("delta buffer", () => {
beforeEach(() => {
resetDeltaBufferForTests()
})
afterEach(() => {
resetDeltaBufferForTests()
})
it("concatenates matching deltas and flushes them once", async () => {
const flushed: DeltaBatch[] = []
setFlushCallback((batch) => flushed.push(batch))
enqueueDelta("instance-1", "message-1", "part-1", "text", "hello")
enqueueDelta("instance-1", "message-1", "part-1", "text", " world")
await delay(75)
assert.equal(flushed.length, 1)
assert.deepEqual(flushed[0], [
{ instanceId: "instance-1", messageId: "message-1", partId: "part-1", field: "text", delta: "hello world" },
])
})
it("clears pending deltas for a full part update before a stale timer flush", async () => {
const flushed: DeltaBatch[] = []
setFlushCallback((batch) => flushed.push(batch))
enqueueDelta("instance-1", "message-1", "part-1", "text", "stale")
clearPendingDeltasForPart("instance-1", "message-1", "part-1")
await delay(75)
assert.deepEqual(flushed, [])
})
it("flushes pending message deltas before applying message.updated", async () => {
const timerFlushes: DeltaBatch[] = []
const applied: Array<{ instanceId: string; delta: { messageId: string; partId: string; field: string; delta: string } }> = []
setFlushCallback((batch) => timerFlushes.push(batch))
enqueueDelta("instance-1", "message-1", "part-1", "text", "before update")
flushPendingDeltasForMessage("instance-1", "message-1", (instanceId, delta) => {
applied.push({ instanceId, delta })
})
await delay(75)
assert.deepEqual(applied, [
{
instanceId: "instance-1",
delta: { messageId: "message-1", partId: "part-1", field: "text", delta: "before update" },
},
])
assert.deepEqual(timerFlushes, [])
})
it("keeps clear and flush operations isolated by instance, message, and part", async () => {
const timerFlushes: DeltaBatch[] = []
const applied: Array<{ instanceId: string; delta: { messageId: string; partId: string; field: string; delta: string } }> = []
setFlushCallback((batch) => timerFlushes.push(batch))
enqueueDelta("instance-1", "message-1", "part-1", "text", "drop")
enqueueDelta("instance-1", "message-1", "part-2", "text", "same message")
enqueueDelta("instance-1", "message-2", "part-1", "text", "other message")
enqueueDelta("instance-2", "message-1", "part-1", "text", "other instance")
clearPendingDeltasForPart("instance-1", "message-1", "part-1")
flushPendingDeltasForMessage("instance-1", "message-1", (instanceId, delta) => {
applied.push({ instanceId, delta })
})
await delay(75)
assert.deepEqual(applied, [
{
instanceId: "instance-1",
delta: { messageId: "message-1", partId: "part-2", field: "text", delta: "same message" },
},
])
assert.deepEqual(timerFlushes, [
[
{ instanceId: "instance-1", messageId: "message-2", partId: "part-1", field: "text", delta: "other message" },
{ instanceId: "instance-2", messageId: "message-1", partId: "part-1", field: "text", delta: "other instance" },
],
])
})
})

View file

@ -0,0 +1,89 @@
/**
* Delta buffer for throttling SSE message.part.delta events.
*
* Accumulates text deltas in a 50ms window to reduce UI churn from
* high-frequency streaming chunks. Provides targeted flush/clear paths
* so full part-update or message-complete events always win over stale
* buffered deltas.
*/
const DELTA_FLUSH_INTERVAL = 50
const pendingDeltas = new Map<string, { instanceId: string; messageId: string; partId: string; field: string; delta: string }>()
let deltaFlushTimer: ReturnType<typeof setTimeout> | null = null
export function enqueueDelta(instanceId: string, messageId: string, partId: string, field: string, delta: string) {
const key = `${instanceId}:${messageId}:${partId}:${field}`
const existing = pendingDeltas.get(key)
const accumulated = existing ? existing.delta + delta : delta
pendingDeltas.set(key, { instanceId, messageId, partId, field, delta: accumulated })
if (deltaFlushTimer === null) {
deltaFlushTimer = setTimeout(flushDeltas, DELTA_FLUSH_INTERVAL)
}
}
export function clearPendingDeltasForPart(instanceId: string, messageId: string, partId: string) {
const keysToDelete: string[] = []
for (const key of pendingDeltas.keys()) {
if (key.startsWith(`${instanceId}:${messageId}:${partId}:`)) {
keysToDelete.push(key)
}
}
for (const key of keysToDelete) {
pendingDeltas.delete(key)
}
}
export function flushPendingDeltasForMessage(
instanceId: string,
messageId: string,
applyDelta: (instanceId: string, delta: { messageId: string; partId: string; field: string; delta: string }) => void
): void {
const prefix = `${instanceId}:${messageId}:`
const keysToFlush: string[] = []
for (const key of pendingDeltas.keys()) {
if (key.startsWith(prefix)) {
keysToFlush.push(key)
}
}
for (const key of keysToFlush) {
const pending = pendingDeltas.get(key)
if (pending) {
pendingDeltas.delete(key)
applyDelta(instanceId, {
messageId: pending.messageId,
partId: pending.partId,
field: pending.field,
delta: pending.delta,
})
}
}
}
export function setFlushCallback(
callback: (batch: Array<{ instanceId: string; messageId: string; partId: string; field: string; delta: string }>) => void
) {
// Store callback for flushDeltas to use
flushCallback = callback
}
export function resetDeltaBufferForTests() {
pendingDeltas.clear()
if (deltaFlushTimer !== null) {
clearTimeout(deltaFlushTimer)
deltaFlushTimer = null
}
flushCallback = null
}
let flushCallback: ((batch: Array<{ instanceId: string; messageId: string; partId: string; field: string; delta: string }>) => void) | null = null
function flushDeltas() {
deltaFlushTimer = null
if (pendingDeltas.size === 0) return
const batch = Array.from(pendingDeltas.values())
pendingDeltas.clear()
if (flushCallback) {
flushCallback(batch)
}
}

View file

@ -17,6 +17,12 @@ import type { MessageStatus } from "./message-v2/types"
import { getLogger } from "../lib/logger"
import { requestData } from "../lib/opencode-api"
import {
enqueueDelta,
clearPendingDeltasForPart,
flushPendingDeltasForMessage,
setFlushCallback,
} from "./delta-buffer"
import {
getPermissionId,
getPermissionKind,
@ -384,6 +390,12 @@ function handleMessageUpdate(instanceId: string, event: MessageUpdateEvent | Mes
upsertMessageInfoV2(instanceId, messageInfo, { status: "streaming" })
}
// Clear any pending deltas for this part before applying the full part update.
// The part update contains the complete state from the server, so accumulated
// deltas would be stale and cause duplication if flushed later.
if (part.id) {
clearPendingDeltasForPart(instanceId, messageId, part.id)
}
applyPartUpdateV2(instanceId, { ...part, sessionID: sessionId, messageID: messageId })
handleConversationAssistantPartUpdated(instanceId, { ...part, sessionID: sessionId, messageID: messageId }, messageInfo)
@ -402,6 +414,14 @@ function handleMessageUpdate(instanceId: string, event: MessageUpdateEvent | Mes
const messageId = typeof info.id === "string" ? info.id : undefined
if (!sessionId || !messageId) return
// Flush any pending deltas for this message before applying the update.
// Deltas are buffered for up to 50ms; if message.updated arrives before
// the buffer flushes, the message could be marked complete/error with
// stale text mutations still pending. Flushing first preserves the
// server's event ordering: all delta content is applied, then the
// message status/metadata update runs on the complete content.
flushPendingDeltasForMessage(instanceId, messageId, applyPartDeltaV2)
const timeInfo = (info.time ?? {}) as { created?: number; updated?: number; end?: number }
const nextUpdated =
typeof timeInfo.end === "number" && timeInfo.end > 0
@ -453,12 +473,19 @@ function handleMessageUpdate(instanceId: string, event: MessageUpdateEvent | Mes
}
}
// Delta buffer callback setup
setFlushCallback((batch) => {
for (const { instanceId, messageId, partId, field, delta } of batch) {
applyPartDeltaV2(instanceId, { messageId, partId, field, delta })
}
})
function handleMessagePartDelta(instanceId: string, event: MessagePartDeltaEvent): void {
const props = event.properties
if (!props) return
const { messageID, partID, field, delta } = props
if (!messageID || !partID || !field || typeof delta !== "string") return
applyPartDeltaV2(instanceId, { messageId: messageID, partId: partID, field, delta })
enqueueDelta(instanceId, messageID, partID, field, delta)
}
function handleSessionUpdate(instanceId: string, event: EventSessionUpdated): void {