diff --git a/packages/core/src/core/openaiContentGenerator/pipeline.test.ts b/packages/core/src/core/openaiContentGenerator/pipeline.test.ts index 5860736230..9e8282d182 100644 --- a/packages/core/src/core/openaiContentGenerator/pipeline.test.ts +++ b/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 () => { diff --git a/packages/core/src/core/openaiContentGenerator/pipeline.ts b/packages/core/src/core/openaiContentGenerator/pipeline.ts index 44ac348d15..1b4de3f98c 100644 --- a/packages/core/src/core/openaiContentGenerator/pipeline.ts +++ b/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; + // 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, - ); + const geminiResponse = + OpenAIContentConverter.convertOpenAIResponseToGemini( + openaiResponse, + context, + ); - return geminiResponse; + 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; + // Per-request child — same rationale as the non-streaming path. + const parentSignal = request.config?.abortSignal; + const perRequestAc = parentSignal + ? createChildAbortController(parentSignal) + : undefined; + let stream: AsyncIterable; + 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; + } 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 { + try { + yield* innerStream; + } finally { + ac.abort(); + } + } + return drainThenCleanup(); }, ); }