From 694577e86f522e51df768cdcfd4ecd80d05e3d28 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sat, 24 Oct 2026 17:18:36 -0500 Subject: [PATCH 1/3] fix(server): long streamed replies no longer store a full copy per paragraph In paragraph mode each passed streaming snapshot becomes a durable event holding the whole text, so a long reasoning block stored bytes growing with the square of its length. Past 8,000 characters, a snapshot now passes only once its text is a quarter longer than the last one. Fixes #23365 Co-Authored-By: Claude Opus 5.5 --- .../assistantStreaming.test.ts | 63 ++++++++++++++++++- .../orchestration-v2/assistantStreaming.ts | 14 +++++ 2 files changed, 76 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/assistantStreaming.test.ts b/apps/server/src/orchestration-v2/assistantStreaming.test.ts index d71042312bfb1..e8b8cd6d20abb 100644 --- a/apps/server/src/orchestration-v2/assistantStreaming.test.ts +++ b/apps/server/src/orchestration-v2/assistantStreaming.test.ts @@ -1,5 +1,15 @@ +import { + NodeId, + ProviderDriverKind, + RunId, + ThreadId, + TurnItemId, + type OrchestrationV2TurnItem, +} from "@t3tools/contracts"; +import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as DateTime from "effect/DateTime"; import { describe, expect, it } from "vite-plus/test"; -import { splitBufferedAssistantText } from "./assistantStreaming.ts"; +import { makeAssistantStreamingFilter, splitBufferedAssistantText } from "./assistantStreaming.ts"; describe("V2 assistant streaming", () => { it("keeps code fences intact", () => { @@ -24,3 +34,54 @@ it("holds a streamed section heading until its content has a boundary", () => { rest: "Next", }); }); + +// Streams one 500-character paragraph per event, each past the 400 ms gate, +// and returns the snapshot lengths the filter passes on to be stored. +function streamParagraphs(paragraphs: number): { stored: number[]; finalLength: number } { + const filter = makeAssistantStreamingFilter("paragraph"); + const now = DateTime.makeUnsafe("2026-10-24T12:00:00.000Z"); + const item = { + id: TurnItemId.make("reasoning"), + threadId: ThreadId.make("thread"), + runId: RunId.make("run"), + nodeId: NodeId.make("node"), + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 0, + status: "running", + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + type: "reasoning", + text: "", + streaming: true, + } satisfies OrchestrationV2TurnItem; + const stored: number[] = []; + let text = ""; + for (let index = 0; index < paragraphs; index++) { + text += `${"x".repeat(498)}\n\n`; + const event = { + type: "turn_item.updated", + driver: ProviderDriverKind.make("opencode"), + turnItem: { ...item, text }, + } satisfies ProviderAdapter.ProviderAdapterV2Event; + const passed = filter(event, index * 400); + if (passed?.type === "turn_item.updated" && passed.turnItem.type === "reasoning") { + stored.push(passed.turnItem.text.length); + } + } + return { stored, finalLength: text.length }; +} + +it("keeps stored snapshots of a long streamed item linear in its length", () => { + const { stored, finalLength } = streamParagraphs(200); + const storedBytes = stored.reduce((total, length) => total + length, 0); + expect(storedBytes).toBeLessThan(finalLength * 8); +}); + +it("still passes every paragraph of a short streamed item", () => { + expect(streamParagraphs(8).stored).toEqual([500, 1000, 1500, 2000, 2500, 3000, 3500, 4000]); +}); diff --git a/apps/server/src/orchestration-v2/assistantStreaming.ts b/apps/server/src/orchestration-v2/assistantStreaming.ts index ee14f363d9af4..3979f7dee677d 100644 --- a/apps/server/src/orchestration-v2/assistantStreaming.ts +++ b/apps/server/src/orchestration-v2/assistantStreaming.ts @@ -81,6 +81,14 @@ export function splitBufferedAssistantText(text: string): { ready: string; rest: return { ready: text.slice(0, boundary), rest: text.slice(boundary) }; } +// Each passed snapshot is stored as a durable event holding the full text, so +// a long item streamed paragraph by paragraph would store bytes growing with +// the square of its length. Past this length, a snapshot passes only once its +// text is a quarter longer than the last one, keeping the stored total within +// a few times the final text. +const SNAPSHOT_GROWTH_START_LENGTH = 8_000; +const SNAPSHOT_GROWTH_FACTOR = 1.25; + export function makeAssistantStreamingFilter(mode: ResponseStreamingMode) { const delivered = new Map(); return ( @@ -111,6 +119,12 @@ export function makeAssistantStreamingFilter(mode: ResponseStreamingMode) { if (previous && now - previous.at < 400) return null; const { ready } = splitBufferedAssistantText(message.text); if (!ready || ready === previous?.text) return null; + if ( + previous && + previous.text.length >= SNAPSHOT_GROWTH_START_LENGTH && + ready.length < previous.text.length * SNAPSHOT_GROWTH_FACTOR + ) + return null; delivered.set(key, { text: ready, at: now }); if (event.type === "message.updated") return { ...event, message: { ...event.message, text: ready } }; From 529787d2f810cc0f6cf5e2a8a1e48ee56d8f3e0f Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sat, 24 Oct 2026 21:57:43 -0500 Subject: [PATCH 2/3] fix(server): cut the snapshot growth comment to four lines Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/assistantStreaming.ts | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/assistantStreaming.ts b/apps/server/src/orchestration-v2/assistantStreaming.ts index 3979f7dee677d..04452c7ddd4e9 100644 --- a/apps/server/src/orchestration-v2/assistantStreaming.ts +++ b/apps/server/src/orchestration-v2/assistantStreaming.ts @@ -81,11 +81,10 @@ export function splitBufferedAssistantText(text: string): { ready: string; rest: return { ready: text.slice(0, boundary), rest: text.slice(boundary) }; } -// Each passed snapshot is stored as a durable event holding the full text, so -// a long item streamed paragraph by paragraph would store bytes growing with -// the square of its length. Past this length, a snapshot passes only once its -// text is a quarter longer than the last one, keeping the stored total within -// a few times the final text. +// Each passed snapshot is a durable event holding the full text, so paragraph +// streaming would store bytes growing with the square of the length. Past this +// length a snapshot passes only once its text is a quarter longer than the +// last, keeping the stored total within a few times the final text. const SNAPSHOT_GROWTH_START_LENGTH = 8_000; const SNAPSHOT_GROWTH_FACTOR = 1.25; From 62bb46d961929ebb14da3f2d036897b31b254029 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sat, 24 Oct 2026 22:09:13 -0500 Subject: [PATCH 3/3] fix(server): long streamed replacements are no longer held by the growth check The snapshot growth check assumed every new snapshot appends to the last one. A provider replacement past 8,000 characters that was shorter or different stayed hidden until it reached 1.25 times the stale snapshot. The check now applies only when the new text extends the delivered text; a replacement passes under the existing 400 ms rule. Co-Authored-By: Claude Opus 5.5 --- .../assistantStreaming.test.ts | 41 ++++++++++++------- .../orchestration-v2/assistantStreaming.ts | 7 ++-- 2 files changed, 31 insertions(+), 17 deletions(-) diff --git a/apps/server/src/orchestration-v2/assistantStreaming.test.ts b/apps/server/src/orchestration-v2/assistantStreaming.test.ts index e8b8cd6d20abb..ef991db0c6db8 100644 --- a/apps/server/src/orchestration-v2/assistantStreaming.test.ts +++ b/apps/server/src/orchestration-v2/assistantStreaming.test.ts @@ -35,10 +35,7 @@ it("holds a streamed section heading until its content has a boundary", () => { }); }); -// Streams one 500-character paragraph per event, each past the 400 ms gate, -// and returns the snapshot lengths the filter passes on to be stored. -function streamParagraphs(paragraphs: number): { stored: number[]; finalLength: number } { - const filter = makeAssistantStreamingFilter("paragraph"); +function reasoningUpdate(text: string) { const now = DateTime.makeUnsafe("2026-10-24T12:00:00.000Z"); const item = { id: TurnItemId.make("reasoning"), @@ -56,22 +53,32 @@ function streamParagraphs(paragraphs: number): { stored: number[]; finalLength: completedAt: null, updatedAt: now, type: "reasoning", - text: "", + text, streaming: true, } satisfies OrchestrationV2TurnItem; + return { + type: "turn_item.updated", + driver: ProviderDriverKind.make("opencode"), + turnItem: item, + } satisfies ProviderAdapter.ProviderAdapterV2Event; +} + +function passedText(passed: ProviderAdapter.ProviderAdapterV2Event | null) { + return passed?.type === "turn_item.updated" && passed.turnItem.type === "reasoning" + ? passed.turnItem.text + : null; +} + +// Streams one 500-character paragraph per event, each past the 400 ms gate, +// and returns the snapshot lengths the filter passes on to be stored. +function streamParagraphs(paragraphs: number): { stored: number[]; finalLength: number } { + const filter = makeAssistantStreamingFilter("paragraph"); const stored: number[] = []; let text = ""; for (let index = 0; index < paragraphs; index++) { text += `${"x".repeat(498)}\n\n`; - const event = { - type: "turn_item.updated", - driver: ProviderDriverKind.make("opencode"), - turnItem: { ...item, text }, - } satisfies ProviderAdapter.ProviderAdapterV2Event; - const passed = filter(event, index * 400); - if (passed?.type === "turn_item.updated" && passed.turnItem.type === "reasoning") { - stored.push(passed.turnItem.text.length); - } + const passed = passedText(filter(reasoningUpdate(text), index * 400)); + if (passed !== null) stored.push(passed.length); } return { stored, finalLength: text.length }; } @@ -85,3 +92,9 @@ it("keeps stored snapshots of a long streamed item linear in its length", () => it("still passes every paragraph of a short streamed item", () => { expect(streamParagraphs(8).stored).toEqual([500, 1000, 1500, 2000, 2500, 3000, 3500, 4000]); }); + +it("passes a long replacement that does not extend the delivered text", () => { + const filter = makeAssistantStreamingFilter("paragraph"); + filter(reasoningUpdate(`${"x".repeat(9_998)}\n\n`), 0); + expect(passedText(filter(reasoningUpdate(`${"y".repeat(8_998)}\n\n`), 400))?.length).toBe(9_000); +}); diff --git a/apps/server/src/orchestration-v2/assistantStreaming.ts b/apps/server/src/orchestration-v2/assistantStreaming.ts index 04452c7ddd4e9..de1bcd3fad23a 100644 --- a/apps/server/src/orchestration-v2/assistantStreaming.ts +++ b/apps/server/src/orchestration-v2/assistantStreaming.ts @@ -83,8 +83,8 @@ export function splitBufferedAssistantText(text: string): { ready: string; rest: // Each passed snapshot is a durable event holding the full text, so paragraph // streaming would store bytes growing with the square of the length. Past this -// length a snapshot passes only once its text is a quarter longer than the -// last, keeping the stored total within a few times the final text. +// length a snapshot that extends the last passes only once it is a quarter +// longer, keeping the stored total within a few times the final text. const SNAPSHOT_GROWTH_START_LENGTH = 8_000; const SNAPSHOT_GROWTH_FACTOR = 1.25; @@ -121,7 +121,8 @@ export function makeAssistantStreamingFilter(mode: ResponseStreamingMode) { if ( previous && previous.text.length >= SNAPSHOT_GROWTH_START_LENGTH && - ready.length < previous.text.length * SNAPSHOT_GROWTH_FACTOR + ready.length < previous.text.length * SNAPSHOT_GROWTH_FACTOR && + ready.startsWith(previous.text) ) return null; delivered.set(key, { text: ready, at: now });