diff --git a/open-sse/config/constants.ts b/open-sse/config/constants.ts index f1fa055c93..d99da4725b 100644 --- a/open-sse/config/constants.ts +++ b/open-sse/config/constants.ts @@ -355,6 +355,23 @@ export const STREAM_RECOVERY = { HOLDBACK_MS: 750, BUFFER_MAX_BYTES: 65536, EARLY_RETRY_MAX: 4, + /** + * Minimum character overlap `trimContinuationOverlap` must find between the + * already-emitted text and a mid-stream continuation for the continuation to be + * accepted as a real resume, rather than an unrelated restart the model produced after + * ignoring the assistant-prefill. + * + * This is a DOCUMENTED TRADE-OFF, not a solved distinction: a model that continues + * cleanly with fewer than this many echoed characters (a legitimate, even preferred, + * outcome — there was nothing to de-duplicate) is indistinguishable, from string data + * alone, from a model that silently restarted on an unrelated sentence. Both produce a + * low/zero overlap. Rejecting below this threshold trades some false-positive rejections + * of legitimate low-overlap continuations (bounded retry, then a clean close — no data + * loss beyond that retry) against not silently gluing two unrelated fragments into one + * corrupted, unrecoverable answer. It does not eliminate the residual false negative + * either (an accidental coincidence at or above this many characters is still accepted). + */ + MIN_CONTINUATION_OVERLAP_CHARS: 8, } as const; /** diff --git a/open-sse/services/streamRecovery.ts b/open-sse/services/streamRecovery.ts index 129039c26c..3a95e9a2a3 100644 --- a/open-sse/services/streamRecovery.ts +++ b/open-sse/services/streamRecovery.ts @@ -541,7 +541,24 @@ export function createRecoverableStream( } const scan = scanOpenAiSseText(raw); - const suffix = trimContinuationOverlap(emittedText, scan.text); + // A continuation whose overlap with what was already emitted falls below the documented + // threshold is treated as a suspected restart rather than a real resume — see + // STREAM_RECOVERY.MIN_CONTINUATION_OVERLAP_CHARS for the full trade-off rationale. This + // is a heuristic, not a proof: it deliberately trades some false-positive rejections of + // legitimate low-overlap continuations against never silently gluing two unrelated + // fragments into one corrupted message. + const overlapResult = trimContinuationOverlap(emittedText, scan.text); + const overlapChars = scan.text.length - overlapResult.length; + const isSuspectedRestart = + emittedText.length > 0 && + scan.text.length > 0 && + overlapChars < STREAM_RECOVERY.MIN_CONTINUATION_OVERLAP_CHARS; + if (isSuspectedRestart) { + if (await tryContinue(controller)) return true; + emitCleanTerminal(controller); + return true; + } + const suffix = overlapResult; if (suffix) { emit( controller, diff --git a/tests/unit/stream-continuation-wiring.test.ts b/tests/unit/stream-continuation-wiring.test.ts index 1081d6ea9b..36a6f445d1 100644 --- a/tests/unit/stream-continuation-wiring.test.ts +++ b/tests/unit/stream-continuation-wiring.test.ts @@ -46,14 +46,16 @@ async function collectText(stream: ReadableStream): Promise const ROLE = 'data: {"choices":[{"delta":{"role":"assistant"}}]}\n\n'; const content = (s: string) => `data: {"choices":[{"delta":{"content":${JSON.stringify(s)}}}]}\n\n`; + const reasoning = (s: string) => `data: {"choices":[{"delta":{"reasoning_content":${JSON.stringify(s)}}}]}\n\n`; const finishStopNoContent = 'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}\n\n'; const finishLengthNoContent = 'data: {"choices":[{"delta":{},"finish_reason":"length"}]}\n\n'; test("mid-stream continuation: stitches the suffix after a silent post-commit truncation", async () => { - // Commits on chunk 1, emits "Hello wor", then ends WITHOUT a terminal marker (silent cut). - const initial = streamFrom([ROLE, content("Hello wor")]); + // Commits on chunk 1, emits "Hello there world", then ends WITHOUT a terminal marker + // (silent cut). + const initial = streamFrom([ROLE, content("Hello there world")]); let finalizeCount = 0; let continueArg = ""; @@ -64,15 +66,22 @@ test("mid-stream continuation: stitches the suffix after a silent post-commit tr now: steppingClock(), continueStream: async (soFar: string) => { continueArg = soFar; - // The model re-emits a small overlap ("wor") which must be trimmed away. - return streamFrom([ROLE, content("world!"), "data: [DONE]\n\n"]); + // The model re-emits only a partial tail of what was already sent ("there world", + // 11 chars — above the 8-char threshold, but NOT the full emitted text, unlike a + // full-string overlap this stays a discriminating test of trimContinuationOverlap's + // partial-tail trim, not just its "accept everything" path) before continuing. + return streamFrom([ROLE, content("there world, nice to meet you!"), "data: [DONE]\n\n"]); }, }); const out = await collectText(stream); const scan = scanOpenAiSseText(out); - assert.equal(continueArg, "Hello wor", "continuation is prefilled with the text already sent"); - assert.equal(scan.text, "Hello world!", "client sees the full answer, overlap trimmed, exactly once"); + assert.equal(continueArg, "Hello there world", "continuation is prefilled with the text already sent"); + assert.equal( + scan.text, + "Hello there world, nice to meet you!", + "client sees the full answer, partial overlap trimmed, exactly once" + ); assert.equal(scan.terminal, true, "the recovered stream ends with a terminal marker"); assert.equal(finalizeCount, 1, "finalize runs exactly once"); }); @@ -84,7 +93,7 @@ test("mid-stream continuation: recovers a post-commit transport error too", asyn const stream = createRecoverableStream(initial, async () => null, { finalize: () => {}, now: steppingClock(), - continueStream: async () => streamFrom([content("answer done."), "data: [DONE]\n\n"]), + continueStream: async () => streamFrom([content("Partial answer done."), "data: [DONE]\n\n"]), }); const scan = scanOpenAiSseText(await collectText(stream)); assert.equal(scan.text, "Partial answer done."); @@ -120,6 +129,82 @@ test("tool-call in flight is never continued (would corrupt tool JSON)", async ( assert.equal(continued, false, "continuation must NOT fire once a tool call has started streaming"); }); +test("mid-stream continuation: a zero-overlap restart is rejected, never concatenated raw", async () => { + // Truncates silently after real, non-empty text — canContinue() fires. + const initial = streamFrom([ROLE, content("Tous les faits sont reunis")]); + let continuations = 0; + const stream = createRecoverableStream(initial, async () => null, { + finalize: () => {}, + now: steppingClock(), + maxContinuations: 1, + continueStream: async () => { + continuations += 1; + // The model ignores the assistant prefill and restarts on an unrelated sentence — + // zero characters of overlap with what was already emitted. + return streamFrom([ + content("Je complete le design - derniere verification"), + "data: [DONE]\n\n", + ]); + }, + }); + const out = await collectText(stream); + const scan = scanOpenAiSseText(out); + assert.equal( + scan.text, + "Tous les faits sont reunis", + "the unrelated restart must never be appended to the already-emitted text" + ); + assert.equal(scan.terminal, true, "closes cleanly instead of leaving the client hanging"); + assert.equal(continuations, 1, "bounded by maxContinuations — does not loop forever"); +}); + +test("mid-stream continuation: a nonzero overlap below the threshold is rejected too", async () => { + // Genuine 4-character overlap ("pret"), well under the 8-char threshold — this is the + // false-negative case a naive `overlapChars === 0` check would miss (a restart that + // happens to share a short accidental fragment with the emitted tail): must still be + // treated as a suspected restart, not accepted as a genuine resume. + const initial = streamFrom([ROLE, content("Le design est pret")]); + let continuations = 0; + const stream = createRecoverableStream(initial, async () => null, { + finalize: () => {}, + now: steppingClock(), + maxContinuations: 1, + continueStream: async () => { + continuations += 1; + // Shares only "pret" (4 chars) with the emitted tail, then diverges completely. + return streamFrom([content("pret a partir de zero"), "data: [DONE]\n\n"]); + }, + }); + const scan = scanOpenAiSseText(await collectText(stream)); + assert.equal( + scan.text, + "Le design est pret", + "a below-threshold (but nonzero) overlap must not be accepted as a real resume" + ); + assert.equal(continuations, 1); +}); + +test("mid-stream continuation: a real overlap at or above the threshold is still stitched correctly", async () => { + // Regression guard: the existing happy path (first test in this file, whose updated + // fixture re-emits the 11-char partial tail "there world") still passes below — this test + // adds an overlap AT the threshold boundary to prove Task 3's new check does not fire when + // it shouldn't. + const initial = streamFrom([ROLE, content("The answer to this question")]); + const stream = createRecoverableStream(initial, async () => null, { + finalize: () => {}, + now: steppingClock(), + continueStream: async () => + // "question" (8 chars) overlaps the tail of emittedText exactly at the threshold. + streamFrom([content("question is forty-two."), "data: [DONE]\n\n"]), + }); + const scan = scanOpenAiSseText(await collectText(stream)); + assert.equal( + scan.text, + "The answer to this question is forty-two.", + "an overlap meeting the threshold is trimmed and stitched, not rejected" + ); +}); + test("mid-stream continuation: a clean stop with reasoning-only output (no answer) triggers a continuation", async () => { const initial = streamFrom([ ROLE,