diff --git a/apps/server/src/orchestration-v2/assistantStreaming.test.ts b/apps/server/src/orchestration-v2/assistantStreaming.test.ts index d71042312bfb1..ef991db0c6db8 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,67 @@ it("holds a streamed section heading until its content has a boundary", () => { rest: "Next", }); }); + +function reasoningUpdate(text: string) { + 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; + 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 passed = passedText(filter(reasoningUpdate(text), index * 400)); + if (passed !== null) stored.push(passed.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]); +}); + +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 ee14f363d9af4..de1bcd3fad23a 100644 --- a/apps/server/src/orchestration-v2/assistantStreaming.ts +++ b/apps/server/src/orchestration-v2/assistantStreaming.ts @@ -81,6 +81,13 @@ export function splitBufferedAssistantText(text: string): { ready: string; rest: return { ready: text.slice(0, boundary), rest: text.slice(boundary) }; } +// 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 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; + export function makeAssistantStreamingFilter(mode: ResponseStreamingMode) { const delivered = new Map(); return ( @@ -111,6 +118,13 @@ 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 && + ready.startsWith(previous.text) + ) + return null; delivered.set(key, { text: ready, at: now }); if (event.type === "message.updated") return { ...event, message: { ...event.message, text: ready } };