fix(server): long streams no longer store a full copy per paragraph

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);
+});

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<string, { text: string; at: number }>();
   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 } };