fix(sse): reject a low-overlap stream-recovery continuation instead of concatenating it raw (#11152)

Merged after sibling #11151 landed: streamRecovery.ts auto-merged byte-identical to the validated combined board; the test-file conflict (both PRs added suites at the same anchor) resolved keeping all 11 tests — #11151's four clean-stop cases plus this PR's three threshold cases, with the PR's updated partial-tail fixture for the pre-existing overlap test. Full chain green: 32/32 (wiring + continuation + toolcall regression). The documented 8-char overlap threshold ends the silent mid-word gluing. Thank you @maxmad64bis!

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;
 
 /**

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,

tests/unit/stream-continuation-wiring.test.ts

@@ -46,14 +46,16 @@ async function collectText(stream: ReadableStream<Uint8Array>): Promise<string>
 
 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,