mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-07-09 17:19:02 +00:00
fix(core): detect non-SSE HTTP 200 responses in OpenAI streaming pipeline (#6466)
* Initial plan * fix(core): detect non-SSE HTTP 200 responses in OpenAI streaming pipeline When a gateway/proxy intercepts a streaming request and returns HTTP 200 with a non-SSE content-type (e.g. text/html), the OpenAI SDK silently produces zero chunks, resulting in a confusing "Model stream ended without a finish reason" error and an empty interaction log (response: null, error: null). Add NonSSEResponseError that is thrown early when the response content-type is incompatible with SSE (not text/event-stream, application/json, etc.). The error carries diagnostic metadata: HTTP status, content-type, bounded body prefix, and request-id header — allowing users and maintainers to distinguish "empty model stream" from "gateway returned non-SSE page". Uses the OpenAI SDK's withResponse() API to access the raw HTTP response headers before consuming the stream iterator. Falls back gracefully when withResponse() is unavailable (e.g. in mocked environments). Closes QwenLM/qwen-code#6465 * fix(core): replace non-null assertion with nullish coalescing in content-type check * fix(core): address non-sse review feedback * fix(core): expose non-sse request id alias --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: yiliang114 <effortyiliang@gmail.com>
This commit is contained in:
Copilot
GitHub
parent
b2b02d27ff
commit
e7f94a5a3c
2 changed files with 430 additions and 3 deletions
|
|
@ -12,6 +12,7 @@ import { GenerateContentResponse, Type, FinishReason } from '@google/genai';
|
|||
import type { ErrorHandler, PipelineConfig } from './types.js';
|
||||
import {
|
||||
ContentGenerationPipeline,
|
||||
NonSSEResponseError,
|
||||
StreamContentError,
|
||||
StreamInactivityTimeoutError,
|
||||
} from './pipeline.js';
|
||||
|
|
@ -1851,6 +1852,321 @@ describe('ContentGenerationPipeline', () => {
|
|||
expect(mockErrorHandler.handle).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should throw NonSSEResponseError when response has non-SSE content-type', async () => {
|
||||
const request: GenerateContentParameters = {
|
||||
model: 'test-model',
|
||||
contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
|
||||
};
|
||||
const userPromptId = 'test-prompt-id';
|
||||
|
||||
const mockStream = {
|
||||
async *[Symbol.asyncIterator]() {
|
||||
// Intentionally yields nothing — simulates an HTML body parsed as SSE
|
||||
},
|
||||
};
|
||||
|
||||
const mockHttpResponse = {
|
||||
headers: new Headers({
|
||||
'content-type': 'text/html;charset=UTF-8',
|
||||
'x-request-id': 'req-123',
|
||||
}),
|
||||
status: 200,
|
||||
body: null,
|
||||
} as unknown as Response;
|
||||
|
||||
// Create a mock API promise that has withResponse()
|
||||
const mockApiPromise = Object.assign(Promise.resolve(mockStream), {
|
||||
withResponse: () =>
|
||||
Promise.resolve({
|
||||
data: mockStream,
|
||||
response: mockHttpResponse,
|
||||
request_id: 'req-123',
|
||||
}),
|
||||
});
|
||||
|
||||
(mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
|
||||
(mockClient.chat.completions.create as Mock).mockReturnValue(
|
||||
mockApiPromise,
|
||||
);
|
||||
|
||||
let thrownError: NonSSEResponseError | undefined;
|
||||
try {
|
||||
await pipeline.executeStream(request, userPromptId);
|
||||
} catch (e) {
|
||||
thrownError = e as NonSSEResponseError;
|
||||
}
|
||||
|
||||
expect(thrownError).toBeInstanceOf(NonSSEResponseError);
|
||||
expect(thrownError!.httpStatus).toBe(200);
|
||||
expect(thrownError!.status).toBe(200);
|
||||
expect(thrownError!.requestId).toBe('req-123');
|
||||
expect(thrownError!.request_id).toBe('req-123');
|
||||
});
|
||||
|
||||
it('should throw NonSSEResponseError for application/json streaming responses', async () => {
|
||||
const request: GenerateContentParameters = {
|
||||
model: 'test-model',
|
||||
contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
|
||||
};
|
||||
|
||||
const jsonBody = '{"error":"gateway blocked streaming request"}';
|
||||
const mockStream = {
|
||||
async *[Symbol.asyncIterator]() {},
|
||||
};
|
||||
|
||||
const bodyStream = new ReadableStream({
|
||||
start(controller) {
|
||||
controller.enqueue(new TextEncoder().encode(jsonBody));
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
const mockHttpResponse = {
|
||||
headers: new Headers({
|
||||
'content-type': 'application/json',
|
||||
}),
|
||||
status: 200,
|
||||
body: bodyStream,
|
||||
} as unknown as Response;
|
||||
|
||||
const mockApiPromise = Object.assign(Promise.resolve(mockStream), {
|
||||
withResponse: () =>
|
||||
Promise.resolve({
|
||||
data: mockStream,
|
||||
response: mockHttpResponse,
|
||||
request_id: 'req-json',
|
||||
}),
|
||||
});
|
||||
|
||||
(mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
|
||||
(mockClient.chat.completions.create as Mock).mockReturnValue(
|
||||
mockApiPromise,
|
||||
);
|
||||
|
||||
let thrownError: NonSSEResponseError | undefined;
|
||||
try {
|
||||
await pipeline.executeStream(request, 'test-id');
|
||||
} catch (e) {
|
||||
thrownError = e as NonSSEResponseError;
|
||||
}
|
||||
|
||||
expect(thrownError).toBeInstanceOf(NonSSEResponseError);
|
||||
expect(thrownError!.contentType).toBe('application/json');
|
||||
expect(thrownError!.bodyPrefix).toContain('gateway blocked');
|
||||
expect(thrownError!.requestId).toBe('req-json');
|
||||
});
|
||||
|
||||
it('should include body prefix in NonSSEResponseError when body is readable', async () => {
|
||||
const request: GenerateContentParameters = {
|
||||
model: 'test-model',
|
||||
contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
|
||||
};
|
||||
|
||||
const htmlBody = '<html>' + 'x'.repeat(700) + '</html>';
|
||||
const mockStream = {
|
||||
async *[Symbol.asyncIterator]() {},
|
||||
};
|
||||
|
||||
// Create a ReadableStream with the HTML content
|
||||
const bodyStream = new ReadableStream({
|
||||
start(controller) {
|
||||
controller.enqueue(new TextEncoder().encode(htmlBody));
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
const mockHttpResponse = {
|
||||
headers: new Headers({
|
||||
'content-type': 'text/html',
|
||||
}),
|
||||
status: 200,
|
||||
body: bodyStream,
|
||||
} as unknown as Response;
|
||||
|
||||
const mockApiPromise = Object.assign(Promise.resolve(mockStream), {
|
||||
withResponse: () =>
|
||||
Promise.resolve({
|
||||
data: mockStream,
|
||||
response: mockHttpResponse,
|
||||
request_id: null,
|
||||
}),
|
||||
});
|
||||
|
||||
(mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
|
||||
(mockClient.chat.completions.create as Mock).mockReturnValue(
|
||||
mockApiPromise,
|
||||
);
|
||||
|
||||
let thrownError: NonSSEResponseError | undefined;
|
||||
try {
|
||||
await pipeline.executeStream(request, 'test-id');
|
||||
} catch (e) {
|
||||
thrownError = e as NonSSEResponseError;
|
||||
}
|
||||
|
||||
expect(thrownError).toBeInstanceOf(NonSSEResponseError);
|
||||
expect(thrownError!.contentType).toBe('text/html');
|
||||
expect(thrownError!.httpStatus).toBe(200);
|
||||
expect(thrownError!.bodyPrefix).toBe(htmlBody.slice(0, 512));
|
||||
expect(thrownError!.bodyPrefix).toHaveLength(512);
|
||||
expect(thrownError!.requestId).toBeNull();
|
||||
});
|
||||
|
||||
it('should still throw NonSSEResponseError when body prefix read fails', async () => {
|
||||
const request: GenerateContentParameters = {
|
||||
model: 'test-model',
|
||||
contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
|
||||
};
|
||||
|
||||
const mockStream = {
|
||||
async *[Symbol.asyncIterator]() {},
|
||||
};
|
||||
|
||||
const bodyStream = new ReadableStream({
|
||||
start(controller) {
|
||||
controller.error(new Error('body already consumed'));
|
||||
},
|
||||
});
|
||||
|
||||
const mockHttpResponse = {
|
||||
headers: new Headers({
|
||||
'content-type': 'text/html',
|
||||
}),
|
||||
status: 200,
|
||||
body: bodyStream,
|
||||
} as unknown as Response;
|
||||
|
||||
const mockApiPromise = Object.assign(Promise.resolve(mockStream), {
|
||||
withResponse: () =>
|
||||
Promise.resolve({
|
||||
data: mockStream,
|
||||
response: mockHttpResponse,
|
||||
request_id: null,
|
||||
}),
|
||||
});
|
||||
|
||||
(mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
|
||||
(mockClient.chat.completions.create as Mock).mockReturnValue(
|
||||
mockApiPromise,
|
||||
);
|
||||
|
||||
let thrownError: NonSSEResponseError | undefined;
|
||||
try {
|
||||
await pipeline.executeStream(request, 'test-id');
|
||||
} catch (e) {
|
||||
thrownError = e as NonSSEResponseError;
|
||||
}
|
||||
|
||||
expect(thrownError).toBeInstanceOf(NonSSEResponseError);
|
||||
expect(thrownError!.bodyPrefix).toBe('');
|
||||
});
|
||||
|
||||
it('should not throw NonSSEResponseError for text/event-stream content-type', async () => {
|
||||
const request: GenerateContentParameters = {
|
||||
model: 'test-model',
|
||||
contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
|
||||
};
|
||||
|
||||
const mockGeminiResponse = new GenerateContentResponse();
|
||||
mockGeminiResponse.candidates = [
|
||||
{
|
||||
content: { parts: [{ text: 'Hello' }], role: 'model' },
|
||||
finishReason: FinishReason.STOP,
|
||||
},
|
||||
];
|
||||
|
||||
const mockStream = {
|
||||
async *[Symbol.asyncIterator]() {
|
||||
yield {
|
||||
id: 'chunk-1',
|
||||
object: 'chat.completion.chunk',
|
||||
created: Date.now(),
|
||||
model: 'test-model',
|
||||
choices: [
|
||||
{ index: 0, delta: { content: 'Hello' }, finish_reason: 'stop' },
|
||||
],
|
||||
} as OpenAI.Chat.ChatCompletionChunk;
|
||||
},
|
||||
};
|
||||
|
||||
const mockHttpResponse = {
|
||||
headers: new Headers({
|
||||
'content-type': 'text/event-stream',
|
||||
}),
|
||||
status: 200,
|
||||
body: null,
|
||||
} as unknown as Response;
|
||||
|
||||
const mockApiPromise = Object.assign(Promise.resolve(mockStream), {
|
||||
withResponse: () =>
|
||||
Promise.resolve({
|
||||
data: mockStream,
|
||||
response: mockHttpResponse,
|
||||
request_id: null,
|
||||
}),
|
||||
});
|
||||
|
||||
(mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
|
||||
(mockConverter.convertOpenAIChunkToGemini as Mock).mockReturnValue(
|
||||
mockGeminiResponse,
|
||||
);
|
||||
(mockClient.chat.completions.create as Mock).mockReturnValue(
|
||||
mockApiPromise,
|
||||
);
|
||||
|
||||
const resultGenerator = await pipeline.executeStream(request, 'test-id');
|
||||
const results = [];
|
||||
for await (const result of resultGenerator) {
|
||||
results.push(result);
|
||||
}
|
||||
expect(results.length).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
it('should fall back to regular await when withResponse is not available', async () => {
|
||||
const request: GenerateContentParameters = {
|
||||
model: 'test-model',
|
||||
contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
|
||||
};
|
||||
|
||||
const mockGeminiResponse = new GenerateContentResponse();
|
||||
mockGeminiResponse.candidates = [
|
||||
{
|
||||
content: { parts: [{ text: 'Hello' }], role: 'model' },
|
||||
finishReason: FinishReason.STOP,
|
||||
},
|
||||
];
|
||||
|
||||
const mockStream = {
|
||||
async *[Symbol.asyncIterator]() {
|
||||
yield {
|
||||
id: 'chunk-1',
|
||||
object: 'chat.completion.chunk',
|
||||
created: Date.now(),
|
||||
model: 'test-model',
|
||||
choices: [
|
||||
{ index: 0, delta: { content: 'Hello' }, finish_reason: 'stop' },
|
||||
],
|
||||
} as OpenAI.Chat.ChatCompletionChunk;
|
||||
},
|
||||
};
|
||||
|
||||
(mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
|
||||
(mockConverter.convertOpenAIChunkToGemini as Mock).mockReturnValue(
|
||||
mockGeminiResponse,
|
||||
);
|
||||
// Regular mockResolvedValue — no withResponse method
|
||||
(mockClient.chat.completions.create as Mock).mockResolvedValue(
|
||||
mockStream,
|
||||
);
|
||||
|
||||
const resultGenerator = await pipeline.executeStream(request, 'test-id');
|
||||
const results = [];
|
||||
for await (const result of resultGenerator) {
|
||||
results.push(result);
|
||||
}
|
||||
expect(results.length).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
it('should pass abort signal to OpenAI client for streaming requests', async () => {
|
||||
const abortController = new AbortController();
|
||||
const request: GenerateContentParameters = {
|
||||
|
|
|
|||
|
|
@ -65,6 +65,54 @@ export class StreamInactivityTimeoutError extends Error {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Maximum bytes of response body to include in NonSSEResponseError diagnostics.
|
||||
*/
|
||||
const NON_SSE_BODY_PREFIX_LIMIT = 512;
|
||||
|
||||
/**
|
||||
* Content-type prefixes that are compatible with SSE streaming. Anything
|
||||
* outside this set (e.g. `text/html`) indicates the upstream did not return
|
||||
* an SSE stream — typically a gateway/proxy interception page.
|
||||
*/
|
||||
function isSSECompatibleContentType(contentType: string | null): boolean {
|
||||
if (!contentType) return true; // absence → assume SSE (SDK default)
|
||||
const mediaType = (contentType.split(';')[0] ?? '').trim().toLowerCase();
|
||||
return (
|
||||
mediaType === 'text/event-stream' ||
|
||||
mediaType === 'application/x-ndjson' ||
|
||||
mediaType === 'application/stream+json'
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Thrown when the HTTP 200 response to a streaming request has a content-type
|
||||
* incompatible with SSE (e.g. `text/html` from a gateway block page). Carries
|
||||
* bounded diagnostic metadata so the user/maintainer can distinguish "model
|
||||
* returned empty stream" from "upstream returned a non-SSE page".
|
||||
*/
|
||||
export class NonSSEResponseError extends Error {
|
||||
readonly status: number;
|
||||
readonly request_id: string | null;
|
||||
|
||||
constructor(
|
||||
readonly contentType: string | null,
|
||||
readonly httpStatus: number,
|
||||
readonly bodyPrefix: string,
|
||||
readonly requestId: string | null,
|
||||
) {
|
||||
const preview = bodyPrefix.length > 0 ? ` Body prefix: ${bodyPrefix}` : '';
|
||||
super(
|
||||
`Streaming request received a non-SSE response ` +
|
||||
`(HTTP ${httpStatus}, Content-Type: ${contentType || 'unknown'}).` +
|
||||
`${preview}`,
|
||||
);
|
||||
this.name = 'NonSSEResponseError';
|
||||
this.status = httpStatus;
|
||||
this.request_id = requestId;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve the effective streaming inactivity timeout (ms). Precedence:
|
||||
* explicit `ContentGeneratorConfig.streamIdleTimeoutMs` (programmatic, wins —
|
||||
|
|
@ -256,9 +304,72 @@ export class ContentGenerationPipeline {
|
|||
// 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>;
|
||||
//
|
||||
// Use withResponse() to access HTTP response headers — this allows
|
||||
// early detection of non-SSE responses (e.g. gateway block pages
|
||||
// returning text/html with HTTP 200).
|
||||
const createPromise = this.client.chat.completions.create(
|
||||
openaiRequest,
|
||||
{ signal: perRequestAc.signal },
|
||||
);
|
||||
|
||||
// withResponse() is available on APIPromise (the OpenAI SDK's
|
||||
// extended Promise). If unavailable (e.g. a mock), fall back.
|
||||
if (
|
||||
typeof (createPromise as { withResponse?: unknown })
|
||||
.withResponse === 'function'
|
||||
) {
|
||||
const {
|
||||
data,
|
||||
response: httpResponse,
|
||||
request_id,
|
||||
} = await (
|
||||
createPromise as unknown as {
|
||||
withResponse(): Promise<{
|
||||
data: AsyncIterable<OpenAI.Chat.ChatCompletionChunk>;
|
||||
response: Response;
|
||||
request_id: string | null;
|
||||
}>;
|
||||
}
|
||||
).withResponse();
|
||||
stream = data;
|
||||
|
||||
// Validate content-type: a non-SSE content-type on a streaming
|
||||
// request means the upstream (gateway/proxy) returned something
|
||||
// other than an event stream — surface it immediately.
|
||||
const contentType =
|
||||
httpResponse.headers.get('content-type') ?? null;
|
||||
if (!isSSECompatibleContentType(contentType)) {
|
||||
// Read a bounded prefix of the body for diagnostics. The body
|
||||
// may already be consumed by the SDK's stream parser; in that
|
||||
// case we fall through with an empty prefix.
|
||||
let bodyPrefix = '';
|
||||
try {
|
||||
if (httpResponse.body) {
|
||||
const reader = httpResponse.body.getReader();
|
||||
const { value } = await reader.read();
|
||||
reader.releaseLock();
|
||||
if (value) {
|
||||
bodyPrefix = new TextDecoder()
|
||||
.decode(value)
|
||||
.slice(0, NON_SSE_BODY_PREFIX_LIMIT);
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// Body already consumed by the SDK — expected; proceed
|
||||
// without the prefix.
|
||||
}
|
||||
throw new NonSSEResponseError(
|
||||
contentType,
|
||||
httpResponse.status,
|
||||
bodyPrefix,
|
||||
request_id,
|
||||
);
|
||||
}
|
||||
} else {
|
||||
stream =
|
||||
(await createPromise) as AsyncIterable<OpenAI.Chat.ChatCompletionChunk>;
|
||||
}
|
||||
} catch (e) {
|
||||
perRequestAc.abort();
|
||||
throw e;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue