perf: Date.now hoist, hasActiveDeltaValue hoist, buffer.split guard in SSE stream

open-sse/utils/stream.ts

@@ -18,6 +18,8 @@ import {
   fixInvalidId,
   formatSSE,
   unwrapGeminiChunk,
+  appendBoundedText,
+  hasActiveDeltaValue,
 } from "./streamHelpers.ts";
 import { calculateCost } from "@/lib/usage/costCalculator";
 import { buildOmniRouteSseMetadataComment } from "@/domain/omnirouteResponseMeta";
@@ -177,15 +179,6 @@ function asRecord(value: unknown): JsonRecord {
   return value && typeof value === "object" && !Array.isArray(value) ? (value as JsonRecord) : {};
 }
 
-const STREAM_SUMMARY_TEXT_LIMIT = 64 * 1024;
-
-function appendBoundedText(current: string, next: string): string {
-  if (!next) return current;
-  const combined = current + next;
-  if (combined.length <= STREAM_SUMMARY_TEXT_LIMIT) return combined;
-  return combined.slice(-STREAM_SUMMARY_TEXT_LIMIT);
-}
-
 function parseTextualToolCallFromContent(text: unknown): { name: string; args: unknown } | null {
   const candidate = parseTextualToolCallCandidate(text);
   return candidate?.kind === "complete" ? { name: candidate.name, args: candidate.args } : null;
@@ -1153,13 +1146,14 @@ export function createSSEStream(options: StreamOptions = {}) {
 
       transform(chunk, controller) {
         if (streamTimedOut) return;
-        lastChunkTime = Date.now();
+        const now = Date.now();
+        lastChunkTime = now;
         const text = decoder.decode(chunk, { stream: true });
         buffer += text;
         reqLogger?.appendProviderChunk?.(text);
-
-        const lines = buffer.split("\n");
-        buffer = lines.pop() || "";
+        const nlIdx = buffer.lastIndexOf("\n");
+        const lines = nlIdx >= 0 ? buffer.slice(0, nlIdx).split("\n") : [];
+        if (nlIdx >= 0) buffer = buffer.slice(nlIdx + 1);
 
         for (const line of multilineSseDataLineNormalizer.normalize(lines)) {
           const trimmed = line.trim();
@@ -1262,15 +1256,6 @@ export function createSSEStream(options: StreamOptions = {}) {
                 // bootstrap chunk (assistant role + empty content) before emitting proper
                 // `response.*` events. That chunk is invalid on /v1/responses and breaks strict
                 // clients like OpenCode, so drop it only for Responses-native consumers.
-                const hasActiveDeltaValue = (value: unknown): boolean => {
-                  if (typeof value === "string") return value.length > 0;
-                  if (Array.isArray(value))
-                    return value.some((entry) => hasActiveDeltaValue(entry));
-                  if (value && typeof value === "object") {
-                    return Object.values(value).some((entry) => hasActiveDeltaValue(entry));
-                  }
-                  return value !== null && value !== undefined;
-                };
 
                 const isEmptyAssistantBootstrapChunkForResponsesClient =
                   clientExpectsResponsesStream &&
@@ -1738,7 +1723,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                   // T18: Track if we saw tool calls & accumulate for call log
                   if (delta?.tool_calls && delta.tool_calls.length > 0) {
                     passthroughHasToolCalls = true;
-                    lastToolCallChunkTime = Date.now();
+                    lastToolCallChunkTime = now;
                     for (const tc of delta.tool_calls) {
                       // Note: sanitizeStreamingChunk above already coerces non-string
                       // tool_call IDs, but this defensive check catches edge cases
@@ -1787,7 +1772,6 @@ export function createSSEStream(options: StreamOptions = {}) {
                       const lastChunkTs = lastToolCallChunkTime;
                       if (toolTs || lastChunkTs) {
                         contentAfterToolSeen = true;
-                        const now = Date.now();
                         try {
                           recordToolLatency(
                             provider || "unknown",
@@ -1824,7 +1808,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                   const isFinishChunk = parsed.choices?.[0]?.finish_reason;
 
                   if (isFinishChunk && passthroughHasToolCalls) {
-                    toolFinishTime = Date.now();
+                    toolFinishTime = now;
                     try {
                       markToolFinish(sessionId);
                     } catch {}
@@ -1949,10 +1933,10 @@ export function createSSEStream(options: StreamOptions = {}) {
           }
 
           if (parsed.choices?.[0]?.delta?.tool_calls) {
-            lastToolCallChunkTime = Date.now();
+            lastToolCallChunkTime = now;
           }
           if (parsed.choices?.[0]?.finish_reason === "tool_calls") {
-            toolFinishTime = Date.now();
+            toolFinishTime = now;
             try {
               markToolFinish(sessionId);
             } catch {}
@@ -2063,7 +2047,6 @@ export function createSSEStream(options: StreamOptions = {}) {
             const lastChunkTs = lastToolCallChunkTime;
             if (toolTs || lastChunkTs) {
               contentAfterToolSeen = true;
-              const now = Date.now();
               try {
                 recordToolLatency(
                   provider || "unknown",

open-sse/utils/streamHelpers.ts

@@ -432,8 +432,8 @@ export function fixInvalidId(parsed: Record<string, unknown>): boolean {
 // Remove null perf_metrics from usage (common across formats)
 function cleanPerfMetrics(data: unknown): unknown {
   if (isRecord(data) && isRecord(data.usage) && data.usage.perf_metrics === null) {
-    const { perf_metrics, ...usageWithoutPerf } = data.usage;
-    return { ...data, usage: usageWithoutPerf };
+    // Mutate in-place to avoid spread copy per chunk — data is ephemeral, used only for serialization.
+    delete data.usage.perf_metrics;
   }
   return data;
 }
@@ -458,3 +458,30 @@ export function formatSSE(data: unknown, sourceFormat: string): string {
 
   return `data: ${JSON.stringify(data)}\n\n`;
 }
+
+const STREAM_SUMMARY_TEXT_LIMIT = 64 * 1024;
+
+// Bounded accumulator for streamed content/reasoning text — caps memory on long streams
+// by keeping only the tail once the limit is reached, instead of growing unbounded.
+export function appendBoundedText(current: string, next: string): string {
+  if (!next) return current;
+  // Avoid allocating `current + next` when already at/above limit — slide the window instead.
+  if (current.length >= STREAM_SUMMARY_TEXT_LIMIT) {
+    const keep = STREAM_SUMMARY_TEXT_LIMIT - next.length;
+    if (keep <= 0) return next.slice(-STREAM_SUMMARY_TEXT_LIMIT);
+    return current.slice(-keep) + next;
+  }
+  const combined = current + next;
+  if (combined.length <= STREAM_SUMMARY_TEXT_LIMIT) return combined;
+  return combined.slice(-STREAM_SUMMARY_TEXT_LIMIT);
+}
+
+/** Per-chunk recursive check for meaningful delta content. Hoisted to avoid closure re-allocation in hot-path. */
+export function hasActiveDeltaValue(value: unknown): boolean {
+  if (typeof value === "string") return value.length > 0;
+  if (Array.isArray(value)) return value.some((entry) => hasActiveDeltaValue(entry));
+  if (value && typeof value === "object") {
+    return Object.values(value).some((entry) => hasActiveDeltaValue(entry));
+  }
+  return value !== null && value !== undefined;
+}

tests/unit/streamHelpers.test.ts

@@ -1,6 +1,11 @@
 import { describe, it } from "node:test";
 import assert from "node:assert";
-import { hasValuableContent, unwrapGeminiChunk } from "../../open-sse/utils/streamHelpers.ts";
+import {
+  hasValuableContent,
+  unwrapGeminiChunk,
+  appendBoundedText,
+  hasActiveDeltaValue,
+} from "../../open-sse/utils/streamHelpers.ts";
 import { FORMATS } from "../../open-sse/translator/formats.ts";
 
 describe("hasValuableContent", () => {
@@ -126,3 +131,97 @@ describe("unwrapGeminiChunk", () => {
     assert.equal(result.candidates[0].content.parts[0].text, "outer");
   });
 });
+
+const STREAM_SUMMARY_TEXT_LIMIT = 64 * 1024;
+
+describe("appendBoundedText", () => {
+  it("returns current unchanged when next is empty", () => {
+    assert.strictEqual(appendBoundedText("abc", ""), "abc");
+  });
+
+  it("concatenates normally while under the limit", () => {
+    assert.strictEqual(appendBoundedText("abc", "def"), "abcdef");
+  });
+
+  it("keeps the tail once the combined length exceeds the limit", () => {
+    const current = "a".repeat(STREAM_SUMMARY_TEXT_LIMIT - 1);
+    const result = appendBoundedText(current, "bb");
+    assert.strictEqual(result.length, STREAM_SUMMARY_TEXT_LIMIT);
+    assert.ok(result.endsWith("bb"), "must retain the newest text");
+  });
+
+  it("slides the window when current is already at the limit", () => {
+    const current = "a".repeat(STREAM_SUMMARY_TEXT_LIMIT);
+    const result = appendBoundedText(current, "xyz");
+    assert.strictEqual(result.length, STREAM_SUMMARY_TEXT_LIMIT);
+    assert.ok(result.endsWith("xyz"), "must retain the newest text");
+  });
+
+  // Regression: `keep` is 0 when next.length === LIMIT. `current.slice(-0)` is
+  // `slice(0)` — the WHOLE string — so a naive impl returns current + next and
+  // blows past the bound. Must return only the tail of next.
+  it("stays bounded when next is exactly the limit (slice(-0) trap)", () => {
+    const current = "a".repeat(STREAM_SUMMARY_TEXT_LIMIT);
+    const next = "b".repeat(STREAM_SUMMARY_TEXT_LIMIT);
+    const result = appendBoundedText(current, next);
+    assert.strictEqual(result.length, STREAM_SUMMARY_TEXT_LIMIT);
+    assert.strictEqual(result, next, "must be next only — no 'a' may survive");
+    assert.ok(!result.includes("a"), "must not leak the old buffer");
+  });
+
+  it("stays bounded when next is larger than the limit", () => {
+    const current = "a".repeat(STREAM_SUMMARY_TEXT_LIMIT);
+    const next = "b".repeat(STREAM_SUMMARY_TEXT_LIMIT + 500);
+    const result = appendBoundedText(current, next);
+    assert.strictEqual(result.length, STREAM_SUMMARY_TEXT_LIMIT);
+    assert.ok(!result.includes("a"), "must not leak the old buffer");
+  });
+
+  it("never exceeds the limit across repeated appends", () => {
+    let acc = "";
+    for (let i = 0; i < 40; i++) {
+      acc = appendBoundedText(acc, "z".repeat(4096));
+      assert.ok(acc.length <= STREAM_SUMMARY_TEXT_LIMIT, `overflow at iteration ${i}`);
+    }
+    assert.strictEqual(acc.length, STREAM_SUMMARY_TEXT_LIMIT);
+  });
+});
+
+describe("hasActiveDeltaValue", () => {
+  it("returns true for a non-empty string", () => {
+    assert.strictEqual(hasActiveDeltaValue("hi"), true);
+  });
+
+  it("returns false for an empty string", () => {
+    assert.strictEqual(hasActiveDeltaValue(""), false);
+  });
+
+  it("returns false for null and undefined", () => {
+    assert.strictEqual(hasActiveDeltaValue(null), false);
+    assert.strictEqual(hasActiveDeltaValue(undefined), false);
+  });
+
+  it("returns false for an empty array and an array of empty strings", () => {
+    assert.strictEqual(hasActiveDeltaValue([]), false);
+    assert.strictEqual(hasActiveDeltaValue(["", ""]), false);
+  });
+
+  it("returns true when any array entry is meaningful", () => {
+    assert.strictEqual(hasActiveDeltaValue(["", "x"]), true);
+  });
+
+  it("returns false for an empty object and an object of empty values", () => {
+    assert.strictEqual(hasActiveDeltaValue({}), false);
+    assert.strictEqual(hasActiveDeltaValue({ a: "", b: null }), false);
+  });
+
+  it("recurses into nested structures", () => {
+    assert.strictEqual(hasActiveDeltaValue({ a: { b: [{ c: "" }] } }), false);
+    assert.strictEqual(hasActiveDeltaValue({ a: { b: [{ c: "found" }] } }), true);
+  });
+
+  it("treats numbers and booleans as meaningful", () => {
+    assert.strictEqual(hasActiveDeltaValue(0), true);
+    assert.strictEqual(hasActiveDeltaValue(false), true);
+  });
+});