fix(core): isolate OpenAI SDK abort listener leak with per-request child controllers

packages/core/src/core/openaiContentGenerator/pipeline.test.ts

@@ -1201,10 +1201,38 @@ describe('ContentGenerationPipeline', () => {
 
       await pipeline.execute(request, 'test-id');
 
-      expect(mockClient.chat.completions.create).toHaveBeenCalledWith(
-        expect.any(Object),
-        expect.objectContaining({ signal: abortController.signal }),
+      // The pipeline wraps the caller's signal in a per-request child
+      // to isolate OpenAI SDK listener leaks, so the SDK receives a
+      // child AbortSignal, not the original.
+      const call = (mockClient.chat.completions.create as Mock).mock.calls[0];
+      const sdkSignal = call[1]?.signal;
+      expect(sdkSignal).toBeInstanceOf(AbortSignal);
+      expect(sdkSignal).not.toBe(abortController.signal);
+    });
+
+    it('should propagate parent abort to SDK child signal', async () => {
+      const abortController = new AbortController();
+      const request: GenerateContentParameters = {
+        model: 'test-model',
+        contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
+        config: { abortSignal: abortController.signal },
+      };
+
+      let capturedSignal: AbortSignal | undefined;
+      (mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
+      (mockClient.chat.completions.create as Mock).mockImplementation(
+        (_req: unknown, opts: { signal: AbortSignal }) => {
+          capturedSignal = opts.signal;
+          abortController.abort();
+          return { choices: [{ message: { content: 'ok' } }] };
+        },
       );
+      (mockConverter.convertOpenAIResponseToGemini as Mock).mockReturnValue(
+        new GenerateContentResponse(),
+      );
+
+      await pipeline.execute(request, 'test-id');
+      expect(capturedSignal!.aborted).toBe(true);
     });
   });
 
@@ -1378,17 +1406,18 @@ describe('ContentGenerationPipeline', () => {
       );
 
       // Assert
-      // The stream should handle the error internally - errors during iteration don't propagate to the consumer
-      // Instead, they are handled internally by the pipeline
+      // The error propagates to the consumer via the async generator;
+      // errorHandler.handle() is also called internally by the pipeline.
       const results = [];
+      let caughtError: unknown;
       try {
         for await (const result of resultGenerator) {
           results.push(result);
         }
       } catch (error) {
-        // This is expected - the error should propagate from the stream processing
-        expect(error).toBe(testError);
+        caughtError = error;
       }
+      expect(caughtError).toBe(testError);
 
       expect(results).toHaveLength(0); // No results due to error
       expect(mockErrorHandler.handle).toHaveBeenCalledWith(
@@ -1579,10 +1608,112 @@ describe('ContentGenerationPipeline', () => {
         // Consume stream
       }
 
-      expect(mockClient.chat.completions.create).toHaveBeenCalledWith(
-        expect.any(Object),
-        expect.objectContaining({ signal: abortController.signal }),
+      // Per-request child signal isolates SDK listener leaks
+      const call = (mockClient.chat.completions.create as Mock).mock.calls[0];
+      const sdkSignal = call[1]?.signal;
+      expect(sdkSignal).toBeInstanceOf(AbortSignal);
+      expect(sdkSignal).not.toBe(abortController.signal);
+    });
+
+    it('should abort child signal after stream is fully consumed', async () => {
+      const abortController = new AbortController();
+      const request: GenerateContentParameters = {
+        model: 'test-model',
+        contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
+        config: { abortSignal: abortController.signal },
+      };
+
+      const mockStream = {
+        async *[Symbol.asyncIterator]() {
+          yield {
+            id: 'chunk-1',
+            choices: [{ delta: { content: 'Hello' }, finish_reason: 'stop' }],
+          };
+        },
+      };
+
+      (mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
+      (mockConverter.convertOpenAIChunkToGemini as Mock).mockReturnValue(
+        new GenerateContentResponse(),
+      );
+      (mockClient.chat.completions.create as Mock).mockResolvedValue(
+        mockStream,
+      );
+
+      const resultGenerator = await pipeline.executeStream(request, 'test-id');
+      const sdkSignal = (mockClient.chat.completions.create as Mock).mock
+        .calls[0][1]?.signal as AbortSignal;
+      expect(sdkSignal.aborted).toBe(false);
+
+      for await (const _result of resultGenerator) {
+        // Consume stream
+      }
+
+      expect(sdkSignal.aborted).toBe(true);
+    });
+
+    it('should abort child signal when consumer breaks early', async () => {
+      const abortController = new AbortController();
+      const request: GenerateContentParameters = {
+        model: 'test-model',
+        contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
+        config: { abortSignal: abortController.signal },
+      };
+
+      const mockStream = {
+        async *[Symbol.asyncIterator]() {
+          yield {
+            id: 'chunk-1',
+            choices: [{ delta: { content: 'a' }, finish_reason: null }],
+          };
+          yield {
+            id: 'chunk-2',
+            choices: [{ delta: { content: 'b' }, finish_reason: 'stop' }],
+          };
+        },
+      };
+
+      (mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
+      (mockConverter.convertOpenAIChunkToGemini as Mock).mockReturnValue(
+        new GenerateContentResponse(),
       );
+      (mockClient.chat.completions.create as Mock).mockResolvedValue(
+        mockStream,
+      );
+
+      const resultGenerator = await pipeline.executeStream(request, 'test-id');
+      const sdkSignal = (mockClient.chat.completions.create as Mock).mock
+        .calls[0][1]?.signal as AbortSignal;
+
+      for await (const _result of resultGenerator) {
+        break;
+      }
+
+      expect(sdkSignal.aborted).toBe(true);
+    });
+
+    it('should abort child signal when SDK create() throws', async () => {
+      const abortController = new AbortController();
+      const request: GenerateContentParameters = {
+        model: 'test-model',
+        contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
+        config: { abortSignal: abortController.signal },
+      };
+
+      let capturedSignal: AbortSignal | undefined;
+      (mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
+      (mockClient.chat.completions.create as Mock).mockImplementation(
+        (_req: unknown, opts: { signal: AbortSignal }) => {
+          capturedSignal = opts.signal;
+          throw new Error('network failure');
+        },
+      );
+
+      await expect(
+        pipeline.executeStream(request, 'test-id'),
+      ).rejects.toThrow();
+
+      expect(capturedSignal!.aborted).toBe(true);
     });
 
     it('should merge finishReason and usageMetadata from separate chunks', async () => {

packages/core/src/core/openaiContentGenerator/pipeline.ts

@@ -19,6 +19,7 @@ import { TaggedThinkingParser } from './taggedThinkingParser.js';
 import type { PipelineConfig, RequestContext } from './types.js';
 import { redactProxyError } from '../../utils/runtimeFetchOptions.js';
 import { runtimeDiagnostics } from '../../utils/runtimeDiagnostics.js';
+import { createChildAbortController } from '../../utils/abortController.js';
 
 /**
  * Error thrown when the API returns an error embedded as stream content
@@ -53,20 +54,32 @@ export class ContentGenerationPipeline {
       userPromptId,
       false,
       async (openaiRequest, context) => {
-        const openaiResponse = (await this.client.chat.completions.create(
-          openaiRequest,
-          {
-            signal: request.config?.abortSignal,
-          },
-        )) as OpenAI.Chat.ChatCompletion;
-
-        const geminiResponse =
-          OpenAIContentConverter.convertOpenAIResponseToGemini(
-            openaiResponse,
-            context,
-          );
-
-        return geminiResponse;
+        // Wrap in a per-request child so the OpenAI SDK's leaked abort
+        // listener (client.mjs fetchWithTimeout — no {once:true}, no
+        // removeEventListener) stays on a short-lived signal instead of
+        // accumulating on the caller's long-lived round signal.
+        const parentSignal = request.config?.abortSignal;
+        const perRequestAc = parentSignal
+          ? createChildAbortController(parentSignal)
+          : undefined;
+        try {
+          const openaiResponse = (await this.client.chat.completions.create(
+            openaiRequest,
+            {
+              signal: perRequestAc?.signal,
+            },
+          )) as OpenAI.Chat.ChatCompletion;
+
+          const geminiResponse =
+            OpenAIContentConverter.convertOpenAIResponseToGemini(
+              openaiResponse,
+              context,
+            );
+
+          return geminiResponse;
+        } finally {
+          perRequestAc?.abort();
+        }
       },
     );
   }
@@ -80,16 +93,48 @@ export class ContentGenerationPipeline {
       userPromptId,
       true,
       async (openaiRequest, context) => {
-        // Stage 1: Create OpenAI stream
-        const stream = (await this.client.chat.completions.create(
-          openaiRequest,
-          {
-            signal: request.config?.abortSignal,
-          },
-        )) as AsyncIterable<OpenAI.Chat.ChatCompletionChunk>;
+        // Per-request child — same rationale as the non-streaming path.
+        const parentSignal = request.config?.abortSignal;
+        const perRequestAc = parentSignal
+          ? createChildAbortController(parentSignal)
+          : undefined;
+        let stream: AsyncIterable<OpenAI.Chat.ChatCompletionChunk>;
+        try {
+          // Stage 1: Create OpenAI stream. Wrapped in try so a network /
+          // DNS / proxy error during the SDK call still cleans up the
+          // per-request child (same pattern as the non-streaming path).
+          stream = (await this.client.chat.completions.create(openaiRequest, {
+            signal: perRequestAc?.signal,
+          })) as AsyncIterable<OpenAI.Chat.ChatCompletionChunk>;
+        } catch (e) {
+          perRequestAc?.abort();
+          throw e;
+        }
 
-        // Stage 2: Process stream with conversion and logging
-        return this.processStreamWithLogging(stream, context, request);
+        // Stage 2: Process stream with conversion and logging.
+        // When a per-request controller exists, wrap in an async generator
+        // that aborts it once the stream is fully consumed or abandoned, so
+        // the child signal's reverse-cleanup fires and the parent listener
+        // is released.
+        if (!perRequestAc) {
+          return this.processStreamWithLogging(stream, context, request);
+        }
+        // Capture the narrowed controller so the closure below sees a non-
+        // nullable type (TS does not propagate narrowing into nested funcs).
+        const ac = perRequestAc;
+        const innerStream = this.processStreamWithLogging(
+          stream,
+          context,
+          request,
+        );
+        async function* drainThenCleanup(): AsyncGenerator<GenerateContentResponse> {
+          try {
+            yield* innerStream;
+          } finally {
+            ac.abort();
+          }
+        }
+        return drainThenCleanup();
       },
     );
   }