From f58e8bef6fb2fb9712a336006784e8457fb86ca4 Mon Sep 17 00:00:00 2001 From: Yao Lu Date: Mon, 24 Aug 2026 18:03:34 +0800 Subject: [PATCH] fix(opencode): close Muse Responses streams at completion --- open-sse/executors/opencode.ts | 89 +++++++++++++++---- open-sse/utils/proxyFetch.ts | 39 ++++---- ...proxyfetch-no-proxy-context-bypass.test.ts | 16 +++- .../opencode-muse-spark-min-output.test.ts | 57 ++++++++++-- ...-session-fingerprint-headers-10571.test.ts | 21 +++++ 5 files changed, 173 insertions(+), 49 deletions(-) diff --git a/open-sse/executors/opencode.ts b/open-sse/executors/opencode.ts index fdb31fd132..c83b08ae86 100644 --- a/open-sse/executors/opencode.ts +++ b/open-sse/executors/opencode.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import { BaseExecutor, type ExecuteInput, @@ -10,7 +11,7 @@ import { injectReasoningContentForThinkingModel, isThinkingMessageModel, } from "../utils/reasoningContentInjector.ts"; -import { runWithProxyContext } from "../utils/proxyFetch.ts"; +import { runWithDirectFetchContext, runWithProxyContext } from "../utils/proxyFetch.ts"; import { forwardOpencodeClientHeaders } from "../utils/opencodeHeaders.ts"; import { type AccountProxyConfig, @@ -245,6 +246,17 @@ export function createMuseSparkStreamFinishNormalizer( }; } +function isResponsesTerminalLine(line: string): boolean { + const trimmed = line.trim(); + if (!trimmed.startsWith("data:")) return false; + try { + const payload = JSON.parse(trimmed.slice(5).trim()) as Record; + return payload.type === "response.completed"; + } catch { + return false; + } +} + export class OpencodeExecutor extends BaseExecutor { /** Delegates to `isPremiumOpencodeModel`. Exported for testability. */ static isPremiumModel(model: string, provider: string): boolean { @@ -384,24 +396,51 @@ export class OpencodeExecutor extends BaseExecutor { const encoder = new TextEncoder(); let buffer = ""; const reader = response.body.getReader(); + let closed = false; const stream = new ReadableStream({ - async pull(controller) { + async start(controller) { try { - const { done, value } = await reader.read(); - if (done) { - if (buffer.length > 0) controller.enqueue(encoder.encode(normalizer(buffer))); - controller.close(); - return; + while (!closed) { + const { done, value } = await reader.read(); + if (done) { + buffer += decoder.decode(); + if (buffer.length > 0 && !closed) { + controller.enqueue(encoder.encode(normalizer(buffer))); + } + if (!closed) { + closed = true; + controller.close(); + } + return; + } + + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split("\n"); + buffer = lines.pop() ?? ""; + for (const line of lines) { + const normalized = normalizer(line); + controller.enqueue(encoder.encode(normalized + "\n")); + if (isResponsesTerminalLine(line)) { + // OpenCode Zen sends a ping after response.completed and may keep + // the HTTP connection alive. The Responses terminal event is + // authoritative; do not let those post-completion pings hold Chat + // Completions open. + closed = true; + void reader.cancel().catch(() => undefined); + controller.close(); + return; + } + } } - buffer += decoder.decode(value, { stream: true }); - const lines = buffer.split("\n"); - buffer = lines.pop() ?? ""; - for (const line of lines) controller.enqueue(encoder.encode(normalizer(line) + "\n")); } catch (err) { - controller.error(err); + if (!closed) { + closed = true; + controller.error(err); + } } }, cancel(reason) { + closed = true; reader.cancel(reason).catch(() => undefined); }, }); @@ -450,7 +489,10 @@ export class OpencodeExecutor extends BaseExecutor { // 200s ("Provider returned empty content"). Raise tiny budgets to the // floor before dispatch (see MUSE_SPARK_MIN_OUTPUT_TOKENS). if (input.body && typeof input.body === "object" && !Array.isArray(input.body)) { - applyMuseSparkMinOutputTokens(String(input.model ?? ""), input.body as Record); + applyMuseSparkMinOutputTokens( + String(input.model ?? ""), + input.body as Record + ); } this.syncAccountsFromCredentials(input.credentials); @@ -463,7 +505,9 @@ export class OpencodeExecutor extends BaseExecutor { // else passes untouched: this path deliberately preserves BaseExecutor's // intra-URL 429 retries (no skipUpstreamRetry here). if (this.accounts.length === 1 && !hasProxies) { - const single = (await super.execute(input)) as HttpExecuteResult; + const single = (await runWithDirectFetchContext(() => + super.execute(input) + )) as HttpExecuteResult; if (single.response.status === 400) { let bodyText: string | null = null; try { @@ -630,10 +674,7 @@ export class OpencodeExecutor extends BaseExecutor { } // All accounts returned 429 (or errored) — surface the last response. - return this.normalizeMuseSparkResponse( - input, - lastResult ?? (await super.execute(input)) - ); + return this.normalizeMuseSparkResponse(input, lastResult ?? (await super.execute(input))); } finally { this._requestFormat = null; } @@ -735,6 +776,18 @@ export class OpencodeExecutor extends BaseExecutor { }); } + // Muse's Responses endpoint rejects the short conversation fingerprint used + // by the Chat endpoint in practice. Keep the workaround scoped to Muse. + if ( + this._requestFormat === "openai-responses" && + model.startsWith("muse-spark") && + !/^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test( + headers["x-opencode-session"] || "" + ) + ) { + headers["x-opencode-session"] = randomUUID(); + } + void model; return headers; diff --git a/open-sse/utils/proxyFetch.ts b/open-sse/utils/proxyFetch.ts index 4eedd2dad5..e1a32add91 100644 --- a/open-sse/utils/proxyFetch.ts +++ b/open-sse/utils/proxyFetch.ts @@ -351,10 +351,7 @@ function sanitizeTransportError( typeof source.code === "string" && /^[A-Z0-9_:-]{1,64}$/.test(source.code) ? source.code : fallbackCode; - if ( - typeof source.errorCode === "string" && - /^[a-zA-Z0-9_:-]{1,64}$/.test(source.errorCode) - ) { + if (typeof source.errorCode === "string" && /^[a-zA-Z0-9_:-]{1,64}$/.test(source.errorCode)) { sanitized.errorCode = source.errorCode; } if (typeof source.statusCode === "number" && Number.isFinite(source.statusCode)) { @@ -547,10 +544,7 @@ export function resolveProxyForRequest(targetUrl) { * Dependency-internal TimeoutError/AbortError values are transport failures and * retain the normal safe-method fallback behavior. */ -function isCallerAbort( - _error: unknown, - signal: AbortSignal | null | undefined -): boolean { +function isCallerAbort(_error: unknown, signal: AbortSignal | null | undefined): boolean { return signal?.aborted === true; } @@ -573,8 +567,7 @@ export async function runWithProxyContext( // sentinel must remain direct without being mistaken for a proxy config. const currentContext = proxyContext.getStore(); const inheritsDirect = currentContext === DIRECT_PROXY_CONTEXT && !proxyConfig; - const effectiveProxyConfig = - proxyConfig || (inheritsDirect ? null : currentContext) || null; + const effectiveProxyConfig = proxyConfig || (inheritsDirect ? null : currentContext) || null; const contextValue = inheritsDirect ? DIRECT_PROXY_CONTEXT : effectiveProxyConfig; const resolvedProxyUrl = effectiveProxyConfig ? proxyConfigToUrl(effectiveProxyConfig) : null; @@ -711,6 +704,11 @@ export async function runWithProxyContext( }); } +/** Run a request with an explicit direct-egress sentinel, bypassing proxy env/context lookup. */ +export function runWithDirectFetchContext(fn: () => T): T { + return proxyContext.run(DIRECT_PROXY_CONTEXT, fn); +} + /** * Like {@link runWithProxyContext}, but if the assigned proxy is unreachable or fails * its pre-checks the request can degrade to a DIRECT connection instead of throwing. @@ -732,6 +730,12 @@ async function patchedFetch( options: FetchWithDispatcherOptions = {}, deps: ProxyFetchDeps = {} ) { + // Explicit direct contexts must win even when a caller supplied a stale + // dispatcher. Native fetch preserves direct streaming semantics. + if (proxyContext.getStore() === DIRECT_PROXY_CONTEXT) { + return originalFetch(input, options); + } + if (options?.dispatcher) { // When a dispatcher is present, we MUST use the undici library fetch // to ensure version compatibility. Node 22 built-in fetch (undici v6) @@ -1133,9 +1137,7 @@ async function patchedFetch( ); const sanitized = sanitizeTransportError( error, - originalMsg - ? `Proxy request failed: ${originalMsg}` - : "Proxy request failed", + originalMsg ? `Proxy request failed: ${originalMsg}` : "Proxy request failed", "PROXY_REQUEST_FAILED" ); console.error( @@ -1190,8 +1192,7 @@ export async function runWithTlsTracking( providerOrIdentityOrFn: string | null | undefined | TlsTrackingIdentity | (() => T), maybeFn?: () => T ): Promise<{ result: Awaited; tlsFingerprintUsed: boolean }> { - const legacyFn = - typeof providerOrIdentityOrFn === "function" ? providerOrIdentityOrFn : maybeFn; + const legacyFn = typeof providerOrIdentityOrFn === "function" ? providerOrIdentityOrFn : maybeFn; if (typeof legacyFn !== "function") { throw new TypeError("runWithTlsTracking requires a callback function"); } @@ -1201,8 +1202,7 @@ export async function runWithTlsTracking( typeof providerOrIdentityOrFn !== "function" ? providerOrIdentityOrFn : { - provider: - typeof providerOrIdentityOrFn === "string" ? providerOrIdentityOrFn : undefined, + provider: typeof providerOrIdentityOrFn === "string" ? providerOrIdentityOrFn : undefined, }; const store: TlsFingerprintStore = { used: false, @@ -1214,10 +1214,7 @@ export async function runWithTlsTracking( } /** Check whether TLS fingerprint transport is enabled for this route identity. */ -export function isTlsFingerprintActive( - provider?: string | null, - proxied = false -): boolean { +export function isTlsFingerprintActive(provider?: string | null, proxied = false): boolean { return ( isTlsFingerprintEnabled() && activeTlsClient.available && diff --git a/tests/unit/9551-proxyfetch-no-proxy-context-bypass.test.ts b/tests/unit/9551-proxyfetch-no-proxy-context-bypass.test.ts index 04a3fd2b17..c7a6d72122 100644 --- a/tests/unit/9551-proxyfetch-no-proxy-context-bypass.test.ts +++ b/tests/unit/9551-proxyfetch-no-proxy-context-bypass.test.ts @@ -1,6 +1,10 @@ import test from "node:test"; import assert from "node:assert/strict"; -import { runWithProxyContext, resolveProxyForRequest } from "../../open-sse/utils/proxyFetch.ts"; +import { + runWithDirectFetchContext, + runWithProxyContext, + resolveProxyForRequest, +} from "../../open-sse/utils/proxyFetch.ts"; async function withEnv( overrides: Record, @@ -59,3 +63,13 @@ test("[9551] resolveProxyForRequest: context-proxy respects NO_PROXY=*", async ( } ); }); + +test("direct fetch context overrides an inherited proxy context", async () => { + await runWithProxyContext({ type: "http", host: "127.0.0.1", port: 7897 }, () => + runWithDirectFetchContext(() => { + const resolved = resolveProxyForRequest("https://api.commandcode.ai/alpha/generate"); + assert.equal(resolved.source, "direct"); + assert.equal(resolved.proxyUrl, null); + }) + ); +}); diff --git a/tests/unit/opencode-muse-spark-min-output.test.ts b/tests/unit/opencode-muse-spark-min-output.test.ts index 44b151c826..8424d36636 100644 --- a/tests/unit/opencode-muse-spark-min-output.test.ts +++ b/tests/unit/opencode-muse-spark-min-output.test.ts @@ -16,13 +16,10 @@ import test from "node:test"; import assert from "node:assert/strict"; -const { applyMuseSparkMinOutputTokens, MUSE_SPARK_MIN_OUTPUT_TOKENS } = await import( - "../../open-sse/executors/opencode.ts" -); -const { - normalizeMuseSparkFinishReason, - createMuseSparkStreamFinishNormalizer, -} = await import("../../open-sse/executors/opencode.ts"); +const { applyMuseSparkMinOutputTokens, MUSE_SPARK_MIN_OUTPUT_TOKENS } = + await import("../../open-sse/executors/opencode.ts"); +const { normalizeMuseSparkFinishReason, createMuseSparkStreamFinishNormalizer, OpencodeExecutor } = + await import("../../open-sse/executors/opencode.ts"); test("RED: muse-spark tiny max_tokens is raised to the floor", () => { const body: Record = { model: "x", max_tokens: 64, messages: [] }; @@ -97,8 +94,7 @@ test("RED: stream normalizer rewrites the finish frame after the usage frame", ( const usageLine = 'data: {"id":"r","object":"chat.completion.chunk","choices":[],"usage":{"completion_tokens":270}}'; assert.equal(norm(usageLine), usageLine, "usage frame itself must not change"); - const finishLine = - 'data: {"choices":[{"index":0,"delta":{},"finish_reason":"length"}]}'; + const finishLine = 'data: {"choices":[{"index":0,"delta":{},"finish_reason":"length"}]}'; const out = JSON.parse(norm(finishLine).slice(5).trim()); assert.equal(out.choices[0].finish_reason, "stop"); }); @@ -109,3 +105,46 @@ test("RED: stream normalizer passes through [DONE], comments and non-JSON lines" assert.equal(norm(": keepalive"), ": keepalive"); assert.equal(norm("data: not-json"), "data: not-json"); }); + +test("closes the Muse Responses stream at response.completed before post-completion pings", async () => { + const originalFetch = globalThis.fetch; + try { + globalThis.fetch = (async () => + new Response( + [ + "event: response.output_text.delta", + 'data: {"type":"response.output_text.delta","delta":"OK"}', + "event: response.completed", + 'data: {"type":"response.completed","response":{"usage":{"input_tokens":1,"output_tokens":1}}}', + "event: ping", + 'data: {"type":"ping"}', + "", + ].join("\n"), + { status: 200, headers: { "Content-Type": "text/event-stream" } } + )) as typeof fetch; + + const result = await new OpencodeExecutor("opencode").execute({ + model: "muse-spark-1.2-contributor-free", + body: { + model: "muse-spark-1.2-contributor-free", + max_output_tokens: 512, + stream: true, + }, + stream: true, + credentials: { + providerSpecificData: { + fingerprints: ["test-account-a", "test-account-b"], + accountProxies: [], + }, + }, + }); + const text = await Promise.race([ + result.response.text(), + new Promise((_, reject) => setTimeout(() => reject(new Error("stream hung")), 1000)), + ]); + assert.match(text, /response.completed/); + assert.doesNotMatch(text, /\"type\":\"ping\"/); + } finally { + globalThis.fetch = originalFetch; + } +}); diff --git a/tests/unit/opencode-session-fingerprint-headers-10571.test.ts b/tests/unit/opencode-session-fingerprint-headers-10571.test.ts index 692435daaa..f05da45300 100644 --- a/tests/unit/opencode-session-fingerprint-headers-10571.test.ts +++ b/tests/unit/opencode-session-fingerprint-headers-10571.test.ts @@ -160,6 +160,27 @@ test("OpencodeExecutor.buildHeaders derives a stable x-opencode-session from the assert.equal(headersFirst["x-opencode-session"], headersSecond["x-opencode-session"]); }); +test("Responses requests use a UUID x-opencode-session for Muse compatibility", () => { + const executor = new OpencodeExecutor("opencode"); + executor._requestFormat = "openai-responses"; + const headers = executor.buildHeaders( + null, + true, + null, + "muse-spark-1.2-contributor-free", + undefined, + { + model: "muse-spark-1.2-contributor-free", + input: [], + } + ); + assert.match( + headers["x-opencode-session"] ?? "", + UUID_RE, + "Responses transport must use a UUID session" + ); +}); + test("OpencodeExecutor.buildHeaders derives a DIFFERENT x-opencode-session for a different conversation body", () => { const executor = new OpencodeExecutor("opencode-go"); const headersA = executor.buildHeaders(null, true, null, "big-pickle", undefined, {