From 6d5133eb1394ae2a0925411b5e0aa40bb68bcacd Mon Sep 17 00:00:00 2001 From: Fahad Iftikhar Date: Thu, 20 Aug 2026 23:43:14 +0500 Subject: [PATCH] fix(streaming): detect truncated streams and complete promptly on [DONE] The shared streaming engine only ended a response on connection close or cancellation, ignoring OpenCode's data: [DONE] terminator. This left the request hanging until the idle timeout when the gateway kept the connection alive after [DONE] (user cancelled silently), and treated a connection that closed without [DONE] (proxy reset, upstream crash, truncated payload) as a successful empty response. - parseServerSentEvent now fires an onDone callback on data: [DONE] and no longer passes it to the extractor. - engine breaks the read loop as soon as [DONE] arrives (prompt completion instead of waiting on a lingering socket). - isStreamTruncated throws a clear OpenCodeRequestError when the stream ended with neither [DONE] nor a captured finish_reason after content was already received, instead of silently succeeding. - unit tests added in src/test/sse.test.ts. --- CHANGELOG.md | 2 + ...-20260820-stream-silent-stop-truncation.md | 82 +++++++++++++++++++ src/test/sse.test.ts | 80 ++++++++++++++++++ src/transports/engine.ts | 59 ++++++++++--- src/transports/sse.ts | 26 +++++- 5 files changed, 238 insertions(+), 11 deletions(-) create mode 100644 docs/issues/70-20260820-stream-silent-stop-truncation.md create mode 100644 src/test/sse.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 9846d42..4eb0df6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,8 @@ All notable changes to the **OpenCode Go BYOK Provider** extension are documente ### Fixed +- **`[Chat]` Streams no longer end silently when the gateway drops the connection mid-response.** A chat could appear to "stop working" with no error: the engine only ended a stream on connection close or user cancellation, so (a) a gateway that kept the connection alive after its `data: [DONE]` terminator left the request hanging until the 2-minute idle timeout (user manually cancelled → silent), and (b) a connection that closed _without_ `[DONE]` (proxy reset, upstream crash, truncated payload) was treated as a successful empty response. `parseServerSentEvent` (`src/transports/sse.ts`) now fires an `onDone` callback on `data: [DONE]` and the engine breaks the read loop immediately on it (prompt completion instead of waiting on a lingering socket); a new `isStreamTruncated` check throws a clear `OpenCodeRequestError` ("stopped sending data before the response was complete… try again / check your connection, VPN, or firewall") when the stream closed with neither `[DONE]` nor a captured `finish_reason` after content was already received. Unit tests added in `src/test/sse.test.ts`. + - **`[Thinking]` Sampling-level `repetition_penalty` for MiMo curbs infinite thinking loops (#36).** A `repetition_penalty: 1.2` is now applied unconditionally in every MiMo payload (regardless of reasoning effort) to suppress the degenerate token-repeat pattern that triggered the Go gateway's infinite-loop detector upstream. This is a third mitigation layer alongside the `budget_tokens` cap and suffix-repetition stream detection shipped earlier. Applies with reasoning off (where the loop fires less often) and on; does not trip `bodyRequestsThinking()` so the gateway workaround path is unaffected. - **DeepSeek / Mimo thinking content no longer leaks into the chat transcript.** `treatReasoningAsContent` was mis-detecting native-reasoning families as "no reasoning in body" and echoing their `reasoning_content` as plain chat text. The decision now comes from the provider strategy (always `false` for DeepSeek and Mimo), so chain-of-thought stays in the thinking panel. diff --git a/docs/issues/70-20260820-stream-silent-stop-truncation.md b/docs/issues/70-20260820-stream-silent-stop-truncation.md new file mode 100644 index 0000000..dc16ee7 --- /dev/null +++ b/docs/issues/70-20260820-stream-silent-stop-truncation.md @@ -0,0 +1,82 @@ +**Status:** ✅ Resolved +**Fix PR:** (this branch) + +# Stream no longer ends silently on a dropped/truncated connection + +**Topic:** chat / provider / transport / streaming / resilience +**Updated:** 2026-08-20 +**Tags:** #chat #provider #transport #streaming #bug #resilience + +--- + +## Problem + +During development the model would appear to "stop working" and the session would +end with no warning or error. Two distinct failure modes produce the same silent +symptom: + +1. **Hang-then-cancel.** The response generates fine, then nothing happens. The + user waits and eventually cancels manually (or the 2-minute stream-idle + timeout fires). No error is ever shown. +2. **Silent truncation.** The response cuts off partway through (or comes back + empty) and the session ends as if the request succeeded — no "request + failed", no retry prompt, just a partial/empty answer. + +Both look like "the model stopped" but are really transport-level stream +termination problems that the extension swallowed. + +## Root cause + +`provideLanguageModelChatResponse` → `runStream*()` → `streamOpenCodeResponse` +(shared engine in `src/transports/engine.ts`) reads the SSE body in a loop that +only exits on `reader.read()` returning `done` (connection closed) or +cancellation. It **ignored OpenCode's `data: [DONE]` stream terminator** +(confirmed in `tmp/opencode-dev/.../server/transport/ws.ts`, which enqueues +`data: [DONE]\n\n` and then closes the controller). Consequences: + +- If the gateway keeps the TCP/keep-alive connection open after `[DONE]` (or the + HTTP→SSE proxy does not forward the close promptly), the loop never sees + `done` and sits until the idle timeout → mode 1 (hang-then-cancel). +- If the connection is instead **dropped** before `[DONE]` (proxy reset, upstream + crash, truncated payload, VPN/firewall cut), `reader.read()` returns `done` + with no `[DONE]` and no `finish_reason`. The engine had no signal that this was + abnormal, so it completed "successfully" with whatever partial content arrived + → mode 2 (silent truncation). The extractors do capture `finish_reason` / + `stop_reason` from the final chunk, but a truncated stream never delivers that + final chunk. + +The previous `diag-empty-response` path only logged format mismatches; it did not +catch a stream that carried real content and then died. + +## Fix + +- **`src/transports/sse.ts`** — `parseServerSentEvent` gains an `onDone` callback, + invoked when a `data: [DONE]` line is seen. `[DONE]` is no longer passed to the + extractor (it was already skipped, but the callback makes the terminator + observable to the caller). +- **`src/transports/engine.ts`** — + - Tracks `streamFlags.sawDone` and **breaks the read loop as soon as `[DONE]` + arrives**, so completion is prompt instead of waiting on a possibly-lingering + connection (fixes mode 1). + - After the loop, calls the new `isStreamTruncated` helper: if the stream ended + with **no `[DONE]` AND no captured `finish_reason`** while we had already + extracted content and received bytes, it throws a clear + `OpenCodeRequestError` ("` stopped sending data before the response + was complete (the connection closed unexpectedly). Your message may be cut + off — try sending it again. If this keeps happening, check your connection, + VPN, or firewall.") instead of silently succeeding (fixes mode 2). The + partial content already streamed to VS Code stays visible; the error tells + the user to retry. +- **`src/transports/sse.ts`** — `isStreamTruncated` is a small pure helper + (exported) so the decision is unit-testable and not buried in the engine. +- **`src/test/sse.test.ts`** — unit tests for `[DONE]` handling (fires `onDone`, + skipped by the extractor, fires once across multiple events) and for + `isStreamTruncated` (flags a closed stream with content but no `[DONE]` / + `finish_reason`; does not flag a stream that saw `[DONE]`, captured a + `finish_reason`, was empty, or carried no bytes). + +## Verification + +- `npm run lint` (editorconfig, eslint, markdown, prettier, shellcheck, + typecheck, unit tests) — green. +- `src/test/sse.test.ts` — new tests pass. diff --git a/src/test/sse.test.ts b/src/test/sse.test.ts new file mode 100644 index 0000000..3917d85 --- /dev/null +++ b/src/test/sse.test.ts @@ -0,0 +1,80 @@ +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; +import type * as vscode from "vscode"; +import { parseServerSentEvent, isStreamTruncated } from "../transports/sse.js"; + +// Recording extractor factory: returns no parts but counts invocations so we can +// assert [DONE] is skipped (extractParts must NOT be called for the [DONE] line). +const makeExtractor = (counter: { calls: number }) => (): vscode.LanguageModelResponsePart[] => { + counter.calls += 1; + return []; +}; + +describe("parseServerSentEvent — [DONE] handling", () => { + it("invokes onDone and skips extractParts for [DONE]", () => { + let doneCalls = 0; + const counter = { calls: 0 }; + parseServerSentEvent( + 'data: {"content":"hello"}\n\ndata: [DONE]\n\n', + makeExtractor(counter), + () => {}, + () => { + doneCalls += 1; + }, + ); + assert.equal(doneCalls, 1); + assert.equal(counter.calls, 1); + }); + + it("does not invoke onDone when [DONE] is absent", () => { + let doneCalls = 0; + const counter = { calls: 0 }; + parseServerSentEvent( + 'data: {"content":"hi"}\n\n', + makeExtractor(counter), + () => {}, + () => { + doneCalls += 1; + }, + ); + assert.equal(doneCalls, 0); + assert.equal(counter.calls, 1); + }); + + it("invokes onDone once even when [DONE] appears among multiple events", () => { + let doneCalls = 0; + const counter = { calls: 0 }; + parseServerSentEvent( + 'data: {"content":"a"}\n\ndata: {"content":"b"}\n\ndata: [DONE]\n\n', + makeExtractor(counter), + () => {}, + () => { + doneCalls += 1; + }, + ); + assert.equal(doneCalls, 1); + assert.equal(counter.calls, 2); + }); +}); + +describe("isStreamTruncated", () => { + it("flags a closed stream that carried content but no [DONE]/finish_reason", () => { + assert.equal(isStreamTruncated({ sawDone: false, finishReason: undefined, extractedPartCount: 5, totalBytes: 100 }), true); + }); + + it("does not flag a stream that saw [DONE]", () => { + assert.equal(isStreamTruncated({ sawDone: true, finishReason: undefined, extractedPartCount: 5, totalBytes: 100 }), false); + }); + + it("does not flag a stream that captured a finish_reason", () => { + assert.equal(isStreamTruncated({ sawDone: false, finishReason: "stop", extractedPartCount: 5, totalBytes: 100 }), false); + }); + + it("does not flag an empty stream (no content extracted)", () => { + assert.equal(isStreamTruncated({ sawDone: false, finishReason: undefined, extractedPartCount: 0, totalBytes: 100 }), false); + }); + + it("does not flag a stream with no bytes", () => { + assert.equal(isStreamTruncated({ sawDone: false, finishReason: undefined, extractedPartCount: 5, totalBytes: 0 }), false); + }); +}); diff --git a/src/transports/engine.ts b/src/transports/engine.ts index 9a45d19..df244a0 100644 --- a/src/transports/engine.ts +++ b/src/transports/engine.ts @@ -22,7 +22,7 @@ import { } from "../contextWindowHookBridge"; import { formatUsageLogLine } from "../usage/usage"; import { getErrorMessage, sleepWithCancellation } from "../utils"; -import { parseServerSentEvent } from "./sse"; +import { parseServerSentEvent, isStreamTruncated } from "./sse"; import { reportProgressPart, type RequestUsageSummary, type StreamOpenCodeResponseOptions } from "./streamParts"; import type { TransportRequestSummary } from "../core/transport"; import { updateRequestUsageSummary } from "./extract"; @@ -301,9 +301,13 @@ export async function streamOpenCodeResponse(options: StreamOpenCodeResponseOpti // format mismatches between gateway output and our extractor (issue #93). const rawSseData: unknown[] = []; let extractedPartCount = 0; + // Whether we received OpenCode's `data: [DONE]` stream-terminator. A + // successful stream always sends it; its absence at connection close + // signals a truncated/aborted response (see isStreamTruncated below). + const streamFlags: { sawDone: boolean } = { sawDone: false }; resetStreamIdleTimeout(); - while (!options.token.isCancellationRequested) { + while (!options.token.isCancellationRequested && !streamFlags.sawDone) { const { value, done } = await reader.read(); if (done) { break; @@ -327,10 +331,17 @@ export async function streamOpenCodeResponse(options: StreamOpenCodeResponseOpti if (options.debugReasoning && options.output && event.trim()) { options.output.appendLine(`[sse] ${truncateForLog(event)}`); } - for (const part of parseServerSentEvent(event, options.extractStreamParts, (data) => { - updateRequestUsageSummary(usageSummary, data); - rawSseData.push(data); - })) { + for (const part of parseServerSentEvent( + event, + options.extractStreamParts, + (data) => { + updateRequestUsageSummary(usageSummary, data); + rawSseData.push(data); + }, + () => { + streamFlags.sawDone = true; + }, + )) { extractedPartCount += 1; reportProgressPart(localRequestId, options.progress, part); } @@ -341,10 +352,17 @@ export async function streamOpenCodeResponse(options: StreamOpenCodeResponseOpti if (options.debugReasoning && options.output) { options.output.appendLine(`[sse-tail] ${truncateForLog(buffer)}`); } - for (const part of parseServerSentEvent(buffer, options.extractStreamParts, (data) => { - updateRequestUsageSummary(usageSummary, data); - rawSseData.push(data); - })) { + for (const part of parseServerSentEvent( + buffer, + options.extractStreamParts, + (data) => { + updateRequestUsageSummary(usageSummary, data); + rawSseData.push(data); + }, + () => { + streamFlags.sawDone = true; + }, + )) { extractedPartCount += 1; reportProgressPart(localRequestId, options.progress, part); } @@ -369,6 +387,27 @@ export async function streamOpenCodeResponse(options: StreamOpenCodeResponseOpti } } + // Detect abnormal stream termination. OpenCode always terminates a + // successful stream with a `data: [DONE]` sentinel, and the extractors + // capture a `finish_reason`/`stop_reason` from the final chunk. A stream + // that ends (connection closed) WITHOUT either signal while we had already + // extracted content was truncated or aborted (gateway dropped the + // connection, proxy reset, upstream crash). Previously this was treated as + // a silent success, leaving the user with a partial/empty response and no + // indication of what happened — the "model stopped working / session ended + // with no warning" bug. + if (isStreamTruncated({ sawDone: streamFlags.sawDone, finishReason: usageSummary.finishReason, extractedPartCount, totalBytes })) { + const requestError = new OpenCodeRequestError( + `${options.providerDisplayName} response stream ended before completion (no [DONE] or finish_reason after ${String(totalBytes)} bytes / ${String(totalEvents)} events).`, + `${options.providerDisplayName} stopped sending data before the response was complete (the connection closed unexpectedly). Your message may be cut off — try sending it again. If this keeps happening, check your connection, VPN, or firewall.`, + ); + emitSummary(totalBytes, totalEvents, { + errorMessage: requestError.message, + rateLimitSummary, + }); + throw requestError; + } + emitSummary(totalBytes, totalEvents, { rateLimitSummary }); } catch (error) { if (abortReason === "cancelled") { diff --git a/src/transports/sse.ts b/src/transports/sse.ts index caf93ca..a2cd030 100644 --- a/src/transports/sse.ts +++ b/src/transports/sse.ts @@ -5,6 +5,7 @@ export function parseServerSentEvent( event: string, extractParts: (data: unknown) => vscode.LanguageModelResponsePart[], onData?: (data: unknown) => void, + onDone?: () => void, ): vscode.LanguageModelResponsePart[] { const lines = event .split(/\r?\n/) @@ -14,7 +15,11 @@ export function parseServerSentEvent( const parts: vscode.LanguageModelResponsePart[] = []; for (const line of lines) { - if (!line || line === "[DONE]") { + if (!line) { + continue; + } + if (line === "[DONE]") { + onDone?.(); continue; } @@ -29,3 +34,22 @@ export function parseServerSentEvent( return parts; } + +/** + * Pure: decide whether an SSE stream ended abnormally (truncated/aborted). + * + * A successful OpenCode stream always terminates with a `data: [DONE]` + * sentinel, and the extractors capture a `finish_reason`/`stop_reason` from the + * final chunk. If the connection closed (`done`) without EITHER signal while we + * had already extracted content, the stream was cut off mid-response (gateway + * dropped the connection, proxy reset, upstream crash) and must not be treated + * as a silent success. + */ +export function isStreamTruncated(params: { + sawDone: boolean; + finishReason: string | undefined; + extractedPartCount: number; + totalBytes: number; +}): boolean { + return !params.sawDone && params.finishReason === undefined && params.extractedPartCount > 0 && params.totalBytes > 0; +}