fix(opencode): close Muse Responses streams at completion

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<string, unknown>;
+    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<Uint8Array>({
-      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<string, unknown>);
+        applyMuseSparkMinOutputTokens(
+          String(input.model ?? ""),
+          input.body as Record<string, unknown>
+        );
       }
 
       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;

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<T>(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<T>(
   providerOrIdentityOrFn: string | null | undefined | TlsTrackingIdentity | (() => T),
   maybeFn?: () => T
 ): Promise<{ result: Awaited<T>; 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<T>(
     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<T>(
 }
 
 /** 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 &&

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<string, string | undefined>,
@@ -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);
+    })
+  );
+});

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<string, unknown> = { 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<string>((_, reject) => setTimeout(() => reject(new Error("stream hung")), 1000)),
+    ]);
+    assert.match(text, /response.completed/);
+    assert.doesNotMatch(text, /\"type\":\"ping\"/);
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});

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, {