feat: Implement compatibility strategies for responses and chat endpoints
- Added `routeCompatibility.ts` to handle response compatibility for different endpoints. - Introduced `chatEndpointStrategy.ts` to manage chat endpoint recovery and downgrading logic. - Created tests for chat formats and input file handling in `chatFormatsCore.test.ts` and `inputFile.test.ts`. - Developed `endpointCompatibility.ts` to define compatibility checks and header building for various endpoints. - Implemented protocol lifecycle management in `protocolLifecycle.ts` and corresponding tests in `protocolModel.test.ts`. - Enhanced tooltip functionality with a new `TooltipLayer` component and associated tests. - Refactored existing components to improve tooltip handling and ensure proper rendering in the application.
This commit is contained in:
cita
@@ -2,7 +2,9 @@ import { describe, expect, it } from 'vitest';
|
||||
|
||||
import { createStreamTransformContext } from '../../shared/normalized.js';
|
||||
import {
|
||||
completeResponsesStream,
|
||||
createOpenAiResponsesAggregateState,
|
||||
failResponsesStream,
|
||||
serializeConvertedResponsesEvents,
|
||||
} from './aggregator.js';
|
||||
|
||||
@@ -23,6 +25,35 @@ function parseSsePayloads(lines: string[]): Array<Record<string, unknown>> {
|
||||
.filter((item): item is Record<string, unknown> => !!item);
|
||||
}
|
||||
|
||||
function parseSseEvents(lines: string[]): Array<{ event: string | null; payload: Record<string, unknown> | '[DONE]' }> {
|
||||
return lines
|
||||
.flatMap((line) => line.split('\n\n').filter((block) => block.trim().length > 0))
|
||||
.map((block) => {
|
||||
const eventLine = block
|
||||
.split('\n')
|
||||
.find((line) => line.startsWith('event: '));
|
||||
const dataLine = block
|
||||
.split('\n')
|
||||
.find((line) => line.startsWith('data: '));
|
||||
if (!dataLine) return null;
|
||||
if (dataLine === 'data: [DONE]') {
|
||||
return {
|
||||
event: eventLine ? eventLine.slice('event: '.length) : null,
|
||||
payload: '[DONE]' as const,
|
||||
};
|
||||
}
|
||||
try {
|
||||
return {
|
||||
event: eventLine ? eventLine.slice('event: '.length) : null,
|
||||
payload: JSON.parse(dataLine.slice('data: '.length)) as Record<string, unknown>,
|
||||
};
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
})
|
||||
.filter((item): item is { event: string | null; payload: Record<string, unknown> | '[DONE]' } => !!item);
|
||||
}
|
||||
|
||||
describe('serializeConvertedResponsesEvents', () => {
|
||||
it('aggregates reasoning summary events into the completed response payload', () => {
|
||||
const state = createOpenAiResponsesAggregateState('gpt-5');
|
||||
@@ -301,4 +332,174 @@ describe('serializeConvertedResponsesEvents', () => {
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
it('emits canonical message done events before response.completed when recovering a sparse text stream', () => {
|
||||
const state = createOpenAiResponsesAggregateState('gpt-5');
|
||||
const streamContext = createStreamTransformContext('gpt-5');
|
||||
const usage = {
|
||||
promptTokens: 2,
|
||||
completionTokens: 4,
|
||||
totalTokens: 6,
|
||||
};
|
||||
|
||||
serializeConvertedResponsesEvents({
|
||||
state,
|
||||
streamContext,
|
||||
usage,
|
||||
event: {
|
||||
contentDelta: 'hello world',
|
||||
} as any,
|
||||
});
|
||||
|
||||
const completionLines = completeResponsesStream(state, streamContext, usage);
|
||||
const events = parseSseEvents(completionLines);
|
||||
|
||||
expect(events.map((entry) => entry.event ?? (entry.payload === '[DONE]' ? '[DONE]' : 'data'))).toEqual([
|
||||
'response.output_text.done',
|
||||
'response.content_part.done',
|
||||
'response.output_item.done',
|
||||
'response.completed',
|
||||
'[DONE]',
|
||||
]);
|
||||
|
||||
expect(events[0]?.payload).toMatchObject({
|
||||
type: 'response.output_text.done',
|
||||
output_index: 0,
|
||||
item_id: 'msg_0',
|
||||
text: 'hello world',
|
||||
});
|
||||
expect(events[1]?.payload).toMatchObject({
|
||||
type: 'response.content_part.done',
|
||||
output_index: 0,
|
||||
item_id: 'msg_0',
|
||||
content_index: 0,
|
||||
part: {
|
||||
type: 'output_text',
|
||||
text: 'hello world',
|
||||
},
|
||||
});
|
||||
expect(events[2]?.payload).toMatchObject({
|
||||
type: 'response.output_item.done',
|
||||
output_index: 0,
|
||||
item: {
|
||||
id: 'msg_0',
|
||||
type: 'message',
|
||||
status: 'completed',
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it('emits canonical reasoning done events before response.completed when recovering sparse reasoning output', () => {
|
||||
const state = createOpenAiResponsesAggregateState('gpt-5');
|
||||
const streamContext = createStreamTransformContext('gpt-5');
|
||||
const usage = {
|
||||
promptTokens: 2,
|
||||
completionTokens: 4,
|
||||
totalTokens: 6,
|
||||
};
|
||||
|
||||
serializeConvertedResponsesEvents({
|
||||
state,
|
||||
streamContext,
|
||||
usage,
|
||||
event: {
|
||||
reasoningSignature: 'enc-signature',
|
||||
reasoningDelta: 'plan first',
|
||||
} as any,
|
||||
});
|
||||
|
||||
const completionLines = completeResponsesStream(state, streamContext, usage);
|
||||
const events = parseSseEvents(completionLines);
|
||||
|
||||
expect(events.map((entry) => entry.event ?? (entry.payload === '[DONE]' ? '[DONE]' : 'data'))).toEqual([
|
||||
'response.reasoning_summary_text.done',
|
||||
'response.reasoning_summary_part.done',
|
||||
'response.output_item.done',
|
||||
'response.completed',
|
||||
'[DONE]',
|
||||
]);
|
||||
|
||||
expect(events[0]?.payload).toMatchObject({
|
||||
type: 'response.reasoning_summary_text.done',
|
||||
item_id: 'rs_0',
|
||||
output_index: 0,
|
||||
summary_index: 0,
|
||||
text: 'plan first',
|
||||
});
|
||||
expect(events[1]?.payload).toMatchObject({
|
||||
type: 'response.reasoning_summary_part.done',
|
||||
item_id: 'rs_0',
|
||||
output_index: 0,
|
||||
summary_index: 0,
|
||||
part: {
|
||||
type: 'summary_text',
|
||||
text: 'plan first',
|
||||
},
|
||||
});
|
||||
expect(events[2]?.payload).toMatchObject({
|
||||
type: 'response.output_item.done',
|
||||
output_index: 0,
|
||||
item: {
|
||||
id: 'rs_0',
|
||||
type: 'reasoning',
|
||||
status: 'completed',
|
||||
encrypted_content: 'enc-signature',
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it('emits canonical message done events before response.failed when failing a sparse text stream', () => {
|
||||
const state = createOpenAiResponsesAggregateState('gpt-5');
|
||||
const streamContext = createStreamTransformContext('gpt-5');
|
||||
const usage = {
|
||||
promptTokens: 2,
|
||||
completionTokens: 4,
|
||||
totalTokens: 6,
|
||||
};
|
||||
|
||||
serializeConvertedResponsesEvents({
|
||||
state,
|
||||
streamContext,
|
||||
usage,
|
||||
event: {
|
||||
contentDelta: 'partial answer',
|
||||
} as any,
|
||||
});
|
||||
|
||||
const failedLines = failResponsesStream(state, streamContext, usage, {
|
||||
error: {
|
||||
message: 'upstream stream failed',
|
||||
},
|
||||
});
|
||||
const events = parseSseEvents(failedLines);
|
||||
|
||||
expect(events.map((entry) => entry.event ?? (entry.payload === '[DONE]' ? '[DONE]' : 'data'))).toEqual([
|
||||
'response.output_text.done',
|
||||
'response.content_part.done',
|
||||
'response.output_item.done',
|
||||
'response.failed',
|
||||
'[DONE]',
|
||||
]);
|
||||
|
||||
expect(events[2]?.payload).toMatchObject({
|
||||
type: 'response.output_item.done',
|
||||
output_index: 0,
|
||||
item: {
|
||||
id: 'msg_0',
|
||||
type: 'message',
|
||||
status: 'failed',
|
||||
},
|
||||
});
|
||||
expect(events[3]?.payload).toMatchObject({
|
||||
type: 'response.failed',
|
||||
response: {
|
||||
status: 'failed',
|
||||
output_text: 'partial answer',
|
||||
},
|
||||
error: {
|
||||
message: 'upstream stream failed',
|
||||
type: 'upstream_error',
|
||||
},
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -768,6 +768,99 @@ function buildSyntheticToolEvents(
|
||||
return lines;
|
||||
}
|
||||
|
||||
function buildSyntheticTerminalItemDoneEvents(
|
||||
state: OpenAiResponsesAggregateState,
|
||||
status: 'completed' | 'failed',
|
||||
): string[] {
|
||||
const lines: string[] = [];
|
||||
|
||||
for (let index = 0; index < state.outputItems.length; index += 1) {
|
||||
const item = state.outputItems[index];
|
||||
if (!isRecord(item)) continue;
|
||||
|
||||
const currentStatus = asTrimmedString(item.status).toLowerCase();
|
||||
if (currentStatus === 'completed' || currentStatus === 'failed') continue;
|
||||
|
||||
const itemType = asTrimmedString(item.type).toLowerCase();
|
||||
const itemId = ensureOutputItemId(asTrimmedString(item.id), 'out', index);
|
||||
item.id = itemId;
|
||||
|
||||
if (itemType === 'message') {
|
||||
const content = Array.isArray(item.content) ? item.content as AggregateOutputItem[] : [];
|
||||
const part = content[0];
|
||||
if (isRecord(part)) {
|
||||
const text = typeof part.text === 'string' ? part.text : String(part.text ?? '');
|
||||
lines.push(serializeSse('response.output_text.done', {
|
||||
type: 'response.output_text.done',
|
||||
output_index: index,
|
||||
item_id: itemId,
|
||||
text,
|
||||
}));
|
||||
lines.push(serializeSse('response.content_part.done', {
|
||||
type: 'response.content_part.done',
|
||||
output_index: index,
|
||||
item_id: itemId,
|
||||
content_index: 0,
|
||||
part: cloneJson(part),
|
||||
}));
|
||||
}
|
||||
} else if (itemType === 'reasoning') {
|
||||
const summary = Array.isArray(item.summary) ? item.summary as AggregateOutputItem[] : [];
|
||||
const part = summary[0];
|
||||
if (isRecord(part)) {
|
||||
const text = typeof part.text === 'string' ? part.text : String(part.text ?? '');
|
||||
lines.push(serializeSse('response.reasoning_summary_text.done', {
|
||||
type: 'response.reasoning_summary_text.done',
|
||||
item_id: itemId,
|
||||
output_index: index,
|
||||
summary_index: 0,
|
||||
text,
|
||||
}));
|
||||
lines.push(serializeSse('response.reasoning_summary_part.done', {
|
||||
type: 'response.reasoning_summary_part.done',
|
||||
item_id: itemId,
|
||||
output_index: index,
|
||||
summary_index: 0,
|
||||
part: cloneJson(part),
|
||||
}));
|
||||
}
|
||||
} else if (itemType === 'function_call') {
|
||||
const callId = asTrimmedString(item.call_id) || itemId;
|
||||
const name = asTrimmedString(item.name);
|
||||
const argumentsText = typeof item.arguments === 'string' ? item.arguments : String(item.arguments ?? '');
|
||||
lines.push(serializeSse('response.function_call_arguments.done', {
|
||||
type: 'response.function_call_arguments.done',
|
||||
item_id: itemId,
|
||||
call_id: callId,
|
||||
output_index: index,
|
||||
...(name ? { name } : {}),
|
||||
arguments: argumentsText,
|
||||
}));
|
||||
} else if (itemType === 'custom_tool_call') {
|
||||
const callId = asTrimmedString(item.call_id) || itemId;
|
||||
const name = asTrimmedString(item.name);
|
||||
const inputText = typeof item.input === 'string' ? item.input : String(item.input ?? '');
|
||||
lines.push(serializeSse('response.custom_tool_call_input.done', {
|
||||
type: 'response.custom_tool_call_input.done',
|
||||
item_id: itemId,
|
||||
call_id: callId,
|
||||
output_index: index,
|
||||
...(name ? { name } : {}),
|
||||
input: inputText,
|
||||
}));
|
||||
}
|
||||
|
||||
item.status = status;
|
||||
lines.push(serializeSse('response.output_item.done', {
|
||||
type: 'response.output_item.done',
|
||||
output_index: index,
|
||||
item: cloneJson(item),
|
||||
}));
|
||||
}
|
||||
|
||||
return lines;
|
||||
}
|
||||
|
||||
export function serializeConvertedResponsesEvents(input: {
|
||||
state: OpenAiResponsesAggregateState;
|
||||
streamContext: StreamTransformContext;
|
||||
@@ -806,8 +899,10 @@ export function completeResponsesStream(
|
||||
if (state.failed || state.completed) {
|
||||
return [serializeDone()];
|
||||
}
|
||||
const lines = buildSyntheticTerminalItemDoneEvents(state, 'completed');
|
||||
state.completed = true;
|
||||
return [
|
||||
...lines,
|
||||
serializeSse('response.completed', {
|
||||
type: 'response.completed',
|
||||
response: materializeResponse(state, streamContext, usage, null, 'completed'),
|
||||
@@ -825,6 +920,7 @@ export function failResponsesStream(
|
||||
if (state.failed) {
|
||||
return [serializeDone()];
|
||||
}
|
||||
const lines = buildSyntheticTerminalItemDoneEvents(state, 'failed');
|
||||
state.failed = true;
|
||||
const errorPayload = cloneRecord(payload);
|
||||
const message = (
|
||||
@@ -833,6 +929,7 @@ export function failResponsesStream(
|
||||
: (typeof errorPayload?.message === 'string' ? errorPayload.message : 'upstream stream failed')
|
||||
);
|
||||
return [
|
||||
...lines,
|
||||
serializeSse('response.failed', {
|
||||
type: 'response.failed',
|
||||
response: materializeResponse(state, streamContext, usage, cloneRecord(errorPayload?.response), 'failed'),
|
||||
|
||||
@@ -1,6 +1,55 @@
|
||||
import { normalizeResponsesInputForCompatibility, normalizeResponsesMessageItem } from './compatibility.js';
|
||||
import { createProtocolRequestEnvelope } from '../../shared/protocolModel.js';
|
||||
import { sanitizeResponsesBodyForProxy } from './conversion.js';
|
||||
import type {
|
||||
OpenAiResponsesParsedRequest,
|
||||
OpenAiResponsesRequestEnvelope,
|
||||
} from './model.js';
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return !!value && typeof value === 'object' && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function invalidRequest(message: string): { statusCode: number; payload: unknown } {
|
||||
return {
|
||||
statusCode: 400,
|
||||
payload: {
|
||||
error: {
|
||||
message,
|
||||
type: 'invalid_request_error',
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export const openAiResponsesInbound = {
|
||||
normalizeInput: normalizeResponsesInputForCompatibility,
|
||||
normalizeMessage: normalizeResponsesMessageItem,
|
||||
parse(
|
||||
body: unknown,
|
||||
options?: { defaultEncryptedReasoningInclude?: boolean },
|
||||
): { value?: OpenAiResponsesRequestEnvelope; error?: { statusCode: number; payload: unknown } } {
|
||||
const rawBody = isRecord(body) ? body : {};
|
||||
const requestedModel = typeof rawBody.model === 'string' ? rawBody.model.trim() : '';
|
||||
if (!requestedModel) {
|
||||
return { error: invalidRequest('model is required') };
|
||||
}
|
||||
|
||||
const isStream = rawBody.stream === true;
|
||||
const normalizedBody = sanitizeResponsesBodyForProxy(
|
||||
rawBody,
|
||||
requestedModel,
|
||||
isStream,
|
||||
{ defaultEncryptedReasoningInclude: options?.defaultEncryptedReasoningInclude },
|
||||
);
|
||||
|
||||
return {
|
||||
value: createProtocolRequestEnvelope({
|
||||
protocol: 'openai/responses',
|
||||
model: requestedModel,
|
||||
stream: isStream,
|
||||
rawBody: body,
|
||||
parsed: {
|
||||
normalizedBody,
|
||||
} satisfies OpenAiResponsesParsedRequest,
|
||||
}),
|
||||
};
|
||||
},
|
||||
};
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import { openAiResponsesTransformer } from './index.js';
|
||||
|
||||
describe('openAiResponsesTransformer.inbound', () => {
|
||||
it('returns a protocol request envelope with a normalized responses body', () => {
|
||||
const result = openAiResponsesTransformer.transformRequest({
|
||||
model: 'gpt-5',
|
||||
input: 'hello',
|
||||
reasoning: {
|
||||
effort: 'high',
|
||||
},
|
||||
});
|
||||
|
||||
expect(result.error).toBeUndefined();
|
||||
expect(result.value).toMatchObject({
|
||||
protocol: 'openai/responses',
|
||||
model: 'gpt-5',
|
||||
stream: false,
|
||||
rawBody: {
|
||||
model: 'gpt-5',
|
||||
input: 'hello',
|
||||
},
|
||||
parsed: {
|
||||
normalizedBody: {
|
||||
model: 'gpt-5',
|
||||
input: [
|
||||
{
|
||||
type: 'message',
|
||||
role: 'user',
|
||||
content: [
|
||||
{
|
||||
type: 'input_text',
|
||||
text: 'hello',
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
stream: false,
|
||||
},
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it('rejects requests without a model at the transformer boundary', () => {
|
||||
const result = openAiResponsesTransformer.transformRequest({
|
||||
input: 'hello',
|
||||
});
|
||||
|
||||
expect(result.error).toEqual({
|
||||
statusCode: 400,
|
||||
payload: {
|
||||
error: {
|
||||
message: 'model is required',
|
||||
type: 'invalid_request_error',
|
||||
},
|
||||
},
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -21,12 +21,20 @@ import {
|
||||
serializeConvertedResponsesEvents,
|
||||
} from './aggregator.js';
|
||||
import { openAiResponsesOutbound } from './outbound.js';
|
||||
import { openAiResponsesInbound } from './inbound.js';
|
||||
import { createResponsesProxyStreamSession } from './proxyStream.js';
|
||||
import { createResponsesEndpointStrategy } from './routeCompatibility.js';
|
||||
import { openAiResponsesStream } from './stream.js';
|
||||
import { openAiResponsesUsage } from './usage.js';
|
||||
import type {
|
||||
OpenAiResponsesParsedRequest as OpenAiResponsesParsedRequestModel,
|
||||
OpenAiResponsesRequestEnvelope as OpenAiResponsesRequestEnvelopeModel,
|
||||
} from './model.js';
|
||||
|
||||
export const openAiResponsesTransformer = {
|
||||
protocol: 'openai/responses' as const,
|
||||
inbound: {
|
||||
parse: openAiResponsesInbound.parse,
|
||||
normalizeInput: normalizeResponsesInputForCompatibility,
|
||||
normalizeMessage: normalizeResponsesMessageItem,
|
||||
normalizeContent: normalizeResponsesMessageContent,
|
||||
@@ -38,6 +46,7 @@ export const openAiResponsesTransformer = {
|
||||
stream: openAiResponsesStream,
|
||||
usage: openAiResponsesUsage,
|
||||
compatibility: {
|
||||
createEndpointStrategy: createResponsesEndpointStrategy,
|
||||
buildRetryBodies,
|
||||
buildRetryHeaders,
|
||||
shouldRetry,
|
||||
@@ -49,8 +58,14 @@ export const openAiResponsesTransformer = {
|
||||
complete: completeResponsesStream,
|
||||
fail: failResponsesStream,
|
||||
},
|
||||
transformRequest(body: unknown) {
|
||||
return body;
|
||||
proxyStream: {
|
||||
createSession: createResponsesProxyStreamSession,
|
||||
},
|
||||
transformRequest(
|
||||
body: unknown,
|
||||
options?: { defaultEncryptedReasoningInclude?: boolean },
|
||||
): { value?: OpenAiResponsesRequestEnvelopeModel; error?: { statusCode: number; payload: unknown } } {
|
||||
return openAiResponsesInbound.parse(body, options);
|
||||
},
|
||||
createStreamContext(modelName: string): StreamTransformContext {
|
||||
return openAiResponsesStream.createContext(modelName);
|
||||
@@ -68,6 +83,8 @@ export const openAiResponsesTransformer = {
|
||||
|
||||
export type OpenAiResponsesTransformer = typeof openAiResponsesTransformer;
|
||||
export type OpenAiResponsesAggregate = OpenAiResponsesAggregateState;
|
||||
export type OpenAiResponsesParsedRequest = OpenAiResponsesParsedRequestModel;
|
||||
export type OpenAiResponsesRequestEnvelope = OpenAiResponsesRequestEnvelopeModel;
|
||||
export {
|
||||
convertOpenAiBodyToResponsesBody,
|
||||
convertResponsesBodyToOpenAiBody,
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
import type { ProtocolRequestEnvelope } from '../../shared/protocolModel.js';
|
||||
|
||||
export type OpenAiResponsesParsedRequest = {
|
||||
normalizedBody: Record<string, unknown>;
|
||||
};
|
||||
|
||||
export type OpenAiResponsesRequestEnvelope = ProtocolRequestEnvelope<
|
||||
'openai/responses',
|
||||
OpenAiResponsesParsedRequest
|
||||
>;
|
||||
@@ -0,0 +1,111 @@
|
||||
import { createProxyStreamLifecycle } from '../../shared/protocolLifecycle.js';
|
||||
import { type ParsedSseEvent } from '../../shared/normalized.js';
|
||||
import { completeResponsesStream, createOpenAiResponsesAggregateState, failResponsesStream, serializeConvertedResponsesEvents } from './aggregator.js';
|
||||
import { openAiResponsesStream } from './stream.js';
|
||||
|
||||
type StreamReader = {
|
||||
read(): Promise<{ done: boolean; value?: Uint8Array }>;
|
||||
cancel(reason?: unknown): Promise<unknown>;
|
||||
releaseLock(): void;
|
||||
};
|
||||
|
||||
type ResponseSink = {
|
||||
end(): void;
|
||||
};
|
||||
|
||||
type ResponsesProxyStreamSessionInput = {
|
||||
modelName: string;
|
||||
successfulUpstreamPath: string;
|
||||
getUsage: () => {
|
||||
promptTokens: number;
|
||||
completionTokens: number;
|
||||
totalTokens: number;
|
||||
cacheReadTokens: number;
|
||||
cacheCreationTokens: number;
|
||||
promptTokensIncludeCache: boolean | null;
|
||||
};
|
||||
onParsedPayload?: (payload: unknown) => void;
|
||||
writeLines: (lines: string[]) => void;
|
||||
writeRaw: (chunk: string) => void;
|
||||
};
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return !!value && typeof value === 'object';
|
||||
}
|
||||
|
||||
export function createResponsesProxyStreamSession(input: ResponsesProxyStreamSessionInput) {
|
||||
const streamContext = openAiResponsesStream.createContext(input.modelName);
|
||||
const responsesState = createOpenAiResponsesAggregateState(input.modelName);
|
||||
let finalized = false;
|
||||
|
||||
const finalize = () => {
|
||||
if (finalized) return;
|
||||
finalized = true;
|
||||
input.writeLines(completeResponsesStream(responsesState, streamContext, input.getUsage()));
|
||||
};
|
||||
|
||||
const handleEventBlock = (eventBlock: ParsedSseEvent): boolean => {
|
||||
if (eventBlock.data === '[DONE]') {
|
||||
finalize();
|
||||
return true;
|
||||
}
|
||||
|
||||
let parsedPayload: unknown = null;
|
||||
try {
|
||||
parsedPayload = JSON.parse(eventBlock.data);
|
||||
} catch {
|
||||
parsedPayload = null;
|
||||
}
|
||||
|
||||
if (parsedPayload && typeof parsedPayload === 'object') {
|
||||
input.onParsedPayload?.(parsedPayload);
|
||||
}
|
||||
|
||||
const payloadType = (isRecord(parsedPayload) && typeof parsedPayload.type === 'string')
|
||||
? parsedPayload.type
|
||||
: '';
|
||||
const isFailureEvent = (
|
||||
eventBlock.event === 'error'
|
||||
|| eventBlock.event === 'response.failed'
|
||||
|| payloadType === 'error'
|
||||
|| payloadType === 'response.failed'
|
||||
);
|
||||
if (isFailureEvent) {
|
||||
input.writeLines(failResponsesStream(responsesState, streamContext, input.getUsage(), parsedPayload));
|
||||
finalized = true;
|
||||
return true;
|
||||
}
|
||||
|
||||
if (parsedPayload && typeof parsedPayload === 'object') {
|
||||
const normalizedEvent = openAiResponsesStream.normalizeEvent(parsedPayload, streamContext, input.modelName);
|
||||
input.writeLines(serializeConvertedResponsesEvents({
|
||||
state: responsesState,
|
||||
streamContext,
|
||||
event: normalizedEvent,
|
||||
usage: input.getUsage(),
|
||||
}));
|
||||
return false;
|
||||
}
|
||||
|
||||
input.writeLines(serializeConvertedResponsesEvents({
|
||||
state: responsesState,
|
||||
streamContext,
|
||||
event: { contentDelta: eventBlock.data },
|
||||
usage: input.getUsage(),
|
||||
}));
|
||||
return false;
|
||||
};
|
||||
|
||||
return {
|
||||
async run(reader: StreamReader | null | undefined, response: ResponseSink) {
|
||||
const lifecycle = createProxyStreamLifecycle<ParsedSseEvent>({
|
||||
reader,
|
||||
response,
|
||||
pullEvents: (buffer) => openAiResponsesStream.pullSseEvents(buffer),
|
||||
handleEvent: handleEventBlock,
|
||||
onEof: finalize,
|
||||
});
|
||||
await lifecycle.run();
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
import {
|
||||
buildMinimalJsonHeadersForCompatibility,
|
||||
isEndpointDowngradeError,
|
||||
isUnsupportedMediaTypeError,
|
||||
type CompatibilityEndpoint,
|
||||
} from '../../shared/endpointCompatibility.js';
|
||||
import type { EndpointAttemptContext, EndpointRecoverResult } from '../../../routes/proxy/endpointFlow.js';
|
||||
import {
|
||||
buildResponsesCompatibilityBodies,
|
||||
buildResponsesCompatibilityHeaderCandidates,
|
||||
shouldDowngradeResponsesChatToMessages,
|
||||
shouldRetryResponsesCompatibility,
|
||||
} from './compatibility.js';
|
||||
|
||||
type CompatibilityRequest = {
|
||||
endpoint: CompatibilityEndpoint;
|
||||
path: string;
|
||||
headers: Record<string, string>;
|
||||
body: Record<string, unknown>;
|
||||
};
|
||||
|
||||
type UpstreamResponse = Exclude<EndpointRecoverResult, null>['upstream'];
|
||||
|
||||
type CreateResponsesEndpointStrategyInput = {
|
||||
isStream: boolean;
|
||||
requiresNativeResponsesFileUrl: boolean;
|
||||
dispatchRequest: (
|
||||
request: CompatibilityRequest,
|
||||
targetUrl?: string,
|
||||
) => Promise<UpstreamResponse>;
|
||||
};
|
||||
|
||||
export function createResponsesEndpointStrategy(input: CreateResponsesEndpointStrategyInput) {
|
||||
return {
|
||||
async tryRecover(ctx: EndpointAttemptContext): Promise<EndpointRecoverResult> {
|
||||
if (shouldRetryResponsesCompatibility({
|
||||
endpoint: ctx.request.endpoint,
|
||||
status: ctx.response.status,
|
||||
rawErrText: ctx.rawErrText,
|
||||
})) {
|
||||
const compatibilityBodies = buildResponsesCompatibilityBodies(ctx.request.body);
|
||||
const compatibilityHeaders = buildResponsesCompatibilityHeaderCandidates(
|
||||
ctx.request.headers,
|
||||
input.isStream,
|
||||
);
|
||||
|
||||
for (const compatibilityHeadersCandidate of compatibilityHeaders) {
|
||||
for (const compatibilityBody of compatibilityBodies) {
|
||||
const compatibilityRequest = {
|
||||
...ctx.request,
|
||||
headers: compatibilityHeadersCandidate,
|
||||
body: compatibilityBody,
|
||||
};
|
||||
const compatibilityResponse = await input.dispatchRequest(
|
||||
compatibilityRequest,
|
||||
ctx.targetUrl,
|
||||
);
|
||||
if (compatibilityResponse.ok) {
|
||||
return {
|
||||
upstream: compatibilityResponse,
|
||||
upstreamPath: compatibilityRequest.path,
|
||||
};
|
||||
}
|
||||
|
||||
ctx.request = compatibilityRequest;
|
||||
ctx.response = compatibilityResponse;
|
||||
ctx.rawErrText = await compatibilityResponse.text().catch(() => 'unknown error');
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (!isUnsupportedMediaTypeError(ctx.response.status, ctx.rawErrText)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const minimalRequest = {
|
||||
...ctx.request,
|
||||
headers: buildMinimalJsonHeadersForCompatibility({
|
||||
headers: ctx.request.headers,
|
||||
endpoint: ctx.request.endpoint,
|
||||
stream: input.isStream,
|
||||
}),
|
||||
};
|
||||
const minimalResponse = await input.dispatchRequest(minimalRequest, ctx.targetUrl);
|
||||
if (minimalResponse.ok) {
|
||||
return {
|
||||
upstream: minimalResponse,
|
||||
upstreamPath: minimalRequest.path,
|
||||
};
|
||||
}
|
||||
|
||||
ctx.request = minimalRequest;
|
||||
ctx.response = minimalResponse;
|
||||
ctx.rawErrText = await minimalResponse.text().catch(() => 'unknown error');
|
||||
return null;
|
||||
},
|
||||
shouldDowngrade(ctx: EndpointAttemptContext): boolean {
|
||||
if (input.requiresNativeResponsesFileUrl) return false;
|
||||
return (
|
||||
ctx.response.status >= 500
|
||||
|| isEndpointDowngradeError(ctx.response.status, ctx.rawErrText)
|
||||
|| shouldDowngradeResponsesChatToMessages(
|
||||
ctx.request.path,
|
||||
ctx.response.status,
|
||||
ctx.rawErrText,
|
||||
)
|
||||
);
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user