fix: propagate websocket aborts to sidecars

Author: lidge-jun

src/abort.ts

@@ -0,0 +1,29 @@
+export interface LinkedAbortSignal {
+  signal: AbortSignal;
+  cleanup: () => void;
+}
+
+export function signalWithTimeout(timeoutMs: number, parent?: AbortSignal): LinkedAbortSignal {
+  const controller = new AbortController();
+  const timeout = setTimeout(() => {
+    if (!controller.signal.aborted) controller.abort(new DOMException("Timeout elapsed", "TimeoutError"));
+  }, timeoutMs);
+
+  const abortFromParent = () => {
+    if (!controller.signal.aborted) controller.abort(parent?.reason);
+  };
+
+  if (parent?.aborted) {
+    abortFromParent();
+  } else {
+    parent?.addEventListener("abort", abortFromParent, { once: true });
+  }
+
+  return {
+    signal: controller.signal,
+    cleanup: () => {
+      clearTimeout(timeout);
+      parent?.removeEventListener("abort", abortFromParent);
+    },
+  };
+}

src/server.ts

@@ -150,7 +150,7 @@ async function handleResponses(
   // with text BEFORE the main call, so the text-only model can reason about it.
   const visionPlan = planVisionSidecar(config, route.provider, route.modelId, parsed, req.headers);
   if (visionPlan) {
-    await describeImagesInPlace(parsed, visionPlan.forwardProvider, req.headers, visionPlan.settings);
+    await describeImagesInPlace(parsed, visionPlan.forwardProvider, req.headers, visionPlan.settings, options.abortSignal);
   }
 
   const adapter = resolveAdapter(route.provider);
@@ -192,6 +192,7 @@ async function handleResponses(
       incomingHeaders: req.headers,
       settings: wsPlan.settings,
       maxSearches: wsPlan.maxSearches,
+      abortSignal: options.abortSignal,
     });
   }
 

src/vision/describe.ts

@@ -1,5 +1,6 @@
 import type { OcxProviderConfig } from "../types";
 import { FORWARD_HEADERS } from "../adapters/openai-responses";
+import { signalWithTimeout } from "../abort";
 import { parseSidecarSSE } from "../web-search/parse";
 
 export interface VisionSettings {
@@ -48,6 +49,7 @@ export async function describeImage(
   forwardProvider: OcxProviderConfig,
   incomingHeaders: Headers,
   settings: VisionSettings,
+  abortSignal?: AbortSignal,
 ): Promise<DescribeOutcome> {
   const invalid = validateImageUrl(imageUrl);
   if (invalid) return { text: "", error: invalid };
@@ -76,12 +78,13 @@ export async function describeImage(
     store: false,
     stream: true,
   };
+  const linkedSignal = signalWithTimeout(settings.timeoutMs, abortSignal);
   try {
     const res = await fetch(`${forwardProvider.baseUrl}/responses`, {
       method: "POST",
       headers,
       body: JSON.stringify(body),
-      signal: AbortSignal.timeout(settings.timeoutMs),
+      signal: linkedSignal.signal,
     });
     if (!res.ok) {
       const t = await res.text().catch(() => "");
@@ -94,5 +97,7 @@ export async function describeImage(
     return { text: parsed.text };
   } catch (e) {
     return { text: "", error: e instanceof Error ? e.message : String(e) };
+  } finally {
+    linkedSignal.cleanup();
   }
 }

src/vision/index.ts

@@ -107,6 +107,7 @@ export async function describeImagesInPlace(
   forwardProvider: OcxProviderConfig,
   incomingHeaders: Headers,
   settings: VisionSettings,
+  abortSignal?: AbortSignal,
 ): Promise<void> {
   // 1. Gather every image part across messages, each with its own message's text as context.
   const jobs: ImageJob[] = [];
@@ -129,7 +130,7 @@ export async function describeImagesInPlace(
 
   // 2. Describe all images with bounded concurrency (order preserved).
   const outcomes = await runBounded(jobs, VISION_CONCURRENCY, j =>
-    describeImage(j.imageUrl, j.detail, j.contextText, forwardProvider, incomingHeaders, settings));
+    describeImage(j.imageUrl, j.detail, j.contextText, forwardProvider, incomingHeaders, settings, abortSignal));
 
   // 3. Rebuild each message, replacing image parts with their descriptions in order.
   let oi = 0;

src/web-search/executor.ts

@@ -1,5 +1,6 @@
 import type { OcxProviderConfig } from "../types";
 import { FORWARD_HEADERS } from "../adapters/openai-responses";
+import { signalWithTimeout } from "../abort";
 import { parseSidecarSSE, type WebSearchResult } from "./parse";
 
 export interface SidecarSettings {
@@ -36,6 +37,7 @@ export async function runWebSearch(
   forwardProvider: OcxProviderConfig,
   incomingHeaders: Headers,
   settings: SidecarSettings,
+  abortSignal?: AbortSignal,
 ): Promise<SidecarOutcome> {
   const headers: Record<string, string> = { "Content-Type": "application/json" };
   if (forwardProvider.headers) Object.assign(headers, forwardProvider.headers);
@@ -57,12 +59,13 @@ export async function runWebSearch(
     stream: true,
   };
   const url = `${forwardProvider.baseUrl}/responses`;
+  const linkedSignal = signalWithTimeout(settings.timeoutMs, abortSignal);
   try {
     const res = await fetch(url, {
       method: "POST",
       headers,
       body: JSON.stringify(body),
-      signal: AbortSignal.timeout(settings.timeoutMs),
+      signal: linkedSignal.signal,
     });
     if (!res.ok) {
       const t = await res.text().catch(() => "");
@@ -71,5 +74,7 @@ export async function runWebSearch(
     return await parseSidecarSSE(res);
   } catch (e) {
     return { text: "", sources: [], error: e instanceof Error ? e.message : String(e) };
+  } finally {
+    linkedSignal.cleanup();
   }
 }

src/web-search/loop.ts

@@ -95,6 +95,7 @@ export interface WebSearchLoopDeps {
   incomingHeaders: Headers;
   settings: SidecarSettings;
   maxSearches: number;
+  abortSignal?: AbortSignal;
 }
 
 /**
@@ -104,7 +105,7 @@ export interface WebSearchLoopDeps {
  * streamed Responses SSE. web_search calls are executed internally and never relayed to Codex.
  */
 export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise<Response> {
-  const { parsed, adapter, incomingHeaders, forwardProvider, hostedTool, settings, maxSearches } = deps;
+  const { parsed, adapter, incomingHeaders, forwardProvider, hostedTool, settings, maxSearches, abortSignal } = deps;
   if (!adapter.parseResponse) return jsonError(500, "web-search sidecar requires a non-streaming adapter");
 
   const messages: OcxMessage[] = [...parsed.context.messages];
@@ -159,7 +160,7 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise<Respons
         outcome = { text: "", sources: [], error: "the model called web_search with an empty query" };
         searchesExecuted++;
       } else {
-        outcome = await runWebSearch(call.query, hostedTool, forwardProvider, incomingHeaders, settings);
+        outcome = await runWebSearch(call.query, hostedTool, forwardProvider, incomingHeaders, settings, abortSignal);
         searchesExecuted++;
         if (outcome.error) failedQueries.add(normalizeQuery(call.query));
       }

tests/sidecar-abort.test.ts

@@ -0,0 +1,71 @@
+import { afterEach, describe, expect, test } from "bun:test";
+import { runWebSearch } from "../src/web-search/executor";
+import { describeImage } from "../src/vision/describe";
+import type { OcxProviderConfig } from "../src/types";
+
+const originalFetch = globalThis.fetch;
+
+const forwardProvider: OcxProviderConfig = {
+  adapter: "openai-responses",
+  baseUrl: "https://chatgpt.test",
+  authMode: "forward",
+};
+
+afterEach(() => {
+  globalThis.fetch = originalFetch;
+});
+
+function installAbortAwareFetch(): () => AbortSignal {
+  let seenSignal: AbortSignal | undefined;
+  globalThis.fetch = ((_, init) => {
+    seenSignal = init?.signal as AbortSignal | undefined;
+    return new Promise<Response>((_, reject) => {
+      seenSignal?.addEventListener("abort", () => reject(new Error("aborted by turn")), { once: true });
+    });
+  }) as typeof fetch;
+  return () => {
+    if (!seenSignal) throw new Error("fetch was not called");
+    return seenSignal;
+  };
+}
+
+describe("sidecar abort propagation", () => {
+  test("web-search sidecar fetch observes the WebSocket turn abort signal", async () => {
+    const getSignal = installAbortAwareFetch();
+    const turn = new AbortController();
+    const outcome = runWebSearch(
+      "current docs",
+      { type: "web_search" },
+      forwardProvider,
+      new Headers({ authorization: "Bearer token" }),
+      { model: "gpt-5.4-mini", reasoning: "low", timeoutMs: 30_000 },
+      turn.signal,
+    );
+
+    const signal = getSignal();
+    expect(signal.aborted).toBe(false);
+    turn.abort("replacement turn");
+    expect(signal.aborted).toBe(true);
+    expect((await outcome).error).toBe("aborted by turn");
+  });
+
+  test("vision sidecar fetch observes the WebSocket turn abort signal", async () => {
+    const getSignal = installAbortAwareFetch();
+    const turn = new AbortController();
+    const outcome = describeImage(
+      "data:image/png;base64,iVBORw0KGgo=",
+      "high",
+      "inspect screenshot",
+      forwardProvider,
+      new Headers({ authorization: "Bearer token" }),
+      { model: "gpt-5.4-mini", timeoutMs: 30_000 },
+      turn.signal,
+    );
+
+    const signal = getSignal();
+    expect(signal.aborted).toBe(false);
+    turn.abort("replacement turn");
+    expect(signal.aborted).toBe(true);
+    expect((await outcome).error).toBe("aborted by turn");
+  });
+});