forked from boos/BoosAPI
Merge pull request #237 from cita-777/codex/codex-review-fixes
fix: harden codex runtime routing and oauth recovery
This commit is contained in:
Cita
@@ -51,12 +51,21 @@ describe('buildConfig', () => {
|
||||
const config = buildConfig({});
|
||||
|
||||
expect(config.codexClientId).toBe('app_EMoamEEZ73f0CkXaXp7hrann');
|
||||
expect(config.codexResponsesWebsocketBeta).toBe('responses_websockets=2026-02-06');
|
||||
expect(config.claudeClientId).toBe('9d1c250a-e61b-44d9-88ed-5944d1962f5e');
|
||||
expect(config.claudeClientSecret).toBe('');
|
||||
expect(config.geminiCliClientId).toBe('681255809395-oo8ft2oprdrnp9e3aqf6av3hmdib135j.apps.googleusercontent.com');
|
||||
expect(config.geminiCliClientSecret).toBe('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');
|
||||
});
|
||||
|
||||
it('allows overriding the codex websocket beta gate from environment', () => {
|
||||
const config = buildConfig({
|
||||
CODEX_RESPONSES_WEBSOCKET_BETA: 'responses_websockets=2099-01-01',
|
||||
});
|
||||
|
||||
expect(config.codexResponsesWebsocketBeta).toBe('responses_websockets=2099-01-01');
|
||||
});
|
||||
|
||||
it('accepts JSON request bodies larger than Fastify default 1 MiB', async () => {
|
||||
const app = Fastify(buildFastifyOptions(buildConfig({})));
|
||||
const largeText = 'a'.repeat(2 * 1024 * 1024);
|
||||
|
||||
@@ -114,6 +114,7 @@ export function buildConfig(env: NodeJS.ProcessEnv) {
|
||||
proxyFileRetentionPruneIntervalMinutes: Math.max(1, Math.trunc(parseNumber(env.PROXY_FILE_RETENTION_PRUNE_INTERVAL_MINUTES, 60))),
|
||||
proxyErrorKeywords: parseCsvList(env.PROXY_ERROR_KEYWORDS),
|
||||
proxyEmptyContentFailEnabled: parseBoolean(env.PROXY_EMPTY_CONTENT_FAIL, false),
|
||||
codexResponsesWebsocketBeta: parseOptionalSecret(env.CODEX_RESPONSES_WEBSOCKET_BETA) || 'responses_websockets=2026-02-06',
|
||||
codexHeaderDefaults: {
|
||||
userAgent: parseOptionalSecret(env.CODEX_HEADER_DEFAULTS_USER_AGENT),
|
||||
betaFeatures: parseOptionalSecret(env.CODEX_HEADER_DEFAULTS_BETA_FEATURES),
|
||||
|
||||
@@ -4,7 +4,6 @@ import { config } from '../../config.js';
|
||||
|
||||
const CODEX_CLIENT_VERSION = '0.101.0';
|
||||
const CODEX_DEFAULT_USER_AGENT = 'codex_cli_rs/0.101.0 (Mac OS 26.0.1; arm64) Apple_Terminal/464';
|
||||
const CODEX_RESPONSES_WEBSOCKET_BETA = 'responses_websockets=2026-02-06';
|
||||
|
||||
function asTrimmedString(value: unknown): string {
|
||||
return typeof value === 'string' ? value.trim() : '';
|
||||
@@ -81,7 +80,7 @@ function buildCodexRuntimeHeaders(input: {
|
||||
const codexTurnMetadata = getInputHeader(input.baseHeaders, 'x-codex-turn-metadata');
|
||||
const timingMetrics = getInputHeader(input.baseHeaders, 'x-responsesapi-include-timing-metrics');
|
||||
const openAiBeta = getInputHeader(input.baseHeaders, 'openai-beta')
|
||||
|| (websocketTransport ? CODEX_RESPONSES_WEBSOCKET_BETA : null);
|
||||
|| (websocketTransport ? asTrimmedString(config.codexResponsesWebsocketBeta) : null);
|
||||
const explicitSessionId = asTrimmedString(input.explicitSessionId);
|
||||
const continuityKey = asTrimmedString(input.continuityKey);
|
||||
const sessionId = (
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
const CODEX_RESPONSES_WEBSOCKET_BETA = 'responses_websockets=2026-02-06';
|
||||
import { config } from '../../config.js';
|
||||
|
||||
function getHeaderValue(headers: Record<string, string>, key: string): string {
|
||||
const expected = key.trim().toLowerCase();
|
||||
@@ -11,13 +11,14 @@ function getHeaderValue(headers: Record<string, string>, key: string): string {
|
||||
|
||||
export function buildCodexWebsocketHandshakeHeaders(headers: Record<string, string>): Record<string, string> {
|
||||
const next = { ...headers };
|
||||
const websocketBeta = (config.codexResponsesWebsocketBeta || '').trim() || 'responses_websockets=2026-02-06';
|
||||
const openAiBeta = getHeaderValue(next, 'openai-beta').trim();
|
||||
if (!openAiBeta) {
|
||||
next['OpenAI-Beta'] = CODEX_RESPONSES_WEBSOCKET_BETA;
|
||||
next['OpenAI-Beta'] = websocketBeta;
|
||||
return next;
|
||||
}
|
||||
if (!openAiBeta.includes('responses_websockets=')) {
|
||||
next['OpenAI-Beta'] = `${openAiBeta},${CODEX_RESPONSES_WEBSOCKET_BETA}`;
|
||||
next['OpenAI-Beta'] = `${openAiBeta},${websocketBeta}`;
|
||||
}
|
||||
return next;
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ describe('codexWebsocketRuntime', () => {
|
||||
let upstreamWsUrl: string;
|
||||
let upstreamConnectionCount = 0;
|
||||
let upstreamRequests: Record<string, unknown>[] = [];
|
||||
let upstreamMessageHandler: (socket: import('ws').WebSocket, parsed: Record<string, unknown>, requestIndex: number) => void;
|
||||
|
||||
beforeAll(async () => {
|
||||
upstreamServer = new WebSocketServer({ port: 0 });
|
||||
@@ -15,22 +16,7 @@ describe('codexWebsocketRuntime', () => {
|
||||
socket.on('message', (payload) => {
|
||||
const parsed = JSON.parse(String(payload)) as Record<string, unknown>;
|
||||
upstreamRequests.push(parsed);
|
||||
const responseId = `resp-${upstreamRequests.length}`;
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.completed',
|
||||
response: {
|
||||
id: responseId,
|
||||
object: 'response',
|
||||
model: parsed.model || 'gpt-5.4',
|
||||
status: 'completed',
|
||||
output: [],
|
||||
usage: {
|
||||
input_tokens: 1,
|
||||
output_tokens: 1,
|
||||
total_tokens: 2,
|
||||
},
|
||||
},
|
||||
}));
|
||||
upstreamMessageHandler(socket, parsed, upstreamRequests.length);
|
||||
});
|
||||
});
|
||||
await new Promise<void>((resolve) => upstreamServer.once('listening', () => resolve()));
|
||||
@@ -41,6 +27,24 @@ describe('codexWebsocketRuntime', () => {
|
||||
beforeEach(() => {
|
||||
upstreamConnectionCount = 0;
|
||||
upstreamRequests = [];
|
||||
upstreamMessageHandler = (socket, parsed, requestIndex) => {
|
||||
const responseId = `resp-${requestIndex}`;
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.completed',
|
||||
response: {
|
||||
id: responseId,
|
||||
object: 'response',
|
||||
model: parsed.model || 'gpt-5.4',
|
||||
status: 'completed',
|
||||
output: [],
|
||||
usage: {
|
||||
input_tokens: 1,
|
||||
output_tokens: 1,
|
||||
total_tokens: 2,
|
||||
},
|
||||
},
|
||||
}));
|
||||
};
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
@@ -134,4 +138,99 @@ describe('codexWebsocketRuntime', () => {
|
||||
expect(upstreamConnectionCount).toBe(2);
|
||||
await runtime.closeSession('exec-session-close');
|
||||
});
|
||||
|
||||
it('treats response.incomplete as a terminal websocket failure instead of waiting for socket close', async () => {
|
||||
upstreamMessageHandler = (socket, parsed) => {
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.incomplete',
|
||||
response: {
|
||||
id: 'resp-incomplete',
|
||||
model: parsed.model || 'gpt-5.4',
|
||||
status: 'incomplete',
|
||||
incomplete_details: {
|
||||
reason: 'max_output_tokens',
|
||||
},
|
||||
},
|
||||
}));
|
||||
};
|
||||
|
||||
const { createCodexWebsocketRuntime, CodexWebsocketRuntimeError } = await import('./codexWebsocketRuntime.js');
|
||||
const runtime = createCodexWebsocketRuntime();
|
||||
|
||||
let error: unknown;
|
||||
try {
|
||||
await runtime.sendRequest({
|
||||
sessionId: 'exec-session-incomplete',
|
||||
requestUrl: upstreamWsUrl,
|
||||
headers: {
|
||||
Authorization: 'Bearer oauth-access-token',
|
||||
'OpenAI-Beta': 'responses_websockets=2026-02-06',
|
||||
},
|
||||
body: {
|
||||
model: 'gpt-5.4',
|
||||
input: [],
|
||||
},
|
||||
});
|
||||
} catch (caught) {
|
||||
error = caught;
|
||||
}
|
||||
|
||||
expect(error).toBeInstanceOf(CodexWebsocketRuntimeError);
|
||||
expect(error).toMatchObject({
|
||||
message: 'max_output_tokens',
|
||||
status: 502,
|
||||
});
|
||||
expect((error as CodexWebsocketRuntimeError).events).toEqual([
|
||||
expect.objectContaining({
|
||||
type: 'response.incomplete',
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it('treats top-level error frames as terminal websocket failures', async () => {
|
||||
upstreamMessageHandler = (socket) => {
|
||||
socket.send(JSON.stringify({
|
||||
type: 'error',
|
||||
error: {
|
||||
message: 'account mismatch',
|
||||
type: 'invalid_request_error',
|
||||
},
|
||||
}));
|
||||
};
|
||||
|
||||
const { createCodexWebsocketRuntime, CodexWebsocketRuntimeError } = await import('./codexWebsocketRuntime.js');
|
||||
const runtime = createCodexWebsocketRuntime();
|
||||
|
||||
let error: unknown;
|
||||
try {
|
||||
await runtime.sendRequest({
|
||||
sessionId: 'exec-session-error',
|
||||
requestUrl: upstreamWsUrl,
|
||||
headers: {
|
||||
Authorization: 'Bearer oauth-access-token',
|
||||
'OpenAI-Beta': 'responses_websockets=2026-02-06',
|
||||
},
|
||||
body: {
|
||||
model: 'gpt-5.4',
|
||||
input: [],
|
||||
},
|
||||
});
|
||||
} catch (caught) {
|
||||
error = caught;
|
||||
}
|
||||
|
||||
expect(error).toBeInstanceOf(CodexWebsocketRuntimeError);
|
||||
expect(error).toMatchObject({
|
||||
message: 'account mismatch',
|
||||
status: 502,
|
||||
});
|
||||
expect((error as CodexWebsocketRuntimeError).events).toEqual([
|
||||
expect.objectContaining({
|
||||
type: 'error',
|
||||
error: expect.objectContaining({
|
||||
message: 'account mismatch',
|
||||
}),
|
||||
}),
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -23,7 +23,57 @@ function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
|
||||
function isTerminalEvent(payload: Record<string, unknown>): boolean {
|
||||
const type = asTrimmedString(payload.type);
|
||||
return type === 'response.completed' || type === 'response.failed';
|
||||
return type === 'response.completed'
|
||||
|| type === 'response.failed'
|
||||
|| type === 'response.incomplete'
|
||||
|| type === 'error';
|
||||
}
|
||||
|
||||
function isFailureTerminalEvent(payload: Record<string, unknown>): boolean {
|
||||
const type = asTrimmedString(payload.type);
|
||||
return type === 'response.failed' || type === 'response.incomplete' || type === 'error';
|
||||
}
|
||||
|
||||
function asFiniteNumber(value: unknown): number | undefined {
|
||||
return typeof value === 'number' && Number.isFinite(value) ? value : undefined;
|
||||
}
|
||||
|
||||
function extractFailureTerminalStatus(payload: Record<string, unknown>): number {
|
||||
const response = isRecord(payload.response) ? payload.response : null;
|
||||
const responseError = response && isRecord(response.error) ? response.error : null;
|
||||
const topLevelError = isRecord(payload.error) ? payload.error : null;
|
||||
const candidates = [
|
||||
payload.status,
|
||||
payload.statusCode,
|
||||
payload.code,
|
||||
topLevelError?.status,
|
||||
topLevelError?.statusCode,
|
||||
topLevelError?.code,
|
||||
responseError?.status,
|
||||
responseError?.statusCode,
|
||||
responseError?.code,
|
||||
];
|
||||
for (const candidate of candidates) {
|
||||
const status = asFiniteNumber(candidate);
|
||||
if (status !== undefined) return status;
|
||||
}
|
||||
return 502;
|
||||
}
|
||||
|
||||
function extractTerminalErrorMessage(payload: Record<string, unknown>): string {
|
||||
const type = asTrimmedString(payload.type);
|
||||
if (type === 'error' && isRecord(payload.error)) {
|
||||
return asTrimmedString(payload.error.message) || 'upstream websocket error';
|
||||
}
|
||||
if ((type === 'response.failed' || type === 'response.incomplete') && isRecord(payload.response)) {
|
||||
if (isRecord(payload.response.error)) {
|
||||
return asTrimmedString(payload.response.error.message) || `upstream ${type}`;
|
||||
}
|
||||
if (isRecord(payload.response.incomplete_details)) {
|
||||
return asTrimmedString(payload.response.incomplete_details.reason) || `upstream ${type}`;
|
||||
}
|
||||
}
|
||||
return `upstream ${type || 'websocket error'}`;
|
||||
}
|
||||
|
||||
export class CodexWebsocketRuntimeError extends Error {
|
||||
@@ -212,6 +262,18 @@ async function sendSessionRequest(
|
||||
events.push(parsed);
|
||||
if (!isTerminalEvent(parsed)) return;
|
||||
if (settled) return;
|
||||
if (isFailureTerminalEvent(parsed)) {
|
||||
settled = true;
|
||||
cleanup();
|
||||
clearSessionSocket(session, socket);
|
||||
void closeSocket(socket);
|
||||
reject(new CodexWebsocketRuntimeError(extractTerminalErrorMessage(parsed), {
|
||||
events: [...events],
|
||||
status: extractFailureTerminalStatus(parsed),
|
||||
payload: parsed,
|
||||
}));
|
||||
return;
|
||||
}
|
||||
settled = true;
|
||||
cleanup();
|
||||
resolve({
|
||||
|
||||
@@ -234,7 +234,7 @@ export async function handleChatSurfaceRequest(
|
||||
dispatchRequest,
|
||||
});
|
||||
const tryRecover = async (ctx: Parameters<NonNullable<typeof endpointStrategy.tryRecover>>[0]) => {
|
||||
if (ctx.response.status === 401 && oauth) {
|
||||
if ((ctx.response.status === 401 || ctx.response.status === 403) && oauth) {
|
||||
try {
|
||||
const refreshed = await refreshOauthAccessTokenSingleflight(selected.account.id);
|
||||
selected.tokenValue = refreshed.accessToken;
|
||||
@@ -924,7 +924,7 @@ export async function handleClaudeCountTokensSurfaceRequest(
|
||||
}, resolveChannelProxyUrl(selected.site, selected.account.extraConfig)),
|
||||
});
|
||||
|
||||
if (upstream.status === 401 && oauth) {
|
||||
if ((upstream.status === 401 || upstream.status === 403) && oauth) {
|
||||
try {
|
||||
const refreshed = await refreshOauthAccessTokenSingleflight(selected.account.id);
|
||||
selected.tokenValue = refreshed.accessToken;
|
||||
|
||||
@@ -135,15 +135,21 @@ function carriesResponsesFileUrlInput(value: unknown): boolean {
|
||||
return Object.values(value).some((entry) => carriesResponsesFileUrlInput(entry));
|
||||
}
|
||||
|
||||
type UsageSummary = ReturnType<typeof parseProxyUsage>;
|
||||
|
||||
function deriveCodexExplicitSessionId(body: Record<string, unknown>): string | null {
|
||||
const promptCacheKey = typeof body.prompt_cache_key === 'string'
|
||||
? body.prompt_cache_key.trim()
|
||||
: '';
|
||||
return promptCacheKey || null;
|
||||
function shouldRefreshOauthResponsesRequest(input: {
|
||||
oauthProvider?: string;
|
||||
status: number;
|
||||
response: { headers: { get(name: string): string | null } };
|
||||
rawErrText: string;
|
||||
}): boolean {
|
||||
if (input.status === 401) return true;
|
||||
if (input.status !== 403 || input.oauthProvider !== 'codex') return false;
|
||||
const authenticate = input.response.headers.get('www-authenticate') || '';
|
||||
const combined = `${authenticate}\n${input.rawErrText || ''}`;
|
||||
return /\b(invalid_token|expired_token|expired|invalid|unauthorized|account mismatch|authentication)\b/i.test(combined);
|
||||
}
|
||||
|
||||
type UsageSummary = ReturnType<typeof parseProxyUsage>;
|
||||
|
||||
export async function handleOpenAiResponsesSurfaceRequest(
|
||||
request: FastifyRequest,
|
||||
reply: FastifyReply,
|
||||
@@ -274,7 +280,6 @@ export async function handleOpenAiResponsesSurfaceRequest(
|
||||
);
|
||||
const buildEndpointRequest = (endpoint: 'chat' | 'messages' | 'responses') => {
|
||||
const upstreamStream = isStream || (isCodexSite && endpoint === 'responses');
|
||||
const codexExplicitSessionId = deriveCodexExplicitSessionId(normalizedResponsesBody);
|
||||
const endpointRequest = buildUpstreamEndpointRequest({
|
||||
endpoint,
|
||||
modelName,
|
||||
@@ -289,7 +294,6 @@ export async function handleOpenAiResponsesSurfaceRequest(
|
||||
responsesOriginalBody: normalizedResponsesBody,
|
||||
downstreamHeaders: request.headers as Record<string, unknown>,
|
||||
providerHeaders: buildProviderHeaders(),
|
||||
codexExplicitSessionId,
|
||||
});
|
||||
const upstreamPath = (
|
||||
isCompactRequest && endpoint === 'responses'
|
||||
@@ -323,7 +327,12 @@ export async function handleOpenAiResponsesSurfaceRequest(
|
||||
dispatchRequest,
|
||||
});
|
||||
const tryRecover = async (ctx: Parameters<NonNullable<typeof endpointStrategy.tryRecover>>[0]) => {
|
||||
if (ctx.response.status === 401 && oauth) {
|
||||
if (oauth && shouldRefreshOauthResponsesRequest({
|
||||
oauthProvider: oauth.provider,
|
||||
status: ctx.response.status,
|
||||
response: ctx.response,
|
||||
rawErrText: ctx.rawErrText || '',
|
||||
})) {
|
||||
try {
|
||||
const refreshed = await refreshOauthAccessTokenSingleflight(selected.account.id);
|
||||
selected.tokenValue = refreshed.accessToken;
|
||||
|
||||
@@ -19,6 +19,16 @@ function buildJwt(payload: Record<string, unknown>) {
|
||||
return `${encode({ alg: 'none', typ: 'JWT' })}.${encode(payload)}.signature`;
|
||||
}
|
||||
|
||||
function createDeferred<T>() {
|
||||
let resolve!: (value: T | PromiseLike<T>) => void;
|
||||
let reject!: (reason?: unknown) => void;
|
||||
const promise = new Promise<T>((res, rej) => {
|
||||
resolve = res;
|
||||
reject = rej;
|
||||
});
|
||||
return { promise, resolve, reject };
|
||||
}
|
||||
|
||||
describe('oauth routes', { timeout: 15_000 }, () => {
|
||||
let app: FastifyInstance;
|
||||
let db: DbModule['db'];
|
||||
@@ -500,6 +510,265 @@ describe('oauth routes', { timeout: 15_000 }, () => {
|
||||
});
|
||||
});
|
||||
|
||||
it('keeps the existing codex connection intact when a rebind callback fails model discovery', async () => {
|
||||
const originalJwt = buildJwt({
|
||||
email: 'codex-existing@example.com',
|
||||
'https://api.openai.com/auth': {
|
||||
chatgpt_account_id: 'chatgpt-account-existing',
|
||||
chatgpt_plan_type: 'team',
|
||||
},
|
||||
});
|
||||
const site = await db.insert(schema.sites).values({
|
||||
name: 'ChatGPT Codex OAuth',
|
||||
url: 'https://chatgpt.com/backend-api/codex',
|
||||
platform: 'codex',
|
||||
status: 'active',
|
||||
}).returning().get();
|
||||
const existing = await db.insert(schema.accounts).values({
|
||||
siteId: site.id,
|
||||
username: 'codex-existing@example.com',
|
||||
accessToken: 'stable-access-token',
|
||||
apiToken: null,
|
||||
status: 'active',
|
||||
oauthProvider: 'codex',
|
||||
oauthAccountKey: 'chatgpt-account-existing',
|
||||
extraConfig: JSON.stringify({
|
||||
credentialMode: 'session',
|
||||
oauth: {
|
||||
provider: 'codex',
|
||||
accountId: 'chatgpt-account-existing',
|
||||
accountKey: 'chatgpt-account-existing',
|
||||
email: 'codex-existing@example.com',
|
||||
planType: 'team',
|
||||
refreshToken: 'stable-refresh-token',
|
||||
idToken: originalJwt,
|
||||
},
|
||||
}),
|
||||
}).returning().get();
|
||||
|
||||
const reboundJwt = buildJwt({
|
||||
email: 'codex-existing@example.com',
|
||||
'https://api.openai.com/auth': {
|
||||
chatgpt_account_id: 'chatgpt-account-rebound',
|
||||
chatgpt_plan_type: 'plus',
|
||||
},
|
||||
});
|
||||
fetchMock
|
||||
.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
status: 200,
|
||||
json: async () => ({
|
||||
access_token: 'rebound-access-token',
|
||||
refresh_token: 'rebound-refresh-token',
|
||||
id_token: reboundJwt,
|
||||
expires_in: 3600,
|
||||
token_type: 'Bearer',
|
||||
}),
|
||||
text: async () => JSON.stringify({ ok: true }),
|
||||
})
|
||||
.mockResolvedValueOnce({
|
||||
ok: false,
|
||||
status: 403,
|
||||
json: async () => ({ error: 'forbidden' }),
|
||||
text: async () => 'forbidden',
|
||||
});
|
||||
|
||||
const startResponse = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/api/oauth/providers/codex/start',
|
||||
headers: {
|
||||
host: 'metapi.example',
|
||||
'x-forwarded-proto': 'https',
|
||||
},
|
||||
payload: {
|
||||
accountId: existing.id,
|
||||
},
|
||||
});
|
||||
const startBody = startResponse.json() as { state: string };
|
||||
|
||||
const callbackResponse = await app.inject({
|
||||
method: 'POST',
|
||||
url: `/api/oauth/sessions/${encodeURIComponent(startBody.state)}/manual-callback`,
|
||||
payload: {
|
||||
callbackUrl: `http://localhost:1455/auth/callback?state=${encodeURIComponent(startBody.state)}&code=oauth-code-rebind-fail`,
|
||||
},
|
||||
});
|
||||
expect(callbackResponse.statusCode).toBe(500);
|
||||
|
||||
const stored = await db.select().from(schema.accounts).where(eq(schema.accounts.id, existing.id)).get();
|
||||
expect(stored).toMatchObject({
|
||||
id: existing.id,
|
||||
username: 'codex-existing@example.com',
|
||||
accessToken: 'stable-access-token',
|
||||
oauthAccountKey: 'chatgpt-account-existing',
|
||||
});
|
||||
expect(JSON.parse(stored?.extraConfig || '{}')).toMatchObject({
|
||||
oauth: {
|
||||
accountId: 'chatgpt-account-existing',
|
||||
refreshToken: 'stable-refresh-token',
|
||||
idToken: originalJwt,
|
||||
},
|
||||
});
|
||||
|
||||
const accounts = await db.select().from(schema.accounts).all();
|
||||
expect(accounts).toHaveLength(1);
|
||||
expect(accounts[0]?.oauthAccountKey).toBe('chatgpt-account-existing');
|
||||
});
|
||||
|
||||
it('keeps the existing codex account non-active while rebind model discovery is still pending', async () => {
|
||||
const originalJwt = buildJwt({
|
||||
email: 'codex-existing@example.com',
|
||||
'https://api.openai.com/auth': {
|
||||
chatgpt_account_id: 'chatgpt-account-existing',
|
||||
chatgpt_plan_type: 'team',
|
||||
},
|
||||
});
|
||||
const site = await db.insert(schema.sites).values({
|
||||
name: 'ChatGPT Codex OAuth',
|
||||
url: 'https://chatgpt.com/backend-api/codex',
|
||||
platform: 'codex',
|
||||
status: 'active',
|
||||
}).returning().get();
|
||||
const existing = await db.insert(schema.accounts).values({
|
||||
siteId: site.id,
|
||||
username: 'codex-existing@example.com',
|
||||
accessToken: 'stable-access-token',
|
||||
apiToken: null,
|
||||
status: 'active',
|
||||
oauthProvider: 'codex',
|
||||
oauthAccountKey: 'chatgpt-account-existing',
|
||||
extraConfig: JSON.stringify({
|
||||
credentialMode: 'session',
|
||||
oauth: {
|
||||
provider: 'codex',
|
||||
accountId: 'chatgpt-account-existing',
|
||||
accountKey: 'chatgpt-account-existing',
|
||||
email: 'codex-existing@example.com',
|
||||
planType: 'team',
|
||||
refreshToken: 'stable-refresh-token',
|
||||
idToken: originalJwt,
|
||||
},
|
||||
}),
|
||||
}).returning().get();
|
||||
|
||||
const reboundJwt = buildJwt({
|
||||
email: 'codex-existing@example.com',
|
||||
'https://api.openai.com/auth': {
|
||||
chatgpt_account_id: 'chatgpt-account-rebound',
|
||||
chatgpt_plan_type: 'plus',
|
||||
},
|
||||
});
|
||||
const discoveryGate = createDeferred<ResponseLike>();
|
||||
fetchMock
|
||||
.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
status: 200,
|
||||
json: async () => ({
|
||||
access_token: 'rebound-access-token',
|
||||
refresh_token: 'rebound-refresh-token',
|
||||
id_token: reboundJwt,
|
||||
expires_in: 3600,
|
||||
token_type: 'Bearer',
|
||||
}),
|
||||
text: async () => JSON.stringify({ ok: true }),
|
||||
})
|
||||
.mockImplementationOnce(() => discoveryGate.promise);
|
||||
|
||||
const startResponse = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/api/oauth/providers/codex/start',
|
||||
headers: {
|
||||
host: 'metapi.example',
|
||||
'x-forwarded-proto': 'https',
|
||||
},
|
||||
payload: {
|
||||
accountId: existing.id,
|
||||
},
|
||||
});
|
||||
const startBody = startResponse.json() as { state: string };
|
||||
|
||||
const callbackPromise = app.inject({
|
||||
method: 'POST',
|
||||
url: `/api/oauth/sessions/${encodeURIComponent(startBody.state)}/manual-callback`,
|
||||
payload: {
|
||||
callbackUrl: `http://localhost:1455/auth/callback?state=${encodeURIComponent(startBody.state)}&code=oauth-code-rebind-pending`,
|
||||
},
|
||||
});
|
||||
|
||||
while (fetchMock.mock.calls.length < 2) {
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
}
|
||||
|
||||
const pendingRow = await db.select().from(schema.accounts).where(eq(schema.accounts.id, existing.id)).get();
|
||||
expect(pendingRow?.status).toBe('disabled');
|
||||
|
||||
discoveryGate.resolve({
|
||||
ok: true,
|
||||
status: 200,
|
||||
json: async () => ({
|
||||
models: [{ id: 'gpt-5.4' }],
|
||||
}),
|
||||
text: async () => JSON.stringify({ ok: true }),
|
||||
} as ResponseLike);
|
||||
|
||||
const callbackResponse = await callbackPromise;
|
||||
expect(callbackResponse.statusCode).toBe(200);
|
||||
|
||||
const stored = await db.select().from(schema.accounts).where(eq(schema.accounts.id, existing.id)).get();
|
||||
expect(stored).toMatchObject({
|
||||
id: existing.id,
|
||||
status: 'active',
|
||||
accessToken: 'rebound-access-token',
|
||||
oauthAccountKey: 'chatgpt-account-rebound',
|
||||
});
|
||||
});
|
||||
|
||||
it('fails codex oauth onboarding when token exchange does not expose chatgpt_account_id', async () => {
|
||||
const jwt = buildJwt({
|
||||
email: 'codex-no-account@example.com',
|
||||
'https://api.openai.com/auth': {
|
||||
chatgpt_plan_type: 'plus',
|
||||
},
|
||||
});
|
||||
fetchMock.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
status: 200,
|
||||
json: async () => ({
|
||||
access_token: 'oauth-access-token',
|
||||
refresh_token: 'oauth-refresh-token',
|
||||
id_token: jwt,
|
||||
expires_in: 3600,
|
||||
token_type: 'Bearer',
|
||||
}),
|
||||
text: async () => JSON.stringify({ ok: true }),
|
||||
});
|
||||
|
||||
const startResponse = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/api/oauth/providers/codex/start',
|
||||
headers: {
|
||||
host: 'metapi.example',
|
||||
'x-forwarded-proto': 'https',
|
||||
},
|
||||
});
|
||||
const startBody = startResponse.json() as { state: string };
|
||||
|
||||
const callbackResponse = await app.inject({
|
||||
method: 'POST',
|
||||
url: `/api/oauth/sessions/${encodeURIComponent(startBody.state)}/manual-callback`,
|
||||
payload: {
|
||||
callbackUrl: `http://localhost:1455/auth/callback?state=${encodeURIComponent(startBody.state)}&code=oauth-code-no-account-id`,
|
||||
},
|
||||
});
|
||||
expect(callbackResponse.statusCode).toBe(500);
|
||||
expect(callbackResponse.json()).toMatchObject({
|
||||
message: expect.stringContaining('chatgpt_account_id'),
|
||||
});
|
||||
|
||||
const accounts = await db.select().from(schema.accounts).all();
|
||||
expect(accounts).toEqual([]);
|
||||
});
|
||||
|
||||
it('marks gemini oauth session as error when token exchange fails before account persistence', async () => {
|
||||
fetchMock.mockResolvedValueOnce({
|
||||
ok: false,
|
||||
|
||||
@@ -361,4 +361,47 @@ describe('chat proxy codex oauth compatibility', () => {
|
||||
expect(secondOptions.headers.Authorization).toBe('Bearer fresh-access-token');
|
||||
expect(response.json()?.choices?.[0]?.message?.content).toBe('ok after refresh');
|
||||
});
|
||||
|
||||
it('retries oauth chat requests after a 403 auth failure', async () => {
|
||||
fetchMock
|
||||
.mockResolvedValueOnce(new Response(JSON.stringify({
|
||||
error: { message: 'forbidden account mismatch', type: 'invalid_request_error' },
|
||||
}), {
|
||||
status: 403,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}))
|
||||
.mockResolvedValueOnce(new Response(JSON.stringify({
|
||||
id: 'chatcmpl_refreshed_403',
|
||||
object: 'chat.completion',
|
||||
created: 1706000000,
|
||||
model: 'gpt-4o-mini',
|
||||
choices: [
|
||||
{
|
||||
index: 0,
|
||||
message: { role: 'assistant', content: 'ok after forbidden refresh' },
|
||||
finish_reason: 'stop',
|
||||
},
|
||||
],
|
||||
usage: { prompt_tokens: 4, completion_tokens: 2, total_tokens: 6 },
|
||||
}), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/chat/completions',
|
||||
payload: {
|
||||
model: 'gpt-4o-mini',
|
||||
messages: [{ role: 'user', content: 'hello oauth' }],
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenCalledWith(33);
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
const [, secondOptions] = fetchMock.mock.calls[1] as [string, any];
|
||||
expect(secondOptions.headers.Authorization).toBe('Bearer fresh-access-token');
|
||||
expect(response.json()?.choices?.[0]?.message?.content).toBe('ok after forbidden refresh');
|
||||
});
|
||||
});
|
||||
|
||||
@@ -186,6 +186,7 @@ describe('responses proxy codex oauth refresh', () => {
|
||||
url: '/v1/responses',
|
||||
headers: {
|
||||
'user-agent': 'CodexClient/1.0',
|
||||
'Chatgpt-Account-Id': 'spoofed-account',
|
||||
},
|
||||
payload: {
|
||||
model: 'gpt-5.2-codex',
|
||||
@@ -214,6 +215,73 @@ describe('responses proxy codex oauth refresh', () => {
|
||||
expect(response.json()?.output_text).toContain('ok after codex token refresh');
|
||||
});
|
||||
|
||||
it('refreshes codex oauth token and retries the same responses request on 403', async () => {
|
||||
fetchMock
|
||||
.mockResolvedValueOnce(new Response(JSON.stringify({
|
||||
error: { message: 'forbidden account mismatch', type: 'invalid_request_error' },
|
||||
}), {
|
||||
status: 403,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}))
|
||||
.mockResolvedValueOnce(new Response(JSON.stringify({
|
||||
id: 'resp_codex_refreshed_403',
|
||||
object: 'response',
|
||||
model: 'gpt-5.2-codex',
|
||||
status: 'completed',
|
||||
output_text: 'ok after codex forbidden refresh',
|
||||
usage: { input_tokens: 4, output_tokens: 2, total_tokens: 6 },
|
||||
}), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/responses',
|
||||
headers: {
|
||||
'user-agent': 'CodexClient/1.0',
|
||||
},
|
||||
payload: {
|
||||
model: 'gpt-5.2-codex',
|
||||
input: 'hello codex',
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenCalledWith(33);
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
const [, secondOptions] = fetchMock.mock.calls[1] as [string, any];
|
||||
expect(secondOptions.headers.Authorization).toBe('Bearer fresh-access-token');
|
||||
expect(response.json()?.output_text).toContain('ok after codex forbidden refresh');
|
||||
});
|
||||
|
||||
it('does not refresh codex oauth token on non-auth 403 responses', async () => {
|
||||
fetchMock.mockResolvedValueOnce(new Response(JSON.stringify({
|
||||
error: { message: 'quota exceeded for workspace', type: 'usage_limit_reached' },
|
||||
}), {
|
||||
status: 403,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/responses',
|
||||
payload: {
|
||||
model: 'gpt-5.2-codex',
|
||||
input: 'hello codex',
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(403);
|
||||
expect(refreshOauthAccessTokenSingleflightMock).not.toHaveBeenCalled();
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
expect(response.json()).toMatchObject({
|
||||
error: {
|
||||
message: expect.stringContaining('quota exceeded for workspace'),
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it('retries oauth responses requests with a normalized upstream URL after refresh', async () => {
|
||||
selectChannelMock.mockReturnValue({
|
||||
channel: { id: 11, routeId: 22 },
|
||||
@@ -316,7 +384,7 @@ describe('responses proxy codex oauth refresh', () => {
|
||||
]);
|
||||
});
|
||||
|
||||
it('preserves explicit prompt_cache_key for codex responses requests', async () => {
|
||||
it('preserves explicit prompt_cache_key for codex responses requests without converting it into codex session headers', async () => {
|
||||
fetchMock.mockResolvedValue(new Response(JSON.stringify({
|
||||
id: 'resp_codex_with_cache_key',
|
||||
object: 'response',
|
||||
@@ -344,8 +412,8 @@ describe('responses proxy codex oauth refresh', () => {
|
||||
|
||||
const [, options] = fetchMock.mock.calls[0] as [string, any];
|
||||
const forwardedBody = JSON.parse(options.body);
|
||||
expect(options.headers.Session_id || options.headers.session_id).toBe('codex-cache-123');
|
||||
expect(options.headers.Conversation_id || options.headers.conversation_id).toBe('codex-cache-123');
|
||||
expect(String(options.headers.Session_id || options.headers.session_id || '')).toMatch(/^[0-9a-f-]{36}$/i);
|
||||
expect(options.headers.Conversation_id || options.headers.conversation_id).toBeUndefined();
|
||||
expect(forwardedBody.prompt_cache_key).toBe('codex-cache-123');
|
||||
});
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@ import { createServer, type Server } from 'node:http';
|
||||
import { AddressInfo } from 'node:net';
|
||||
import WebSocket, { WebSocketServer } from 'ws';
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import { config } from '../../config.js';
|
||||
|
||||
const fetchMock = vi.fn();
|
||||
const selectChannelMock = vi.fn();
|
||||
@@ -10,6 +11,8 @@ const selectNextChannelMock = vi.fn();
|
||||
const previewSelectedChannelMock = vi.fn();
|
||||
const recordSuccessMock = vi.fn();
|
||||
const recordFailureMock = vi.fn();
|
||||
const authorizeDownstreamTokenMock = vi.fn();
|
||||
const consumeManagedKeyRequestMock = vi.fn();
|
||||
const refreshModelsAndRebuildRoutesMock = vi.fn();
|
||||
const reportProxyAllFailedMock = vi.fn();
|
||||
const reportTokenExpiredMock = vi.fn();
|
||||
@@ -47,6 +50,22 @@ vi.mock('../../services/alertService.js', () => ({
|
||||
reportTokenExpired: (...args: unknown[]) => reportTokenExpiredMock(...args),
|
||||
}));
|
||||
|
||||
vi.mock('../../services/downstreamApiKeyService.js', () => ({
|
||||
authorizeDownstreamToken: (...args: unknown[]) => authorizeDownstreamTokenMock(...args),
|
||||
consumeManagedKeyRequest: (...args: unknown[]) => consumeManagedKeyRequestMock(...args),
|
||||
isModelAllowedByPolicyOrAllowedRoutes: async (
|
||||
model: string,
|
||||
policy: { supportedModels?: string[]; allowedRouteIds?: number[]; denyAllWhenEmpty?: boolean },
|
||||
) => {
|
||||
const supportedModels = Array.isArray(policy?.supportedModels) ? policy.supportedModels : [];
|
||||
const allowedRouteIds = Array.isArray(policy?.allowedRouteIds) ? policy.allowedRouteIds : [];
|
||||
if (supportedModels.length === 0 && allowedRouteIds.length === 0) {
|
||||
return policy?.denyAllWhenEmpty === true ? false : true;
|
||||
}
|
||||
return supportedModels.includes(model);
|
||||
},
|
||||
}));
|
||||
|
||||
vi.mock('../../services/alertRules.js', () => ({
|
||||
isTokenExpiredError: () => false,
|
||||
}));
|
||||
@@ -175,7 +194,31 @@ function waitForSocketMessages(socket: WebSocket, count: number, timeoutMs = 100
|
||||
});
|
||||
}
|
||||
|
||||
function createDeferred<T>() {
|
||||
let resolve!: (value: T | PromiseLike<T>) => void;
|
||||
let reject!: (reason?: unknown) => void;
|
||||
const promise = new Promise<T>((res, rej) => {
|
||||
resolve = res;
|
||||
reject = rej;
|
||||
});
|
||||
return { promise, resolve, reject };
|
||||
}
|
||||
|
||||
function createClientSocket(baseUrl: string, headers: Record<string, string> = {}) {
|
||||
return new WebSocket(`${baseUrl}/v1/responses`, {
|
||||
headers: {
|
||||
Authorization: 'Bearer sk-global-proxy-token',
|
||||
...headers,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function createClientSocketForPath(path: string, headers: Record<string, string> = {}) {
|
||||
return new WebSocket(path, { headers });
|
||||
}
|
||||
|
||||
describe('responses websocket transport', () => {
|
||||
const originalCodexResponsesWebsocketBeta = config.codexResponsesWebsocketBeta;
|
||||
let app: FastifyInstance;
|
||||
let baseUrl: string;
|
||||
let upstreamServer: WebSocketServer;
|
||||
@@ -186,6 +229,9 @@ describe('responses websocket transport', () => {
|
||||
let upstreamMessageHandler: (socket: WebSocket, parsed: Record<string, unknown>, requestIndex: number) => void;
|
||||
let rejectedUpgradeServer: Server;
|
||||
let rejectedUpgradeSiteUrl: string;
|
||||
let rejectedUpgradeStatus: number;
|
||||
let rejectedUpgradeStatusText: string;
|
||||
let rejectedUpgradeBody: string;
|
||||
|
||||
beforeAll(async () => {
|
||||
const { responsesProxyRoute } = await import('./responses.js');
|
||||
@@ -214,13 +260,14 @@ describe('responses websocket transport', () => {
|
||||
|
||||
rejectedUpgradeServer = createServer();
|
||||
rejectedUpgradeServer.on('upgrade', (_request, socket) => {
|
||||
const body = rejectedUpgradeBody;
|
||||
socket.write(
|
||||
'HTTP/1.1 426 Upgrade Required\r\n'
|
||||
`HTTP/1.1 ${rejectedUpgradeStatus} ${rejectedUpgradeStatusText}\r\n`
|
||||
+ 'Content-Type: text/plain\r\n'
|
||||
+ 'Content-Length: 16\r\n'
|
||||
+ `Content-Length: ${Buffer.byteLength(body)}\r\n`
|
||||
+ 'Connection: close\r\n'
|
||||
+ '\r\n'
|
||||
+ 'Upgrade Required',
|
||||
+ body,
|
||||
);
|
||||
socket.destroy();
|
||||
});
|
||||
@@ -236,6 +283,8 @@ describe('responses websocket transport', () => {
|
||||
previewSelectedChannelMock.mockReset();
|
||||
recordSuccessMock.mockReset();
|
||||
recordFailureMock.mockReset();
|
||||
authorizeDownstreamTokenMock.mockReset();
|
||||
consumeManagedKeyRequestMock.mockReset();
|
||||
refreshModelsAndRebuildRoutesMock.mockReset();
|
||||
reportProxyAllFailedMock.mockReset();
|
||||
reportTokenExpiredMock.mockReset();
|
||||
@@ -249,6 +298,21 @@ describe('responses websocket transport', () => {
|
||||
upstreamConnectionCount = 0;
|
||||
upstreamUpgradeHeaders = {};
|
||||
upstreamRequests = [];
|
||||
(config as any).codexResponsesWebsocketBeta = originalCodexResponsesWebsocketBeta;
|
||||
rejectedUpgradeStatus = 426;
|
||||
rejectedUpgradeStatusText = 'Upgrade Required';
|
||||
rejectedUpgradeBody = 'Upgrade Required';
|
||||
authorizeDownstreamTokenMock.mockResolvedValue({
|
||||
ok: true,
|
||||
source: 'global',
|
||||
token: 'sk-global-proxy-token',
|
||||
key: null,
|
||||
policy: {
|
||||
supportedModels: [],
|
||||
allowedRouteIds: [],
|
||||
siteWeightMultipliers: {},
|
||||
},
|
||||
});
|
||||
upstreamMessageHandler = (socket, parsed, requestIndex) => {
|
||||
const responseId = `resp_upstream_${requestIndex}`;
|
||||
socket.send(JSON.stringify({
|
||||
@@ -331,7 +395,7 @@ describe('responses websocket transport', () => {
|
||||
}));
|
||||
};
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
const messagesPromise = waitForSocketMessages(socket, 4);
|
||||
|
||||
@@ -362,10 +426,8 @@ describe('responses websocket transport', () => {
|
||||
});
|
||||
|
||||
it('echoes x-codex-turn-state on websocket upgrade responses', async () => {
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`, {
|
||||
headers: {
|
||||
'x-codex-turn-state': 'turn-state-123',
|
||||
},
|
||||
const socket = createClientSocket(baseUrl, {
|
||||
'x-codex-turn-state': 'turn-state-123',
|
||||
});
|
||||
|
||||
const [upgrade] = await Promise.all([
|
||||
@@ -378,17 +440,16 @@ describe('responses websocket transport', () => {
|
||||
});
|
||||
|
||||
it('reuses one upstream codex websocket session across sequential websocket turns', async () => {
|
||||
(config as any).codexResponsesWebsocketBeta = 'responses_websockets=2099-01-01';
|
||||
const selectedChannel = createSelectedChannel({
|
||||
siteUrl: upstreamSiteUrl,
|
||||
});
|
||||
selectChannelMock.mockReturnValue(selectedChannel);
|
||||
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`, {
|
||||
headers: {
|
||||
'x-codex-turn-state': 'turn-state-123',
|
||||
'x-codex-beta-features': 'feature-a,feature-b',
|
||||
},
|
||||
const socket = createClientSocket(baseUrl, {
|
||||
'x-codex-turn-state': 'turn-state-123',
|
||||
'x-codex-beta-features': 'feature-a,feature-b',
|
||||
});
|
||||
await waitForSocketOpen(socket);
|
||||
const firstMessagesPromise = waitForSocketMessages(socket, 1);
|
||||
@@ -427,7 +488,7 @@ describe('responses websocket transport', () => {
|
||||
});
|
||||
expect(upstreamUpgradeHeaders['x-codex-turn-state']).toBe('turn-state-123');
|
||||
expect(upstreamUpgradeHeaders['x-codex-beta-features']).toBe('feature-a,feature-b');
|
||||
expect(upstreamUpgradeHeaders['openai-beta']).toContain('responses_websockets=');
|
||||
expect(upstreamUpgradeHeaders['openai-beta']).toContain('responses_websockets=2099-01-01');
|
||||
});
|
||||
|
||||
it('falls back to the HTTP responses executor when the upstream codex websocket upgrade returns 426', async () => {
|
||||
@@ -442,7 +503,7 @@ describe('responses websocket transport', () => {
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
const messagesPromise = waitForSocketMessages(socket, 1);
|
||||
|
||||
@@ -460,6 +521,128 @@ describe('responses websocket transport', () => {
|
||||
expect(messages[0]?.response?.id).toBe('resp_http_fallback');
|
||||
});
|
||||
|
||||
it('falls back to the HTTP responses executor when the upstream codex websocket upgrade returns 401', async () => {
|
||||
rejectedUpgradeStatus = 401;
|
||||
rejectedUpgradeStatusText = 'Unauthorized';
|
||||
rejectedUpgradeBody = JSON.stringify({
|
||||
error: {
|
||||
message: 'expired token',
|
||||
type: 'invalid_request_error',
|
||||
},
|
||||
});
|
||||
const selectedChannel = createSelectedChannel({
|
||||
siteUrl: rejectedUpgradeSiteUrl,
|
||||
});
|
||||
selectChannelMock.mockReturnValue(selectedChannel);
|
||||
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
|
||||
fetchMock.mockResolvedValueOnce(createSseResponse([
|
||||
'event: response.completed\n',
|
||||
'data: {"type":"response.completed","response":{"id":"resp_http_fallback_401","model":"gpt-5.4","status":"completed","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}\n\n',
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
const messagesPromise = waitForSocketMessages(socket, 1);
|
||||
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.create',
|
||||
model: 'gpt-5.4',
|
||||
input: [],
|
||||
}));
|
||||
|
||||
const messages = await messagesPromise;
|
||||
socket.close();
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
expect(messages.map((message) => message.type)).toEqual(['response.completed']);
|
||||
expect(messages[0]?.response?.id).toBe('resp_http_fallback_401');
|
||||
});
|
||||
|
||||
it('preserves query parameter auth when websocket transport falls back to the HTTP responses route', async () => {
|
||||
rejectedUpgradeStatus = 401;
|
||||
rejectedUpgradeStatusText = 'Unauthorized';
|
||||
rejectedUpgradeBody = JSON.stringify({
|
||||
error: {
|
||||
message: 'expired token',
|
||||
type: 'invalid_request_error',
|
||||
},
|
||||
});
|
||||
const selectedChannel = createSelectedChannel({
|
||||
siteUrl: rejectedUpgradeSiteUrl,
|
||||
});
|
||||
selectChannelMock.mockReturnValue(selectedChannel);
|
||||
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
|
||||
authorizeDownstreamTokenMock.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
source: 'global',
|
||||
token: 'sk-query-auth',
|
||||
key: null,
|
||||
policy: {
|
||||
supportedModels: [],
|
||||
allowedRouteIds: [],
|
||||
siteWeightMultipliers: {},
|
||||
},
|
||||
});
|
||||
fetchMock.mockResolvedValueOnce(createSseResponse([
|
||||
'event: response.completed\n',
|
||||
'data: {"type":"response.completed","response":{"id":"resp_http_fallback_query","model":"gpt-5.4","status":"completed","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}\n\n',
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = createClientSocketForPath(`${baseUrl}/v1/responses?key=sk-query-auth`);
|
||||
await waitForSocketOpen(socket);
|
||||
const messagesPromise = waitForSocketMessages(socket, 1);
|
||||
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.create',
|
||||
model: 'gpt-5.4',
|
||||
input: [],
|
||||
}));
|
||||
|
||||
const messages = await messagesPromise;
|
||||
socket.close();
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
expect(messages[0]?.response?.id).toBe('resp_http_fallback_query');
|
||||
});
|
||||
|
||||
it('rejects websocket turns whose model is blocked by the downstream key policy before channel selection', async () => {
|
||||
authorizeDownstreamTokenMock.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
source: 'managed',
|
||||
token: 'sk-managed-denied',
|
||||
key: {
|
||||
id: 99,
|
||||
name: 'limited-key',
|
||||
},
|
||||
policy: {
|
||||
supportedModels: ['gpt-4.1'],
|
||||
allowedRouteIds: [],
|
||||
siteWeightMultipliers: {},
|
||||
},
|
||||
});
|
||||
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
const messagesPromise = waitForSocketMessages(socket, 1);
|
||||
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.create',
|
||||
model: 'gpt-5.4',
|
||||
input: [],
|
||||
}));
|
||||
|
||||
const messages = await messagesPromise;
|
||||
socket.close();
|
||||
|
||||
expect(messages[0]).toMatchObject({
|
||||
type: 'error',
|
||||
status: 403,
|
||||
});
|
||||
expect(selectChannelMock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('merges follow-up response.create payloads when the selected upstream does not support incremental mode', async () => {
|
||||
const selectedChannel = createSelectedChannel({
|
||||
sitePlatform: 'openai',
|
||||
@@ -483,7 +666,7 @@ describe('responses websocket transport', () => {
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
|
||||
const firstResponsePromise = waitForSocketMessages(socket, 1);
|
||||
@@ -557,6 +740,111 @@ describe('responses websocket transport', () => {
|
||||
]);
|
||||
});
|
||||
|
||||
it('serializes websocket messages per connection so follow-up turns wait for the previous HTTP fallback to finish', async () => {
|
||||
const selectedChannel = createSelectedChannel({
|
||||
sitePlatform: 'openai',
|
||||
actualModel: 'gpt-4.1',
|
||||
});
|
||||
selectChannelMock.mockReturnValue(selectedChannel);
|
||||
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
|
||||
const firstResponseGate = createDeferred<Response>();
|
||||
fetchMock
|
||||
.mockImplementationOnce(() => firstResponseGate.promise)
|
||||
.mockResolvedValueOnce(createSseResponse([
|
||||
'event: response.completed\n',
|
||||
'data: {"type":"response.completed","response":{"id":"resp_ws_2","model":"gpt-4.1","status":"completed","output":[{"id":"msg_2","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"done"}]}],"usage":{"input_tokens":5,"output_tokens":1,"total_tokens":6}}}\n\n',
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.create',
|
||||
model: 'gpt-4.1',
|
||||
instructions: 'be helpful',
|
||||
input: [
|
||||
{
|
||||
id: 'msg_user_1',
|
||||
type: 'message',
|
||||
role: 'user',
|
||||
content: [{ type: 'input_text', text: 'call the tool' }],
|
||||
},
|
||||
],
|
||||
}));
|
||||
|
||||
while (fetchMock.mock.calls.length < 1) {
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
}
|
||||
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.create',
|
||||
previous_response_id: 'resp_ws_1',
|
||||
input: [
|
||||
{
|
||||
id: 'tool_out_1',
|
||||
type: 'function_call_output',
|
||||
call_id: 'call_1',
|
||||
output: 'tool result',
|
||||
},
|
||||
],
|
||||
}));
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
|
||||
const messagesPromise = waitForSocketMessages(socket, 4);
|
||||
firstResponseGate.resolve(createSseResponse([
|
||||
'event: response.output_item.done\n',
|
||||
'data: {"type":"response.output_item.done","output_index":0,"item":{"id":"fc_1","type":"function_call","call_id":"call_1"}}\n\n',
|
||||
'event: response.output_item.done\n',
|
||||
'data: {"type":"response.output_item.done","output_index":1,"item":{"id":"msg_1","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"call tool"}]}}\n\n',
|
||||
'event: response.completed\n',
|
||||
'data: {"type":"response.completed","response":{"id":"resp_ws_1","model":"gpt-4.1","status":"completed","output":[{"id":"fc_1","type":"function_call","call_id":"call_1"},{"id":"msg_1","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"call tool"}]}],"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}\n\n',
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const messages = await messagesPromise;
|
||||
socket.close();
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
expect(messages.map((message) => message.type)).toEqual([
|
||||
'response.output_item.done',
|
||||
'response.output_item.done',
|
||||
'response.completed',
|
||||
'response.completed',
|
||||
]);
|
||||
const [, secondOptions] = fetchMock.mock.calls[1] as [string, RequestInit];
|
||||
const secondBody = JSON.parse(String(secondOptions.body));
|
||||
expect(secondBody.input).toEqual([
|
||||
{
|
||||
id: 'msg_user_1',
|
||||
type: 'message',
|
||||
role: 'user',
|
||||
content: [{ type: 'input_text', text: 'call the tool' }],
|
||||
},
|
||||
{
|
||||
id: 'fc_1',
|
||||
type: 'function_call',
|
||||
call_id: 'call_1',
|
||||
status: 'completed',
|
||||
},
|
||||
{
|
||||
id: 'msg_1',
|
||||
type: 'message',
|
||||
role: 'assistant',
|
||||
status: 'completed',
|
||||
content: [{ type: 'output_text', text: 'call tool' }],
|
||||
},
|
||||
{
|
||||
id: 'tool_out_1',
|
||||
type: 'function_call_output',
|
||||
call_id: 'call_1',
|
||||
output: 'tool result',
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('preserves incremental response.create payloads with previous_response_id for websocket-capable upstreams', async () => {
|
||||
const selectedChannel = createSelectedChannel({
|
||||
siteUrl: upstreamSiteUrl,
|
||||
@@ -609,7 +897,7 @@ describe('responses websocket transport', () => {
|
||||
}));
|
||||
};
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
|
||||
const firstResponsePromise = waitForSocketMessages(socket, 1);
|
||||
@@ -694,7 +982,7 @@ describe('responses websocket transport', () => {
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
const firstResponsePromise = waitForSocketMessages(socket, 1);
|
||||
|
||||
@@ -753,7 +1041,7 @@ describe('responses websocket transport', () => {
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
|
||||
const prewarmMessagesPromise = waitForSocketMessages(socket, 2);
|
||||
@@ -805,7 +1093,7 @@ describe('responses websocket transport', () => {
|
||||
selectChannelMock.mockReturnValue(selectedChannel);
|
||||
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
const messagesPromise = waitForSocketMessages(socket, 1);
|
||||
|
||||
@@ -852,7 +1140,7 @@ describe('responses websocket transport', () => {
|
||||
socket.close();
|
||||
};
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
const socket = createClientSocket(baseUrl);
|
||||
await waitForSocketOpen(socket);
|
||||
const messagesPromise = waitForSocketMessages(socket, 3, 400);
|
||||
|
||||
|
||||
@@ -1,8 +1,15 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import type { IncomingMessage } from 'node:http';
|
||||
import type { Duplex } from 'node:stream';
|
||||
import { WebSocketServer, type RawData, type WebSocket } from 'ws';
|
||||
import { createCodexWebsocketRuntime, CodexWebsocketRuntimeError } from '../../proxy-core/runtime/codexWebsocketRuntime.js';
|
||||
import {
|
||||
authorizeDownstreamToken,
|
||||
consumeManagedKeyRequest,
|
||||
isModelAllowedByPolicyOrAllowedRoutes,
|
||||
type DownstreamTokenAuthSuccess,
|
||||
} from '../../services/downstreamApiKeyService.js';
|
||||
import { tokenRouter } from '../../services/tokenRouter.js';
|
||||
import { buildOauthProviderHeaders } from '../../services/oauth/service.js';
|
||||
import { openAiResponsesTransformer } from '../../transformers/openai/responses/index.js';
|
||||
@@ -15,6 +22,7 @@ const RESPONSES_WEBSOCKET_TRANSPORT_HEADER = 'x-metapi-responses-websocket-trans
|
||||
const codexWebsocketRuntime = createCodexWebsocketRuntime();
|
||||
|
||||
type SelectedChannel = NonNullable<Awaited<ReturnType<typeof tokenRouter.selectChannel>>>;
|
||||
type ResponsesWebsocketAuthContext = DownstreamTokenAuthSuccess;
|
||||
|
||||
type NormalizedResponsesWebsocketRequest =
|
||||
| {
|
||||
@@ -131,8 +139,8 @@ function shouldReuseSelectedChannel(
|
||||
}
|
||||
|
||||
function deriveCodexExplicitSessionId(body: Record<string, unknown>, sessionId: string): string {
|
||||
const promptCacheKey = asTrimmedString(body.prompt_cache_key);
|
||||
return promptCacheKey || sessionId;
|
||||
void body;
|
||||
return sessionId;
|
||||
}
|
||||
|
||||
function parseJsonObject(raw: RawData): Record<string, unknown> | null {
|
||||
@@ -331,15 +339,25 @@ async function forwardResponsesRequestViaHttp(input: {
|
||||
request: IncomingMessage;
|
||||
payload: Record<string, unknown>;
|
||||
preserveIncrementalMode: boolean;
|
||||
authToken: string;
|
||||
}): Promise<unknown[] | null> {
|
||||
const injectHeaders: Record<string, string | string[]> = {
|
||||
...buildInjectHeaders(input.request),
|
||||
[RESPONSES_WEBSOCKET_TRANSPORT_HEADER]: '1',
|
||||
...(input.preserveIncrementalMode ? { [RESPONSES_WEBSOCKET_MODE_HEADER]: 'incremental' } : {}),
|
||||
};
|
||||
if (
|
||||
!headerValueToTrimmedString(injectHeaders.authorization)
|
||||
&& !headerValueToTrimmedString(injectHeaders['x-api-key'])
|
||||
&& !headerValueToTrimmedString(injectHeaders['x-goog-api-key'])
|
||||
) {
|
||||
injectHeaders.authorization = `Bearer ${input.authToken}`;
|
||||
}
|
||||
|
||||
const response = await input.app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/responses',
|
||||
headers: {
|
||||
...buildInjectHeaders(input.request),
|
||||
[RESPONSES_WEBSOCKET_TRANSPORT_HEADER]: '1',
|
||||
...(input.preserveIncrementalMode ? { [RESPONSES_WEBSOCKET_MODE_HEADER]: 'incremental' } : {}),
|
||||
},
|
||||
headers: injectHeaders,
|
||||
payload: input.payload,
|
||||
});
|
||||
|
||||
@@ -417,15 +435,45 @@ function buildInjectHeaders(request: IncomingMessage): Record<string, string | s
|
||||
return headers;
|
||||
}
|
||||
|
||||
function extractWebsocketAuthToken(request: IncomingMessage, url: URL): string {
|
||||
const auth = headerValueToTrimmedString(request.headers.authorization);
|
||||
if (auth) return auth.replace(/^Bearer\s+/i, '').trim();
|
||||
const apiKey = headerValueToTrimmedString(request.headers['x-api-key']);
|
||||
if (apiKey) return apiKey;
|
||||
const googApiKey = headerValueToTrimmedString(request.headers['x-goog-api-key']);
|
||||
if (googApiKey) return googApiKey;
|
||||
return asTrimmedString(url.searchParams.get('key'));
|
||||
}
|
||||
|
||||
function writeUpgradeHttpError(socket: Duplex, status: number, message: string): void {
|
||||
const statusText = status === 401
|
||||
? 'Unauthorized'
|
||||
: status === 403
|
||||
? 'Forbidden'
|
||||
: status === 400
|
||||
? 'Bad Request'
|
||||
: 'Error';
|
||||
const body = JSON.stringify({ error: message });
|
||||
socket.end(
|
||||
`HTTP/1.1 ${status} ${statusText}\r\n`
|
||||
+ 'Content-Type: application/json\r\n'
|
||||
+ `Content-Length: ${Buffer.byteLength(body)}\r\n`
|
||||
+ 'Connection: close\r\n'
|
||||
+ '\r\n'
|
||||
+ body,
|
||||
);
|
||||
}
|
||||
|
||||
async function supportsResponsesWebsocketIncrementalInput(
|
||||
parsed: Record<string, unknown>,
|
||||
lastRequest: Record<string, unknown> | null,
|
||||
authContext: ResponsesWebsocketAuthContext,
|
||||
): Promise<boolean> {
|
||||
const requestModel = asTrimmedString(parsed.model) || asTrimmedString(lastRequest?.model);
|
||||
if (!requestModel) return false;
|
||||
|
||||
try {
|
||||
const selected = await tokenRouter.previewSelectedChannel(requestModel);
|
||||
const selected = await tokenRouter.previewSelectedChannel(requestModel, authContext.policy);
|
||||
return selectedChannelSupportsIncrementalInput(selected, requestModel);
|
||||
} catch {
|
||||
return false;
|
||||
@@ -436,6 +484,7 @@ async function handleResponsesWebsocketConnection(
|
||||
app: FastifyInstance,
|
||||
socket: WebSocket,
|
||||
request: IncomingMessage,
|
||||
authContext: ResponsesWebsocketAuthContext,
|
||||
) {
|
||||
const websocketSessionId = headerValueToTrimmedString(request.headers['session_id'])
|
||||
|| headerValueToTrimmedString(request.headers['session-id'])
|
||||
@@ -443,136 +492,155 @@ async function handleResponsesWebsocketConnection(
|
||||
let lastRequest: Record<string, unknown> | null = null;
|
||||
let lastResponseOutput: unknown[] = [];
|
||||
let selectedChannel: SelectedChannel | null = null;
|
||||
let messageQueue = Promise.resolve();
|
||||
|
||||
socket.once('close', () => {
|
||||
void codexWebsocketRuntime.closeSession(websocketSessionId);
|
||||
});
|
||||
|
||||
socket.on('message', async (raw) => {
|
||||
const parsed = parseJsonObject(raw);
|
||||
if (!parsed) {
|
||||
writeResponsesWebsocketError(socket, 400, 'Invalid websocket JSON payload');
|
||||
return;
|
||||
}
|
||||
socket.on('message', (raw) => {
|
||||
messageQueue = messageQueue
|
||||
.catch(() => undefined)
|
||||
.then(async () => {
|
||||
try {
|
||||
const parsed = parseJsonObject(raw);
|
||||
if (!parsed) {
|
||||
writeResponsesWebsocketError(socket, 400, 'Invalid websocket JSON payload');
|
||||
return;
|
||||
}
|
||||
|
||||
const requestModel = asTrimmedString(parsed.model) || asTrimmedString(lastRequest?.model);
|
||||
const supportsIncrementalInput = selectedChannelSupportsIncrementalInput(selectedChannel, requestModel)
|
||||
|| await supportsResponsesWebsocketIncrementalInput(parsed, lastRequest);
|
||||
const shouldHandleLocalPrewarm = shouldHandleResponsesWebsocketPrewarmLocally(
|
||||
parsed,
|
||||
lastRequest,
|
||||
supportsIncrementalInput,
|
||||
);
|
||||
const normalized = normalizeResponsesWebsocketRequest(
|
||||
parsed,
|
||||
lastRequest,
|
||||
lastResponseOutput,
|
||||
supportsIncrementalInput,
|
||||
);
|
||||
if (!normalized.ok) {
|
||||
writeResponsesWebsocketError(socket, normalized.status, normalized.message);
|
||||
return;
|
||||
}
|
||||
const requestModel = asTrimmedString(parsed.model) || asTrimmedString(lastRequest?.model);
|
||||
if (requestModel && !await isModelAllowedByPolicyOrAllowedRoutes(requestModel, authContext.policy)) {
|
||||
writeResponsesWebsocketError(socket, 403, 'model is not allowed for this downstream key');
|
||||
return;
|
||||
}
|
||||
const supportsIncrementalInput = selectedChannelSupportsIncrementalInput(selectedChannel, requestModel)
|
||||
|| await supportsResponsesWebsocketIncrementalInput(parsed, lastRequest, authContext);
|
||||
const shouldHandleLocalPrewarm = shouldHandleResponsesWebsocketPrewarmLocally(
|
||||
parsed,
|
||||
lastRequest,
|
||||
supportsIncrementalInput,
|
||||
);
|
||||
const normalized = normalizeResponsesWebsocketRequest(
|
||||
parsed,
|
||||
lastRequest,
|
||||
lastResponseOutput,
|
||||
supportsIncrementalInput,
|
||||
);
|
||||
if (!normalized.ok) {
|
||||
writeResponsesWebsocketError(socket, normalized.status, normalized.message);
|
||||
return;
|
||||
}
|
||||
|
||||
lastRequest = normalized.nextRequestSnapshot;
|
||||
if (shouldHandleLocalPrewarm) {
|
||||
lastResponseOutput = [];
|
||||
for (const payload of synthesizePrewarmResponsePayloads(normalized.request)) {
|
||||
socket.send(JSON.stringify(payload));
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (authContext.source === 'managed' && authContext.key?.id) {
|
||||
await consumeManagedKeyRequest(authContext.key.id);
|
||||
}
|
||||
|
||||
if (!shouldReuseSelectedChannel(selectedChannel, requestModel)) {
|
||||
selectedChannel = requestModel
|
||||
? await tokenRouter.selectChannel(requestModel)
|
||||
: null;
|
||||
}
|
||||
lastRequest = normalized.nextRequestSnapshot;
|
||||
if (shouldHandleLocalPrewarm) {
|
||||
lastResponseOutput = [];
|
||||
for (const payload of synthesizePrewarmResponsePayloads(normalized.request)) {
|
||||
socket.send(JSON.stringify(payload));
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
const codexWebsocketChannel = selectedChannelSupportsCodexWebsocketTransport(selectedChannel, requestModel)
|
||||
? selectedChannel
|
||||
: null;
|
||||
if (!shouldReuseSelectedChannel(selectedChannel, requestModel)) {
|
||||
selectedChannel = requestModel
|
||||
? await tokenRouter.selectChannel(requestModel, authContext.policy)
|
||||
: null;
|
||||
}
|
||||
|
||||
if (codexWebsocketChannel) {
|
||||
const downstreamHeaders: Record<string, unknown> = {
|
||||
...(request.headers as Record<string, unknown>),
|
||||
[RESPONSES_WEBSOCKET_TRANSPORT_HEADER]: '1',
|
||||
...(supportsIncrementalInput ? { [RESPONSES_WEBSOCKET_MODE_HEADER]: 'incremental' } : {}),
|
||||
};
|
||||
const providerHeaders = buildOauthProviderHeaders({
|
||||
extraConfig: typeof codexWebsocketChannel.account.extraConfig === 'string'
|
||||
? codexWebsocketChannel.account.extraConfig
|
||||
: null,
|
||||
downstreamHeaders,
|
||||
});
|
||||
const prepared = buildUpstreamEndpointRequest({
|
||||
endpoint: 'responses',
|
||||
modelName: asTrimmedString(codexWebsocketChannel.actualModel) || requestModel,
|
||||
stream: true,
|
||||
tokenValue: codexWebsocketChannel.tokenValue,
|
||||
sitePlatform: codexWebsocketChannel.site.platform,
|
||||
siteUrl: codexWebsocketChannel.site.url,
|
||||
openaiBody: normalized.request,
|
||||
downstreamFormat: 'responses',
|
||||
responsesOriginalBody: normalized.request,
|
||||
downstreamHeaders,
|
||||
providerHeaders,
|
||||
codexExplicitSessionId: deriveCodexExplicitSessionId(normalized.request, websocketSessionId),
|
||||
});
|
||||
const requestUrl = `${codexWebsocketChannel.site.url.replace(/\/+$/, '')}${prepared.path}`;
|
||||
const codexWebsocketChannel = selectedChannelSupportsCodexWebsocketTransport(selectedChannel, requestModel)
|
||||
? selectedChannel
|
||||
: null;
|
||||
|
||||
try {
|
||||
const runtimeResult = await codexWebsocketRuntime.sendRequest({
|
||||
sessionId: websocketSessionId,
|
||||
requestUrl,
|
||||
headers: prepared.headers,
|
||||
body: prepared.body,
|
||||
});
|
||||
lastResponseOutput = collectResponsesOutput(runtimeResult.events);
|
||||
for (const payload of runtimeResult.events) {
|
||||
socket.send(JSON.stringify(payload));
|
||||
}
|
||||
} catch (error) {
|
||||
const runtimeError = error instanceof CodexWebsocketRuntimeError
|
||||
? error
|
||||
: new CodexWebsocketRuntimeError('upstream websocket request failed');
|
||||
if (runtimeError.status === 426) {
|
||||
const fallbackOutput = await forwardResponsesRequestViaHttp({
|
||||
if (codexWebsocketChannel) {
|
||||
const downstreamHeaders: Record<string, unknown> = {
|
||||
...(request.headers as Record<string, unknown>),
|
||||
[RESPONSES_WEBSOCKET_TRANSPORT_HEADER]: '1',
|
||||
...(supportsIncrementalInput ? { [RESPONSES_WEBSOCKET_MODE_HEADER]: 'incremental' } : {}),
|
||||
};
|
||||
const providerHeaders = buildOauthProviderHeaders({
|
||||
extraConfig: typeof codexWebsocketChannel.account.extraConfig === 'string'
|
||||
? codexWebsocketChannel.account.extraConfig
|
||||
: null,
|
||||
downstreamHeaders,
|
||||
});
|
||||
const prepared = buildUpstreamEndpointRequest({
|
||||
endpoint: 'responses',
|
||||
modelName: asTrimmedString(codexWebsocketChannel.actualModel) || requestModel,
|
||||
stream: true,
|
||||
tokenValue: codexWebsocketChannel.tokenValue,
|
||||
sitePlatform: codexWebsocketChannel.site.platform,
|
||||
siteUrl: codexWebsocketChannel.site.url,
|
||||
openaiBody: normalized.request,
|
||||
downstreamFormat: 'responses',
|
||||
responsesOriginalBody: normalized.request,
|
||||
downstreamHeaders,
|
||||
providerHeaders,
|
||||
codexExplicitSessionId: deriveCodexExplicitSessionId(normalized.request, websocketSessionId),
|
||||
});
|
||||
const requestUrl = `${codexWebsocketChannel.site.url.replace(/\/+$/, '')}${prepared.path}`;
|
||||
|
||||
try {
|
||||
const runtimeResult = await codexWebsocketRuntime.sendRequest({
|
||||
sessionId: websocketSessionId,
|
||||
requestUrl,
|
||||
headers: prepared.headers,
|
||||
body: prepared.body,
|
||||
});
|
||||
lastResponseOutput = collectResponsesOutput(runtimeResult.events);
|
||||
for (const payload of runtimeResult.events) {
|
||||
socket.send(JSON.stringify(payload));
|
||||
}
|
||||
} catch (error) {
|
||||
const runtimeError = error instanceof CodexWebsocketRuntimeError
|
||||
? error
|
||||
: new CodexWebsocketRuntimeError('upstream websocket request failed');
|
||||
if (runtimeError.status && runtimeError.events.length === 0) {
|
||||
const fallbackOutput = await forwardResponsesRequestViaHttp({
|
||||
app,
|
||||
socket,
|
||||
request,
|
||||
payload: normalized.request,
|
||||
preserveIncrementalMode: false,
|
||||
authToken: authContext.token,
|
||||
});
|
||||
if (fallbackOutput) {
|
||||
lastResponseOutput = fallbackOutput;
|
||||
}
|
||||
return;
|
||||
}
|
||||
lastResponseOutput = collectResponsesOutput(runtimeError.events);
|
||||
for (const payload of runtimeError.events) {
|
||||
socket.send(JSON.stringify(payload));
|
||||
}
|
||||
writeResponsesWebsocketError(
|
||||
socket,
|
||||
runtimeError.status || 408,
|
||||
runtimeError.message,
|
||||
runtimeError.payload,
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
const forwardedOutput = await forwardResponsesRequestViaHttp({
|
||||
app,
|
||||
socket,
|
||||
request,
|
||||
payload: normalized.request,
|
||||
preserveIncrementalMode: false,
|
||||
preserveIncrementalMode: supportsIncrementalInput,
|
||||
authToken: authContext.token,
|
||||
});
|
||||
if (fallbackOutput) {
|
||||
lastResponseOutput = fallbackOutput;
|
||||
if (forwardedOutput) {
|
||||
lastResponseOutput = forwardedOutput;
|
||||
}
|
||||
return;
|
||||
} catch {
|
||||
writeResponsesWebsocketError(socket, 500, 'internal websocket proxy error');
|
||||
}
|
||||
lastResponseOutput = collectResponsesOutput(runtimeError.events);
|
||||
for (const payload of runtimeError.events) {
|
||||
socket.send(JSON.stringify(payload));
|
||||
}
|
||||
writeResponsesWebsocketError(
|
||||
socket,
|
||||
runtimeError.status || 408,
|
||||
runtimeError.message,
|
||||
runtimeError.payload,
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
const forwardedOutput = await forwardResponsesRequestViaHttp({
|
||||
app,
|
||||
socket,
|
||||
request,
|
||||
payload: normalized.request,
|
||||
preserveIncrementalMode: supportsIncrementalInput,
|
||||
});
|
||||
if (forwardedOutput) {
|
||||
lastResponseOutput = forwardedOutput;
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
@@ -588,10 +656,24 @@ export function ensureResponsesWebsocketTransport(app: FastifyInstance) {
|
||||
});
|
||||
|
||||
app.server.on('upgrade', (request, socket, head) => {
|
||||
const url = new URL(request.url || '/', 'http://localhost');
|
||||
if (url.pathname !== '/v1/responses') return;
|
||||
websocketServer.handleUpgrade(request, socket, head, (client) => {
|
||||
void handleResponsesWebsocketConnection(app, client, request);
|
||||
void (async () => {
|
||||
const url = new URL(request.url || '/', 'http://localhost');
|
||||
if (url.pathname !== '/v1/responses') return;
|
||||
const token = extractWebsocketAuthToken(request, url);
|
||||
if (!token) {
|
||||
writeUpgradeHttpError(socket, 401, 'Missing Authorization, x-api-key, x-goog-api-key, or key query parameter');
|
||||
return;
|
||||
}
|
||||
const authResult = await authorizeDownstreamToken(token);
|
||||
if (!authResult.ok) {
|
||||
writeUpgradeHttpError(socket, authResult.statusCode, authResult.error);
|
||||
return;
|
||||
}
|
||||
websocketServer.handleUpgrade(request, socket, head, (client) => {
|
||||
void handleResponsesWebsocketConnection(app, client, request, authResult);
|
||||
});
|
||||
})().catch(() => {
|
||||
writeUpgradeHttpError(socket, 500, 'internal websocket proxy error');
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -657,7 +657,7 @@ describe('buildUpstreamEndpointRequest', () => {
|
||||
expect(request.body.prompt_cache_key).toBeUndefined();
|
||||
});
|
||||
|
||||
it('preserves explicit prompt_cache_key for native codex responses requests', () => {
|
||||
it('preserves explicit prompt_cache_key for native codex responses requests without mapping it into codex session headers', () => {
|
||||
const request = buildUpstreamEndpointRequest({
|
||||
endpoint: 'responses',
|
||||
modelName: 'gpt-5.4',
|
||||
@@ -675,11 +675,10 @@ describe('buildUpstreamEndpointRequest', () => {
|
||||
providerHeaders: {
|
||||
Originator: 'codex_cli_rs',
|
||||
},
|
||||
codexExplicitSessionId: 'codex-cache-123',
|
||||
} as any);
|
||||
|
||||
expect(request.headers.Session_id).toBe('codex-cache-123');
|
||||
expect(request.headers.Conversation_id).toBe('codex-cache-123');
|
||||
expect(request.headers.Session_id).toMatch(/^[0-9a-f-]{36}$/i);
|
||||
expect(request.headers.Conversation_id).toBeUndefined();
|
||||
expect(request.body.prompt_cache_key).toBe('codex-cache-123');
|
||||
});
|
||||
|
||||
|
||||
@@ -393,6 +393,83 @@ describe('refreshModelsForAccount credential discovery', () => {
|
||||
});
|
||||
});
|
||||
|
||||
it('preserves existing availability when allowInactive refresh fails', async () => {
|
||||
getApiTokenMock.mockResolvedValue(null);
|
||||
getModelsMock.mockRejectedValue(new Error('upstream unavailable'));
|
||||
|
||||
const site = await db.insert(schema.sites).values({
|
||||
name: 'site-rebind-refresh',
|
||||
url: 'https://site-rebind-refresh.example.com',
|
||||
platform: 'new-api',
|
||||
status: 'active',
|
||||
}).returning().get();
|
||||
|
||||
const account = await db.insert(schema.accounts).values({
|
||||
siteId: site.id,
|
||||
username: 'rebind-user',
|
||||
accessToken: 'session-token',
|
||||
apiToken: null,
|
||||
status: 'disabled',
|
||||
extraConfig: JSON.stringify({ credentialMode: 'session' }),
|
||||
}).returning().get();
|
||||
|
||||
const token = await db.insert(schema.accountTokens).values({
|
||||
accountId: account.id,
|
||||
name: 'default',
|
||||
token: 'sk-stored-token',
|
||||
source: 'manual',
|
||||
enabled: true,
|
||||
isDefault: true,
|
||||
valueStatus: 'ready' as any,
|
||||
}).returning().get();
|
||||
|
||||
await db.insert(schema.modelAvailability).values({
|
||||
accountId: account.id,
|
||||
modelName: 'gpt-4.1',
|
||||
available: true,
|
||||
latencyMs: 120,
|
||||
checkedAt: '2026-03-21T11:30:00.000Z',
|
||||
}).run();
|
||||
|
||||
await db.insert(schema.tokenModelAvailability).values({
|
||||
tokenId: token.id,
|
||||
modelName: 'gpt-4.1',
|
||||
available: true,
|
||||
latencyMs: 90,
|
||||
checkedAt: '2026-03-21T11:30:00.000Z',
|
||||
}).run();
|
||||
|
||||
const result = await refreshModelsForAccount(account.id, { allowInactive: true });
|
||||
|
||||
expect(result).toMatchObject({
|
||||
accountId: account.id,
|
||||
refreshed: true,
|
||||
status: 'failed',
|
||||
modelCount: 0,
|
||||
discoveredByCredential: false,
|
||||
});
|
||||
|
||||
const modelRows = await db.select().from(schema.modelAvailability)
|
||||
.where(eq(schema.modelAvailability.accountId, account.id))
|
||||
.all();
|
||||
expect(modelRows).toHaveLength(1);
|
||||
expect(modelRows[0]).toMatchObject({
|
||||
accountId: account.id,
|
||||
modelName: 'gpt-4.1',
|
||||
available: true,
|
||||
});
|
||||
|
||||
const tokenRows = await db.select().from(schema.tokenModelAvailability)
|
||||
.where(eq(schema.tokenModelAvailability.tokenId, token.id))
|
||||
.all();
|
||||
expect(tokenRows).toHaveLength(1);
|
||||
expect(tokenRows[0]).toMatchObject({
|
||||
tokenId: token.id,
|
||||
modelName: 'gpt-4.1',
|
||||
available: true,
|
||||
});
|
||||
});
|
||||
|
||||
it('does not scan masked_pending placeholders as token credentials', async () => {
|
||||
getApiTokenMock.mockResolvedValue(null);
|
||||
getModelsMock.mockImplementation(async (_baseUrl: string, token: string) => (
|
||||
|
||||
@@ -493,7 +493,10 @@ function shouldRetryModelDiscoveryWithOauthRefresh(error: unknown): boolean {
|
||||
|| message.includes('unauthenticated');
|
||||
}
|
||||
|
||||
export async function refreshModelsForAccount(accountId: number): Promise<ModelRefreshResult> {
|
||||
export async function refreshModelsForAccount(
|
||||
accountId: number,
|
||||
options?: { allowInactive?: boolean },
|
||||
): Promise<ModelRefreshResult> {
|
||||
const row = await db.select().from(schema.accounts)
|
||||
.innerJoin(schema.sites, eq(schema.accounts.siteId, schema.sites.id))
|
||||
.where(eq(schema.accounts.id, accountId))
|
||||
@@ -509,26 +512,65 @@ export async function refreshModelsForAccount(accountId: number): Promise<ModelR
|
||||
const adapter = getAdapter(site.platform);
|
||||
const accountProxyUrl = getProxyUrlFromExtraConfig(account.extraConfig);
|
||||
|
||||
const accountTokens = await db.select()
|
||||
.from(schema.accountTokens)
|
||||
.where(eq(schema.accountTokens.accountId, accountId))
|
||||
.all();
|
||||
|
||||
await db.delete(schema.modelAvailability)
|
||||
.where(eq(schema.modelAvailability.accountId, accountId))
|
||||
.run();
|
||||
|
||||
for (const token of accountTokens) {
|
||||
await db.delete(schema.tokenModelAvailability)
|
||||
const restoreAvailabilityOnFailure = options?.allowInactive === true;
|
||||
const previousAccountTokens = restoreAvailabilityOnFailure
|
||||
? await db.select()
|
||||
.from(schema.accountTokens)
|
||||
.where(eq(schema.accountTokens.accountId, accountId))
|
||||
.all()
|
||||
: [];
|
||||
const previousModelAvailability = restoreAvailabilityOnFailure
|
||||
? await db.select()
|
||||
.from(schema.modelAvailability)
|
||||
.where(eq(schema.modelAvailability.accountId, accountId))
|
||||
.all()
|
||||
: [];
|
||||
const previousTokenModelAvailability = restoreAvailabilityOnFailure
|
||||
? (await Promise.all(previousAccountTokens.map(async (token) => db.select()
|
||||
.from(schema.tokenModelAvailability)
|
||||
.where(eq(schema.tokenModelAvailability.tokenId, token.id))
|
||||
.all()))).flat()
|
||||
: [];
|
||||
|
||||
const clearExistingAvailability = async () => {
|
||||
await db.delete(schema.modelAvailability)
|
||||
.where(eq(schema.modelAvailability.accountId, accountId))
|
||||
.run();
|
||||
}
|
||||
|
||||
const currentAccountTokens = await db.select({ id: schema.accountTokens.id })
|
||||
.from(schema.accountTokens)
|
||||
.where(eq(schema.accountTokens.accountId, accountId))
|
||||
.all();
|
||||
|
||||
for (const token of currentAccountTokens) {
|
||||
await db.delete(schema.tokenModelAvailability)
|
||||
.where(eq(schema.tokenModelAvailability.tokenId, token.id))
|
||||
.run();
|
||||
}
|
||||
};
|
||||
|
||||
const restorePreviousAvailability = async () => {
|
||||
if (!restoreAvailabilityOnFailure) return;
|
||||
await clearExistingAvailability();
|
||||
if (previousModelAvailability.length > 0) {
|
||||
await db.insert(schema.modelAvailability).values(
|
||||
previousModelAvailability.map(({ id: _id, ...row }) => row),
|
||||
).run();
|
||||
}
|
||||
if (previousTokenModelAvailability.length > 0) {
|
||||
await db.insert(schema.tokenModelAvailability).values(
|
||||
previousTokenModelAvailability.map(({ id: _id, ...row }) => row),
|
||||
).run();
|
||||
}
|
||||
};
|
||||
|
||||
await clearExistingAvailability();
|
||||
|
||||
if (isSiteDisabled(site.status)) {
|
||||
return buildSkippedRefreshResult(accountId, 'site_disabled', '站点已禁用');
|
||||
}
|
||||
|
||||
if (account.status !== 'active') {
|
||||
if (account.status !== 'active' && !options?.allowInactive) {
|
||||
return buildSkippedRefreshResult(accountId, 'adapter_or_status', '平台不可用或账号未激活');
|
||||
}
|
||||
|
||||
@@ -592,6 +634,7 @@ export async function refreshModelsForAccount(accountId: number): Promise<ModelR
|
||||
source: 'model-discovery',
|
||||
checkedAt,
|
||||
});
|
||||
await restorePreviousAvailability();
|
||||
return buildFailedRefreshResult({
|
||||
accountId,
|
||||
errorCode,
|
||||
@@ -662,6 +705,7 @@ export async function refreshModelsForAccount(accountId: number): Promise<ModelR
|
||||
source: 'model-discovery',
|
||||
checkedAt,
|
||||
});
|
||||
await restorePreviousAvailability();
|
||||
return buildFailedRefreshResult({
|
||||
accountId,
|
||||
errorCode,
|
||||
@@ -751,6 +795,7 @@ export async function refreshModelsForAccount(accountId: number): Promise<ModelR
|
||||
source: 'model-discovery',
|
||||
checkedAt,
|
||||
});
|
||||
await restorePreviousAvailability();
|
||||
return buildFailedRefreshResult({
|
||||
accountId,
|
||||
errorCode,
|
||||
@@ -822,6 +867,7 @@ export async function refreshModelsForAccount(accountId: number): Promise<ModelR
|
||||
source: 'model-discovery',
|
||||
checkedAt,
|
||||
});
|
||||
await restorePreviousAvailability();
|
||||
return buildFailedRefreshResult({
|
||||
accountId,
|
||||
errorCode,
|
||||
@@ -994,6 +1040,7 @@ export async function refreshModelsForAccount(accountId: number): Promise<ModelR
|
||||
source: 'model-discovery',
|
||||
checkedAt: new Date().toISOString(),
|
||||
});
|
||||
await restorePreviousAvailability();
|
||||
return buildFailedRefreshResult({
|
||||
accountId,
|
||||
errorCode,
|
||||
|
||||
@@ -91,11 +91,15 @@ function parseTokenResponsePayload(payload: unknown): CodexTokenExchangeResult {
|
||||
throw new Error('codex token exchange response missing required fields');
|
||||
}
|
||||
const claims = parseJwtClaims(idToken);
|
||||
const accountId = asTrimmedString(claims?.['https://api.openai.com/auth']?.chatgpt_account_id);
|
||||
if (!accountId) {
|
||||
throw new Error('codex token exchange response missing chatgpt_account_id');
|
||||
}
|
||||
return {
|
||||
accessToken,
|
||||
refreshToken,
|
||||
idToken,
|
||||
accountId: asTrimmedString(claims?.['https://api.openai.com/auth']?.chatgpt_account_id),
|
||||
accountId,
|
||||
email: asTrimmedString(claims?.email),
|
||||
planType: asTrimmedString(claims?.['https://api.openai.com/auth']?.chatgpt_plan_type),
|
||||
tokenExpiresAt: Date.now() + expiresIn * 1000,
|
||||
@@ -220,7 +224,7 @@ export const codexOauthProvider: OAuthProviderDefinition = {
|
||||
};
|
||||
},
|
||||
buildProxyHeaders: ({ oauth, downstreamHeaders }) => {
|
||||
const accountId = getHeaderValue(downstreamHeaders, 'chatgpt-account-id') || oauth.accountId || oauth.accountKey;
|
||||
const accountId = oauth.accountId || oauth.accountKey;
|
||||
const originator = getHeaderValue(downstreamHeaders, 'originator') || 'codex_cli_rs';
|
||||
const headers: Record<string, string> = {
|
||||
Originator: originator,
|
||||
|
||||
@@ -178,7 +178,7 @@ async function findExistingOauthAccount(input: {
|
||||
if (byKey) return byKey;
|
||||
}
|
||||
|
||||
if (!accountKey && email) {
|
||||
if (!accountKey && email && input.provider !== 'codex') {
|
||||
const byEmail = await db.select().from(schema.accounts).where(and(
|
||||
eq(schema.accounts.oauthProvider, input.provider),
|
||||
eq(schema.accounts.username, email),
|
||||
@@ -242,7 +242,7 @@ async function upsertOauthAccount(input: {
|
||||
accessToken: input.exchange.accessToken,
|
||||
apiToken: null,
|
||||
checkinEnabled: false,
|
||||
status: 'active',
|
||||
status: 'disabled',
|
||||
oauthProvider: input.definition.metadata.provider,
|
||||
oauthAccountKey: oauth.accountKey || oauth.accountId || null,
|
||||
oauthProjectId: oauth.projectId || null,
|
||||
@@ -253,6 +253,7 @@ async function upsertOauthAccount(input: {
|
||||
account: await db.select().from(schema.accounts).where(eq(schema.accounts.id, existing.id)).get(),
|
||||
site,
|
||||
created: false,
|
||||
previousAccount: existing,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -270,7 +271,7 @@ async function upsertOauthAccount(input: {
|
||||
isPinned: false,
|
||||
sortOrder: await getNextAccountSortOrder(),
|
||||
}).returning().get();
|
||||
return { account: created, site, created: true };
|
||||
return { account: created, site, created: true, previousAccount: null };
|
||||
}
|
||||
|
||||
export function listOauthProviders() {
|
||||
@@ -365,7 +366,7 @@ export async function handleOauthCallback(input: {
|
||||
codeVerifier: session.codeVerifier,
|
||||
projectId: session.projectId,
|
||||
});
|
||||
const { account, site, created } = await upsertOauthAccount({
|
||||
const { account, site, created, previousAccount } = await upsertOauthAccount({
|
||||
definition,
|
||||
exchange,
|
||||
rebindAccountId: session.rebindAccountId,
|
||||
@@ -375,10 +376,27 @@ export async function handleOauthCallback(input: {
|
||||
throw new Error('failed to persist oauth account');
|
||||
}
|
||||
|
||||
const refreshResult = await refreshModelsForAccount(account.id);
|
||||
const refreshResult = await refreshModelsForAccount(
|
||||
account.id,
|
||||
previousAccount ? { allowInactive: true } : undefined,
|
||||
);
|
||||
if (refreshResult.status !== 'success') {
|
||||
if (created) {
|
||||
await db.delete(schema.accounts).where(eq(schema.accounts.id, account.id)).run();
|
||||
} else if (previousAccount) {
|
||||
await db.update(schema.accounts).set({
|
||||
siteId: previousAccount.siteId,
|
||||
username: previousAccount.username,
|
||||
accessToken: previousAccount.accessToken,
|
||||
apiToken: previousAccount.apiToken,
|
||||
checkinEnabled: previousAccount.checkinEnabled,
|
||||
status: previousAccount.status,
|
||||
oauthProvider: previousAccount.oauthProvider,
|
||||
oauthAccountKey: previousAccount.oauthAccountKey,
|
||||
oauthProjectId: previousAccount.oauthProjectId,
|
||||
extraConfig: previousAccount.extraConfig,
|
||||
updatedAt: previousAccount.updatedAt,
|
||||
}).where(eq(schema.accounts.id, previousAccount.id)).run();
|
||||
}
|
||||
await rebuildTokenRoutesFromAvailability();
|
||||
const errorMessage = refreshResult.errorMessage || `${input.provider} model discovery failed`;
|
||||
@@ -386,6 +404,13 @@ export async function handleOauthCallback(input: {
|
||||
throw new Error(errorMessage);
|
||||
}
|
||||
|
||||
if (previousAccount) {
|
||||
await db.update(schema.accounts).set({
|
||||
status: 'active',
|
||||
updatedAt: new Date().toISOString(),
|
||||
}).where(eq(schema.accounts.id, account.id)).run();
|
||||
}
|
||||
|
||||
await rebuildTokenRoutesFromAvailability();
|
||||
markOauthSessionSuccess(input.state, {
|
||||
accountId: account.id,
|
||||
|
||||
@@ -164,4 +164,93 @@ describe('TokenRouter site status guard', () => {
|
||||
expect(selected).not.toBeNull();
|
||||
expect(selected?.tokenValue).toBe('oauth-access-token');
|
||||
});
|
||||
|
||||
it('prefers codex oauth access token over a stale legacy api token on oauth-managed rows', async () => {
|
||||
const site = await db.insert(schema.sites).values({
|
||||
name: 'codex-site',
|
||||
url: 'https://chatgpt.com/backend-api/codex',
|
||||
platform: 'codex',
|
||||
status: 'active',
|
||||
}).returning().get();
|
||||
|
||||
const account = await db.insert(schema.accounts).values({
|
||||
siteId: site.id,
|
||||
username: 'codex-user@example.com',
|
||||
accessToken: 'oauth-access-token',
|
||||
apiToken: 'stale-direct-token',
|
||||
status: 'active',
|
||||
extraConfig: JSON.stringify({
|
||||
credentialMode: 'session',
|
||||
oauth: {
|
||||
provider: 'codex',
|
||||
accountId: 'chatgpt-account-123',
|
||||
email: 'codex-user@example.com',
|
||||
planType: 'plus',
|
||||
},
|
||||
}),
|
||||
}).returning().get();
|
||||
|
||||
const route = await db.insert(schema.tokenRoutes).values({
|
||||
modelPattern: 'gpt-5.2-codex',
|
||||
enabled: true,
|
||||
}).returning().get();
|
||||
|
||||
await db.insert(schema.routeChannels).values({
|
||||
routeId: route.id,
|
||||
accountId: account.id,
|
||||
tokenId: null,
|
||||
priority: 0,
|
||||
weight: 10,
|
||||
enabled: true,
|
||||
manualOverride: false,
|
||||
}).run();
|
||||
|
||||
const selected = await new TokenRouter().selectChannel('gpt-5.2-codex');
|
||||
expect(selected).not.toBeNull();
|
||||
expect(selected?.tokenValue).toBe('oauth-access-token');
|
||||
});
|
||||
|
||||
it('does not fall back to a stale legacy api token when an oauth-managed row has no access token', async () => {
|
||||
const site = await db.insert(schema.sites).values({
|
||||
name: 'codex-site',
|
||||
url: 'https://chatgpt.com/backend-api/codex',
|
||||
platform: 'codex',
|
||||
status: 'active',
|
||||
}).returning().get();
|
||||
|
||||
const account = await db.insert(schema.accounts).values({
|
||||
siteId: site.id,
|
||||
username: 'codex-user@example.com',
|
||||
accessToken: '',
|
||||
apiToken: 'stale-direct-token',
|
||||
status: 'active',
|
||||
extraConfig: JSON.stringify({
|
||||
credentialMode: 'session',
|
||||
oauth: {
|
||||
provider: 'codex',
|
||||
accountId: 'chatgpt-account-123',
|
||||
email: 'codex-user@example.com',
|
||||
planType: 'plus',
|
||||
},
|
||||
}),
|
||||
}).returning().get();
|
||||
|
||||
const route = await db.insert(schema.tokenRoutes).values({
|
||||
modelPattern: 'gpt-5.2-codex',
|
||||
enabled: true,
|
||||
}).returning().get();
|
||||
|
||||
await db.insert(schema.routeChannels).values({
|
||||
routeId: route.id,
|
||||
accountId: account.id,
|
||||
tokenId: null,
|
||||
priority: 0,
|
||||
weight: 10,
|
||||
enabled: true,
|
||||
manualOverride: false,
|
||||
}).run();
|
||||
|
||||
const selected = await new TokenRouter().selectChannel('gpt-5.2-codex');
|
||||
expect(selected).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -2037,14 +2037,15 @@ export class TokenRouter {
|
||||
return token ? token : null;
|
||||
}
|
||||
|
||||
const fallback = candidate.account.apiToken?.trim();
|
||||
if (fallback) return fallback;
|
||||
|
||||
if (getOauthInfoFromExtraConfig(candidate.account.extraConfig)) {
|
||||
const accessToken = candidate.account.accessToken?.trim();
|
||||
return accessToken || null;
|
||||
if (accessToken) return accessToken;
|
||||
return null;
|
||||
}
|
||||
|
||||
const fallback = candidate.account.apiToken?.trim();
|
||||
if (fallback) return fallback;
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user