Preserve the active worktree snapshot on dev for follow-on integration

The root checkout already contained a large in-progress proxy, transformer,
and settings refactor. This commit captures that exact work on a dedicated
local dev branch without disturbing the original dirty worktree, so the
changes can be continued or split from dev instead of living only as
uncommitted state.

Constraint: The source checkout was already dirty on another branch, so the move had to preserve that worktree while excluding local .idea and .omx residue from the commit
Rejected: Switch the dirty root checkout directly to dev | risked disrupting the original branch and in-progress state
Confidence: medium
Scope-risk: broad
Reversibility: clean
Directive: Treat this as a transfer snapshot; fix the failing oauth/accountTokens/normalized/docs test regressions before merge or publication
Tested: git diff --check; npm run typecheck; npm run repo:drift-check
Not-tested: Clean test pass; npm test currently fails in oauth routes, accountTokens sync, shared normalized helpers, gemini route expectations, architecture-boundaries, and docs vitepress config
This commit is contained in:
cita-777
2026-04-17 19:41:14 +08:00
parent 6cca74b039
commit 88f40e91f7
133 changed files with 13780 additions and 5670 deletions
@@ -1,4 +1,5 @@
import { describe, expect, it } from 'vitest';
import { afterEach, describe, expect, it, vi } from 'vitest';
import { config } from '../../config.js';
import {
buildCodexSessionResponseStoreKey,
clearCodexSessionResponseId,
@@ -7,6 +8,10 @@ import {
setCodexSessionResponseId,
} from './codexSessionResponseStore.js';
afterEach(() => {
vi.useRealTimers();
});
describe('codexSessionResponseStore', () => {
it('evicts the oldest session id when the store exceeds the cap', () => {
resetCodexSessionResponseStore();
@@ -154,4 +159,39 @@ describe('codexSessionResponseStore', () => {
resetCodexSessionResponseStore();
});
it('expires remembered continuation ids after the sticky-session TTL window', () => {
vi.useFakeTimers();
vi.setSystemTime(new Date('2026-04-09T00:00:00Z'));
resetCodexSessionResponseStore();
setCodexSessionResponseId('session-expire', 'resp-expire');
expect(getCodexSessionResponseId('session-expire')).toBe('resp-expire');
vi.advanceTimersByTime(Math.max(5 * 60 * 1000, Math.trunc(config.proxyStickySessionTtlMs || 0)) + 1);
expect(getCodexSessionResponseId('session-expire')).toBeNull();
resetCodexSessionResponseStore();
});
it('refreshes the remembered continuation TTL when an active session reads it', () => {
vi.useFakeTimers();
vi.setSystemTime(new Date('2026-04-09T00:00:00Z'));
resetCodexSessionResponseStore();
const ttlMs = Math.max(5 * 60 * 1000, Math.trunc(config.proxyStickySessionTtlMs || 0));
setCodexSessionResponseId('session-touch', 'resp-touch');
vi.advanceTimersByTime(ttlMs - 1_000);
expect(getCodexSessionResponseId('session-touch')).toBe('resp-touch');
vi.advanceTimersByTime(2_000);
expect(getCodexSessionResponseId('session-touch')).toBe('resp-touch');
vi.advanceTimersByTime(ttlMs + 1);
expect(getCodexSessionResponseId('session-touch')).toBeNull();
resetCodexSessionResponseStore();
});
});
@@ -1,6 +1,14 @@
const MAX_CODEX_SESSION_RESPONSE_IDS = 10_000;
import { config } from '../../config.js';
const codexSessionResponseIds = new Map<string, string>();
const MAX_CODEX_SESSION_RESPONSE_IDS = 10_000;
const MIN_CODEX_SESSION_RESPONSE_TTL_MS = 5 * 60 * 1000;
type SessionResponseEntry = {
responseId: string;
updatedAtMs: number;
};
const codexSessionResponseIds = new Map<string, SessionResponseEntry>();
const SCOPED_SESSION_SEGMENT_PREFIX = 'session:';
const SCOPED_STORE_KEY_SEGMENT_PATTERN = /^(site|account|channel):\d+$/;
@@ -71,6 +79,33 @@ function normalizeSessionId(sessionId: string): string {
return sessionId.trim();
}
function getCodexSessionResponseTtlMs(): number {
return Math.max(
MIN_CODEX_SESSION_RESPONSE_TTL_MS,
Math.trunc(config.proxyStickySessionTtlMs || 0),
);
}
function isExpiredSessionResponseEntry(entry: SessionResponseEntry, nowMs = Date.now()): boolean {
return (entry.updatedAtMs + getCodexSessionResponseTtlMs()) <= nowMs;
}
function sweepExpiredSessionResponseIds(nowMs = Date.now()): void {
for (const [key, entry] of codexSessionResponseIds.entries()) {
if (!isExpiredSessionResponseEntry(entry, nowMs)) continue;
codexSessionResponseIds.delete(key);
}
}
function touchSessionResponseEntry(sessionId: string, nowMs = Date.now()): void {
for (const key of getSessionStoreKeys(sessionId)) {
const entry = codexSessionResponseIds.get(key);
if (!entry) continue;
entry.updatedAtMs = nowMs;
codexSessionResponseIds.set(key, entry);
}
}
export function buildCodexSessionResponseStoreKey(input: {
sessionId: string;
siteId?: number | null;
@@ -89,17 +124,23 @@ export function buildCodexSessionResponseStoreKey(input: {
}
export function getCodexSessionResponseId(sessionId: string): string | null {
const nowMs = Date.now();
sweepExpiredSessionResponseIds(nowMs);
const normalized = normalizeSessionId(sessionId);
if (!normalized) return null;
const direct = codexSessionResponseIds.get(normalized);
if (direct) return direct;
if (direct) {
touchSessionResponseEntry(normalized, nowMs);
return direct.responseId;
}
for (const fallbackKey of getFallbackSessionStoreKeys(normalized)) {
const fallback = codexSessionResponseIds.get(fallbackKey);
if (fallback) {
reconcileScopedSessionFallback(fallbackKey);
return fallback;
touchSessionResponseEntry(fallbackKey, nowMs);
return fallback.responseId;
}
}
@@ -107,6 +148,8 @@ export function getCodexSessionResponseId(sessionId: string): string | null {
}
export function setCodexSessionResponseId(sessionId: string, responseId: string): void {
const nowMs = Date.now();
sweepExpiredSessionResponseIds(nowMs);
const normalizedSessionId = normalizeSessionId(sessionId);
const normalizedResponseId = responseId.trim();
if (!normalizedSessionId || !normalizedResponseId) return;
@@ -117,7 +160,10 @@ export function setCodexSessionResponseId(sessionId: string, responseId: string)
if (codexSessionResponseIds.has(key)) {
codexSessionResponseIds.delete(key);
}
codexSessionResponseIds.set(key, normalizedResponseId);
codexSessionResponseIds.set(key, {
responseId: normalizedResponseId,
updatedAtMs: nowMs,
});
}
while (codexSessionResponseIds.size > MAX_CODEX_SESSION_RESPONSE_IDS) {
@@ -141,6 +141,67 @@ describe('codexWebsocketRuntime', () => {
await runtime.closeSession('exec-session-close');
});
it('preserves remembered continuation ids across websocket session closes and reconnects', async () => {
const { createCodexWebsocketRuntime } = await import('./codexWebsocketRuntime.js');
const runtime = createCodexWebsocketRuntime();
await runtime.sendRequest({
sessionId: 'exec-session-continue-after-close',
requestUrl: upstreamWsUrl,
headers: {
Authorization: 'Bearer oauth-access-token',
'OpenAI-Beta': 'responses_websockets=2026-02-06',
},
body: {
model: 'gpt-5.4',
input: [],
},
});
await runtime.closeSession('exec-session-continue-after-close');
const recovered = await runtime.sendRequest({
sessionId: 'exec-session-continue-after-close',
requestUrl: upstreamWsUrl,
headers: {
Authorization: 'Bearer oauth-access-token',
'OpenAI-Beta': 'responses_websockets=2026-02-06',
},
body: {
model: 'gpt-5.4',
input: [
{
id: 'tool_out_runtime_1',
type: 'function_call_output',
call_id: 'call_runtime_1',
output: '{"ok":true}',
},
],
},
});
expect(recovered.events[0]).toMatchObject({
type: 'response.completed',
response: { id: 'resp-2' },
});
expect(upstreamConnectionCount).toBe(2);
expect(upstreamRequests).toHaveLength(2);
expect(upstreamRequests[1]).toMatchObject({
type: 'response.create',
previous_response_id: 'resp-1',
input: [
{
id: 'tool_out_runtime_1',
type: 'function_call_output',
call_id: 'call_runtime_1',
output: '{"ok":true}',
},
],
});
await runtime.closeSession('exec-session-continue-after-close');
});
it('returns response.incomplete as a terminal websocket event without rejecting the session turn', async () => {
upstreamMessageHandler = (socket, parsed) => {
socket.send(JSON.stringify({
@@ -413,7 +474,8 @@ describe('codexWebsocketRuntime', () => {
message: 'account mismatch',
status: 502,
});
expect((error as CodexWebsocketRuntimeError).events).toEqual([
const runtimeError = error as InstanceType<typeof CodexWebsocketRuntimeError>;
expect(runtimeError.events).toEqual([
expect.objectContaining({
type: 'error',
error: expect.objectContaining({
@@ -337,7 +337,7 @@ async function sendSessionRequestAttempt(
socket.once('error', onError);
socket.once('close', onClose);
socket.send(JSON.stringify(buildCodexWebsocketRequestBody(input.body)), (error) => {
socket.send(JSON.stringify(buildCodexWebsocketRequestBody(input.body)), (error?: Error) => {
if (!error) return;
clearSessionSocket(session, socket);
rejectWith(error.message || 'failed to send upstream websocket request');
@@ -409,7 +409,9 @@ export function createCodexWebsocketRuntime(input?: {
await closeSocket(session.socket);
session.socket = null;
session.socketUrl = null;
clearCodexSessionResponseId(sessionId);
// Intentionally preserve the remembered previous_response_id across
// websocket reconnects. Closing a transport session should not sever the
// logical downstream continuation chain for the same session key.
},
async closeAllSessions(): Promise<void> {
@@ -0,0 +1,270 @@
import { zstdCompressSync } from 'node:zlib';
import { describe, expect, it } from 'vitest';
import { Response } from 'undici';
import { collectResponsesFinalPayloadFromSse } from './responsesSseFinal.js';
describe('collectResponsesFinalPayloadFromSse', () => {
it('treats event:error payloads as upstream failures', async () => {
const upstream = {
async text() {
return [
'event: error',
'data: {"error":{"message":"quota exceeded"},"type":"error"}',
'',
'data: [DONE]',
'',
].join('\n');
},
};
await expect(collectResponsesFinalPayloadFromSse(upstream, 'gpt-5.4'))
.rejects
.toThrow('quota exceeded');
});
it('prefers aggregated stream content when response.completed only carries an empty output array', async () => {
const upstream = {
async text() {
return [
'event: response.created',
'data: {"type":"response.created","response":{"id":"resp_empty_completed","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}',
'',
'event: response.output_item.added',
'data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_empty_completed","type":"message","role":"assistant","status":"in_progress","content":[]}}',
'',
'event: response.output_text.delta',
'data: {"type":"response.output_text.delta","output_index":0,"item_id":"msg_empty_completed","delta":"pong"}',
'',
'event: response.completed',
'data: {"type":"response.completed","response":{"id":"resp_empty_completed","model":"gpt-5.4","status":"completed","output":[],"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}',
'',
'data: [DONE]',
'',
].join('\n');
},
};
await expect(collectResponsesFinalPayloadFromSse(upstream, 'gpt-5.4'))
.resolves
.toMatchObject({
payload: {
id: 'resp_empty_completed',
status: 'completed',
output: [
{
id: 'msg_empty_completed',
type: 'message',
role: 'assistant',
status: 'completed',
content: [
{
type: 'output_text',
text: 'pong',
},
],
},
],
output_text: 'pong',
usage: {
input_tokens: 3,
output_tokens: 1,
total_tokens: 4,
},
},
});
});
it('decodes zstd-compressed responses SSE before aggregating the final payload', async () => {
const upstream = new Response(zstdCompressSync(Buffer.from([
'event: response.created',
'data: {"type":"response.created","response":{"id":"resp_zstd_completed","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}',
'',
'event: response.output_item.added',
'data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_zstd_completed","type":"message","role":"assistant","status":"in_progress","content":[]}}',
'',
'event: response.output_text.delta',
'data: {"type":"response.output_text.delta","output_index":0,"item_id":"msg_zstd_completed","delta":"你好,来自 zstd 聚合 SSE"}',
'',
'event: response.completed',
'data: {"type":"response.completed","response":{"id":"resp_zstd_completed","model":"gpt-5.4","status":"completed","output":[],"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}',
'',
'data: [DONE]',
'',
].join('\n'))), {
status: 200,
headers: {
'content-encoding': 'zstd',
'content-type': 'text/event-stream; charset=utf-8',
},
});
await expect(collectResponsesFinalPayloadFromSse(upstream, 'gpt-5.4'))
.resolves
.toMatchObject({
payload: {
id: 'resp_zstd_completed',
status: 'completed',
output_text: '你好,来自 zstd 聚合 SSE',
},
rawText: expect.stringContaining('你好,来自 zstd 聚合 SSE'),
});
});
it('returns response.incomplete payloads instead of treating them as upstream failures', async () => {
const upstream = {
async text() {
return [
'event: response.created',
'data: {"type":"response.created","response":{"id":"resp_incomplete_terminal","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}',
'',
'event: response.incomplete',
'data: {"type":"response.incomplete","response":{"id":"resp_incomplete_terminal","model":"gpt-5.4","status":"incomplete","output":[{"id":"msg_incomplete_1","type":"message","role":"assistant","status":"incomplete","content":[{"type":"output_text","text":"partial answer"}]}],"output_text":"partial answer","incomplete_details":{"reason":"max_output_tokens"},"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}',
'',
'data: [DONE]',
'',
].join('\n');
},
};
await expect(collectResponsesFinalPayloadFromSse(upstream, 'gpt-5.4'))
.resolves
.toMatchObject({
payload: {
id: 'resp_incomplete_terminal',
status: 'incomplete',
output_text: 'partial answer',
incomplete_details: {
reason: 'max_output_tokens',
},
},
});
});
it('returns response.incomplete terminal payloads instead of throwing', async () => {
const upstream = {
async text() {
return [
'event: response.created',
'data: {"type":"response.created","response":{"id":"resp_incomplete_1","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}',
'',
'event: response.output_text.delta',
'data: {"type":"response.output_text.delta","output_index":0,"item_id":"msg_incomplete_1","delta":"partial"}',
'',
'event: response.incomplete',
'data: {"type":"response.incomplete","response":{"id":"resp_incomplete_1","model":"gpt-5.4","status":"incomplete","output":[{"id":"msg_incomplete_1","type":"message","role":"assistant","status":"incomplete","content":[{"type":"output_text","text":"partial"}]}],"incomplete_details":{"reason":"max_output_tokens"},"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}',
'',
'data: [DONE]',
'',
].join('\n');
},
};
await expect(collectResponsesFinalPayloadFromSse(upstream, 'gpt-5.4'))
.resolves
.toMatchObject({
payload: {
id: 'resp_incomplete_1',
status: 'incomplete',
incomplete_details: {
reason: 'max_output_tokens',
},
output_text: 'partial',
},
});
});
it('treats response.incomplete as terminal even when only response.output carries the visible text', async () => {
const upstream = {
async text() {
return [
'event: response.created',
'data: {"type":"response.created","response":{"id":"resp_incomplete_output_only","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}',
'',
'event: response.incomplete',
'data: {"type":"response.incomplete","response":{"id":"resp_incomplete_output_only","model":"gpt-5.4","status":"incomplete","output":[{"id":"msg_incomplete_output_only","type":"message","role":"assistant","status":"incomplete","content":[{"type":"output_text","text":"partial from output"}]}],"incomplete_details":{"reason":"max_output_tokens"},"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}',
'',
'data: [DONE]',
'',
].join('\n');
},
};
await expect(collectResponsesFinalPayloadFromSse(upstream, 'gpt-5.4'))
.resolves
.toMatchObject({
payload: {
id: 'resp_incomplete_output_only',
status: 'incomplete',
output: [
{
id: 'msg_incomplete_output_only',
type: 'message',
role: 'assistant',
status: 'incomplete',
content: [
{
type: 'output_text',
text: 'partial from output',
},
],
},
],
output_text: 'partial from output',
incomplete_details: {
reason: 'max_output_tokens',
},
},
});
});
it('preserves incomplete item status when response.incomplete needs aggregate output repair', async () => {
const upstream = {
async text() {
return [
'event: response.created',
'data: {"type":"response.created","response":{"id":"resp_incomplete_repair","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}',
'',
'event: response.output_item.added',
'data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_incomplete_repair","type":"message","role":"assistant","status":"in_progress","content":[]}}',
'',
'event: response.output_text.delta',
'data: {"type":"response.output_text.delta","output_index":0,"item_id":"msg_incomplete_repair","delta":"partial repair"}',
'',
'event: response.incomplete',
'data: {"type":"response.incomplete","response":{"id":"resp_incomplete_repair","model":"gpt-5.4","status":"incomplete","output":[],"incomplete_details":{"reason":"max_output_tokens"},"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}',
'',
'data: [DONE]',
'',
].join('\n');
},
};
await expect(collectResponsesFinalPayloadFromSse(upstream, 'gpt-5.4'))
.resolves
.toMatchObject({
payload: {
id: 'resp_incomplete_repair',
status: 'incomplete',
output: [
{
id: 'msg_incomplete_repair',
type: 'message',
role: 'assistant',
status: 'incomplete',
content: [
{
type: 'output_text',
text: 'partial repair',
},
],
},
],
output_text: 'partial repair',
incomplete_details: {
reason: 'max_output_tokens',
},
},
});
});
});
@@ -0,0 +1,431 @@
import { openAiResponsesTransformer } from '../../transformers/openai/responses/index.js';
import { mergeProxyUsage, parseProxyUsage } from '../../services/proxyUsageParser.js';
import { readRuntimeResponseText } from '../executors/types.js';
type ResponsesTerminalStatus = 'completed' | 'incomplete';
function isRecord(value: unknown): value is Record<string, unknown> {
return !!value && typeof value === 'object';
}
function asTrimmedString(value: unknown): string {
return typeof value === 'string' ? value.trim() : '';
}
function parseResponsesSsePayload(data: string): Record<string, unknown> | null {
try {
const parsed = JSON.parse(data);
return isRecord(parsed) ? parsed : null;
} catch {
return null;
}
}
function getResponsesFailureMessage(payload: Record<string, unknown>): string {
if (isRecord(payload.error) && typeof payload.error.message === 'string' && payload.error.message.trim()) {
return payload.error.message.trim();
}
if (typeof payload.message === 'string' && payload.message.trim()) {
return payload.message.trim();
}
return 'upstream stream failed';
}
function hasMeaningfulMessageContent(content: unknown): boolean {
if (!Array.isArray(content)) return false;
return content.some((part) => {
if (!isRecord(part)) return false;
const partType = typeof part.type === 'string' ? part.type.trim().toLowerCase() : '';
if (partType === 'output_text' || partType === 'text') {
return typeof part.text === 'string' && part.text.length > 0;
}
return true;
});
}
function hasMeaningfulResponsesOutput(output: unknown): boolean {
if (!Array.isArray(output)) return false;
return output.some((item) => {
if (!isRecord(item)) return false;
const itemType = typeof item.type === 'string' ? item.type.trim().toLowerCase() : '';
if (itemType === 'message') {
return hasMeaningfulMessageContent(item.content);
}
if (itemType === 'reasoning') {
return (
(Array.isArray(item.summary) && item.summary.length > 0)
|| (typeof item.encrypted_content === 'string' && item.encrypted_content.trim().length > 0)
);
}
return itemType.length > 0;
});
}
function hasCompleteFinalResponsesPayload(payload: Record<string, unknown>): boolean {
return (
payload.object === 'response.compaction'
|| Array.isArray(payload.output)
|| Object.prototype.hasOwnProperty.call(payload, 'output_text')
);
}
function hasMeaningfulFinalResponsesPayload(payload: Record<string, unknown>): boolean {
if (payload.object === 'response.compaction') {
return Array.isArray(payload.output) && payload.output.length > 0;
}
if (typeof payload.output_text === 'string' && payload.output_text.length > 0) {
return true;
}
return hasMeaningfulResponsesOutput(payload.output);
}
function collectResponsesOutputText(payload: Record<string, unknown>): string {
const output = Array.isArray(payload.output) ? payload.output : [];
const parts: string[] = [];
for (const item of output) {
if (!isRecord(item)) continue;
if (asTrimmedString(item.type).toLowerCase() !== 'message') continue;
const content = Array.isArray(item.content) ? item.content : [];
for (const part of content) {
if (!isRecord(part)) continue;
const partType = asTrimmedString(part.type).toLowerCase();
const text = typeof part.text === 'string' ? part.text : '';
if ((partType === 'output_text' || partType === 'text') && text) {
parts.push(text);
}
}
}
return parts.join('');
}
function rememberStreamResponseEnvelope(
streamContext: ReturnType<typeof openAiResponsesTransformer.createStreamContext>,
payload: Record<string, unknown>,
): void {
const responsePayload = isRecord(payload.response) ? payload.response : payload;
if (typeof responsePayload.id === 'string' && responsePayload.id.trim().length > 0) {
streamContext.id = responsePayload.id;
}
if (typeof responsePayload.model === 'string' && responsePayload.model.trim().length > 0) {
streamContext.model = responsePayload.model;
}
const createdAt = (
typeof responsePayload.created_at === 'number' && Number.isFinite(responsePayload.created_at)
? responsePayload.created_at
: (typeof responsePayload.created === 'number' && Number.isFinite(responsePayload.created)
? responsePayload.created
: null)
);
if (createdAt !== null) {
streamContext.created = createdAt;
}
}
function ensureResponseId(rawId: string): string {
const trimmed = rawId.trim() || `resp_${Date.now()}`;
return trimmed.startsWith('resp_') ? trimmed : `resp_${trimmed}`;
}
function buildUsagePayload(usage: ReturnType<typeof parseProxyUsage>): Record<string, unknown> {
const payload: Record<string, unknown> = {
input_tokens: usage.promptTokens,
output_tokens: usage.completionTokens,
total_tokens: usage.totalTokens,
};
const inputDetails: Record<string, unknown> = {};
if ((usage.cacheReadTokens || 0) > 0) inputDetails.cached_tokens = usage.cacheReadTokens;
if ((usage.cacheCreationTokens || 0) > 0) inputDetails.cache_creation_tokens = usage.cacheCreationTokens;
if (Object.keys(inputDetails).length > 0) payload.input_tokens_details = inputDetails;
return payload;
}
function cloneAggregateOutputItem(
item: unknown,
terminalStatus: ResponsesTerminalStatus,
): Record<string, unknown> | null {
if (!isRecord(item)) return null;
const next = structuredClone(item);
const currentStatus = asTrimmedString(next.status).toLowerCase();
next.status = currentStatus && currentStatus !== 'in_progress'
? currentStatus
: terminalStatus;
return next;
}
function materializeTerminalPayloadFromAggregate(
aggregateState: ReturnType<typeof openAiResponsesTransformer.aggregator.createState>,
streamContext: ReturnType<typeof openAiResponsesTransformer.createStreamContext>,
usage: ReturnType<typeof parseProxyUsage>,
terminalStatus: ResponsesTerminalStatus,
): Record<string, unknown> {
const output = aggregateState.outputItems
.map((item) => cloneAggregateOutputItem(item, terminalStatus))
.filter((item): item is Record<string, unknown> => !!item);
const usagePayload = buildUsagePayload(usage);
const usageWithExtras = Object.keys(aggregateState.usageExtras).length > 0
? { ...usagePayload, ...structuredClone(aggregateState.usageExtras) }
: usagePayload;
return {
id: ensureResponseId(
asTrimmedString(aggregateState.responseId)
|| asTrimmedString(streamContext.id)
|| asTrimmedString(aggregateState.modelName),
),
object: 'response',
created_at: (
typeof aggregateState.createdAt === 'number' && Number.isFinite(aggregateState.createdAt)
? aggregateState.createdAt
: streamContext.created
) || Math.floor(Date.now() / 1000),
status: terminalStatus,
model: asTrimmedString(streamContext.model) || asTrimmedString(aggregateState.modelName),
output,
output_text: collectResponsesOutputText({ output }),
usage: usageWithExtras,
};
}
function enrichTerminalPayload(
payload: Record<string, unknown>,
aggregateState: ReturnType<typeof openAiResponsesTransformer.aggregator.createState>,
streamContext: ReturnType<typeof openAiResponsesTransformer.createStreamContext>,
usage: ReturnType<typeof parseProxyUsage>,
terminalStatus: ResponsesTerminalStatus,
): Record<string, unknown> {
const next = structuredClone(payload);
const materialized = materializeTerminalPayloadFromAggregate(
aggregateState,
streamContext,
usage,
terminalStatus,
);
if (!hasMeaningfulResponsesOutput(next.output) && materialized && hasMeaningfulResponsesOutput(materialized.output)) {
next.output = materialized.output;
}
const currentOutputText = typeof next.output_text === 'string' ? next.output_text : '';
if (!currentOutputText) {
const derivedOutputText = collectResponsesOutputText(next)
|| (materialized && typeof materialized.output_text === 'string' ? materialized.output_text : '');
if (derivedOutputText) {
next.output_text = derivedOutputText;
}
}
if (materialized && next.usage === undefined && materialized.usage !== undefined) {
next.usage = materialized.usage;
}
return next;
}
function mergeMissingResponsesTerminalFields(
payload: Record<string, unknown>,
fallbackPayload: Record<string, unknown> | null,
): Record<string, unknown> {
if (!fallbackPayload) return payload;
const merged = { ...payload };
if (merged.output === undefined && fallbackPayload.output !== undefined) {
merged.output = fallbackPayload.output;
}
if (
(typeof merged.output_text !== 'string' || merged.output_text.length === 0)
&& typeof fallbackPayload.output_text === 'string'
&& fallbackPayload.output_text.length > 0
) {
merged.output_text = fallbackPayload.output_text;
}
if (merged.object === undefined && fallbackPayload.object !== undefined) {
merged.object = fallbackPayload.object;
}
if (merged.created_at === undefined && fallbackPayload.created_at !== undefined) {
merged.created_at = fallbackPayload.created_at;
}
if (merged.usage === undefined && fallbackPayload.usage !== undefined) {
merged.usage = fallbackPayload.usage;
}
return merged;
}
export function looksLikeResponsesSseText(rawText: string): boolean {
const { events, rest } = openAiResponsesTransformer.pullSseEvents(rawText);
if (events.length === 0 || rest.trim().length > 0) return false;
return events.some((event) => {
if (event.data === '[DONE]') return true;
if (event.event === 'error' || event.event.startsWith('response.')) return true;
const payload = parseResponsesSsePayload(event.data);
const payloadType = typeof payload?.type === 'string' ? payload.type : '';
return payloadType === 'error' || payloadType.startsWith('response.');
});
}
export function createSingleChunkStreamReader(rawText: string): {
read(): Promise<{ done: boolean; value?: Uint8Array }>;
cancel(reason?: unknown): Promise<unknown>;
releaseLock(): void;
} {
const chunk = Buffer.from(rawText, 'utf8');
let done = false;
return {
async read() {
if (done) return { done: true };
done = true;
return { done: false, value: chunk };
},
async cancel() {
done = true;
return undefined;
},
releaseLock() {},
};
}
export function collectResponsesFinalPayloadFromSseText(
rawText: string,
modelName: string,
): { payload: Record<string, unknown>; rawText: string } {
const { events } = openAiResponsesTransformer.pullSseEvents(rawText);
const streamContext = openAiResponsesTransformer.createStreamContext(modelName);
const aggregateState = openAiResponsesTransformer.aggregator.createState(modelName);
let usage = {
promptTokens: 0,
completionTokens: 0,
totalTokens: 0,
cacheReadTokens: 0,
cacheCreationTokens: 0,
promptTokensIncludeCache: null as boolean | null,
};
let completedPayload: Record<string, unknown> | null = null;
let terminalStatus: ResponsesTerminalStatus = 'completed';
const captureCompletedPayloadFromEvent = (
eventType: string,
payload: Record<string, unknown>,
) => {
if (completedPayload) return;
if (eventType === 'response.failed' || eventType === 'error') {
throw new Error(getResponsesFailureMessage(payload));
}
if (eventType !== 'response.completed' && eventType !== 'response.incomplete') {
return;
}
terminalStatus = eventType === 'response.incomplete' ? 'incomplete' : 'completed';
if (isRecord(payload.response) && hasCompleteFinalResponsesPayload(payload.response)) {
completedPayload = payload.response;
return;
}
if (hasCompleteFinalResponsesPayload(payload)) {
completedPayload = payload;
}
};
const captureCompletedPayloadFromLines = (lines: string[]) => {
if (completedPayload) return;
const parsed = openAiResponsesTransformer.pullSseEvents(lines.join(''));
for (const event of parsed.events) {
if (event.data === '[DONE]') continue;
const payload = parseResponsesSsePayload(event.data);
if (!payload) continue;
const payloadType = typeof payload.type === 'string' ? payload.type : '';
captureCompletedPayloadFromEvent(payloadType || event.event, payload);
if (completedPayload) {
return;
}
}
};
for (const event of events) {
if (event.data === '[DONE]') continue;
const payload = parseResponsesSsePayload(event.data);
if (!payload) continue;
const payloadType = typeof payload.type === 'string' ? payload.type : '';
const eventType = payloadType || event.event;
rememberStreamResponseEnvelope(streamContext, payload);
usage = mergeProxyUsage(usage, parseProxyUsage(payload));
captureCompletedPayloadFromEvent(eventType, payload);
if (completedPayload) {
continue;
}
const normalizedEvent = openAiResponsesTransformer.transformStreamEvent(
payload,
streamContext,
modelName,
);
captureCompletedPayloadFromLines(openAiResponsesTransformer.aggregator.serialize({
state: aggregateState,
streamContext,
event: normalizedEvent,
usage,
}));
}
if (
completedPayload
&& !hasMeaningfulFinalResponsesPayload(completedPayload)
&& hasMeaningfulResponsesOutput(aggregateState.outputItems)
) {
completedPayload = mergeMissingResponsesTerminalFields(
completedPayload,
materializeTerminalPayloadFromAggregate(
aggregateState,
streamContext,
usage,
terminalStatus,
),
);
}
if (completedPayload) {
completedPayload = mergeMissingResponsesTerminalFields(
completedPayload,
materializeTerminalPayloadFromAggregate(
aggregateState,
streamContext,
usage,
terminalStatus,
),
);
}
if (!completedPayload) {
const materialized = materializeTerminalPayloadFromAggregate(
aggregateState,
streamContext,
usage,
terminalStatus,
);
if (materialized) {
completedPayload = materialized;
}
}
if (completedPayload) {
return {
payload: enrichTerminalPayload(
completedPayload,
aggregateState,
streamContext,
usage,
terminalStatus,
),
rawText,
};
}
throw new Error('stream disconnected before terminal responses event');
}
export async function collectResponsesFinalPayloadFromSse(
upstream: { text(): Promise<string>; headers?: { get(name: string): string | null } },
modelName: string,
): Promise<{ payload: Record<string, unknown>; rawText: string }> {
const rawText = typeof upstream.headers?.get === 'function'
? await readRuntimeResponseText(upstream as Parameters<typeof readRuntimeResponseText>[0])
: await upstream.text();
return collectResponsesFinalPayloadFromSseText(rawText, modelName);
}