diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 581f1f6342c0..09f8653ff679 100644 --- a/open-sse/utils/stream.ts +++ b/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", diff --git a/open-sse/utils/streamHelpers.ts b/open-sse/utils/streamHelpers.ts index 1f4b5f919820..6f526be3838d 100644 --- a/open-sse/utils/streamHelpers.ts +++ b/open-sse/utils/streamHelpers.ts @@ -432,8 +432,8 @@ export function fixInvalidId(parsed: Record): 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; +} diff --git a/tests/unit/streamHelpers.test.ts b/tests/unit/streamHelpers.test.ts index e31b61b9068e..ba9f5f502c1c 100644 --- a/tests/unit/streamHelpers.test.ts +++ b/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); + }); +});