From ee7fd12aa0d8735a679b1c0dda1836bd4d35380e Mon Sep 17 00:00:00 2001 From: danielwpz <5139554+danielwpz@users.noreply.github.com> Date: Fri, 29 May 2026 11:20:15 +0800 Subject: [PATCH 1/4] Route Codex Responses through local converter --- src/agent/llm/upstream-openai.ts | 215 ++++++++++++++++-- .../llm/pi-bridge.codex-service-tier.test.ts | 98 ++++++++ 2 files changed, 295 insertions(+), 18 deletions(-) diff --git a/src/agent/llm/upstream-openai.ts b/src/agent/llm/upstream-openai.ts index 596140e..b311dc0 100644 --- a/src/agent/llm/upstream-openai.ts +++ b/src/agent/llm/upstream-openai.ts @@ -156,6 +156,15 @@ const DEFAULT_COMPAT: ResolvedOpenAICompletionsCompat = { const OPENAI_TOOL_CALL_PROVIDERS = new Set(["openai", "openai-codex", "opencode"]); const STREAM_REASONING_CONTENT_MAX_CHARS = 100_000; const STREAM_REASONING_TRUNCATION_PREFIX = "...[earlier reasoning truncated]\n"; +const CODEX_RESPONSE_STATUSES = new Set([ + "completed", + "incomplete", + "failed", + "cancelled", + "queued", + "in_progress", +]); +const CODEX_JWT_AUTH_CLAIM = "https://api.openai.com/auth"; export function supportsUpstreamCostParser(model: Pick, "baseUrl">): boolean { return COST_PARSERS.some((parser) => parser.supports(model)); @@ -170,6 +179,9 @@ export function shouldUseCustomOpenAICompletionsStream( export function shouldUseCustomOpenAIResponsesStream( model: Pick, "api" | "baseUrl">, ): boolean { + if (model.api === "openai-codex-responses") { + return true; + } return model.api === "openai-responses" && supportsUpstreamCostParser(model); } @@ -511,7 +523,7 @@ function appendStreamReasoning(existing: string, delta: string): string { } function streamOpenAIResponsesWithUpstreamUsage( - model: Model<"openai-responses">, + model: Model<"openai-responses" | "openai-codex-responses">, context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream { @@ -537,20 +549,31 @@ function streamOpenAIResponsesWithUpstreamUsage( const client = new OpenAI({ apiKey, - baseURL: model.baseUrl, + baseURL: resolveResponsesBaseUrl(model), + ...(model.api === "openai-codex-responses" + ? { defaultHeaders: buildCodexDefaultHeaders(apiKey, options?.sessionId) } + : {}), dangerouslyAllowBrowser: true, }); - const params = buildOpenAIResponsesParams(model, context, options); + const params = await applyPayloadHook( + buildOpenAIResponsesParams(model, context, options), + model, + options, + ); const rawStream = await client.responses.create( - params, + params as ResponseCreateParamsStreaming, options?.signal ? { signal: options.signal } : undefined, ); let completedUsage: unknown = null; let completedServiceTier: ResponseCreateParamsStreaming["service_tier"] | undefined; async function* tappedStream(): AsyncIterable { - for await (const event of rawStream) { + const events = + model.api === "openai-codex-responses" + ? mapCodexResponsesEvents(rawStream as AsyncIterable>) + : (rawStream as AsyncIterable); + for await (const event of events) { if (event.type === "response.completed") { completedUsage = event.response?.usage ?? null; completedServiceTier = event.response?.service_tier; @@ -635,40 +658,196 @@ export function buildOpenAICompletionsParams( } export function buildOpenAIResponsesParams( - model: Model<"openai-responses">, + model: Model<"openai-responses" | "openai-codex-responses">, context: Context, options?: SimpleStreamOptions, ): ResponseCreateParamsStreaming { const messages: ResponsesInput = normalizeResponsesInputRoles( - convertResponsesMessages(model, context, OPENAI_TOOL_CALL_PROVIDERS), + convertResponsesMessages(model, context, OPENAI_TOOL_CALL_PROVIDERS, { + includeSystemPrompt: model.api !== "openai-codex-responses", + }), ); - const params: ResponseCreateParamsStreaming = { - model: model.id, - input: messages, - stream: true, - store: false, - max_output_tokens: model.maxTokens, - }; + const params: ResponseCreateParamsStreaming = + model.api === "openai-codex-responses" + ? ({ + model: model.id, + input: messages, + stream: true, + store: false, + instructions: context.systemPrompt, + text: { verbosity: "medium" }, + include: ["reasoning.encrypted_content"], + prompt_cache_key: options?.sessionId, + tool_choice: "auto", + parallel_tool_calls: true, + } as ResponseCreateParamsStreaming) + : { + model: model.id, + input: messages, + stream: true, + store: false, + max_output_tokens: model.maxTokens, + }; if (options?.sessionId) { params.prompt_cache_key = options.sessionId; } + const requestServiceTier = getRequestServiceTier(options); + if (requestServiceTier) { + params.service_tier = requestServiceTier; + } + if (context.tools) { - params.tools = convertResponsesTools(context.tools); + params.tools = convertResponsesTools( + context.tools, + model.api === "openai-codex-responses" ? { strict: null } : undefined, + ); } if (model.reasoning && options?.reasoning) { params.reasoning = { - effort: options.reasoning, - summary: "detailed", + effort: + model.api === "openai-codex-responses" + ? clampCodexReasoningEffort(model.id, options.reasoning) + : options.reasoning, + summary: model.api === "openai-codex-responses" ? "auto" : "detailed", }; - params.include = ["reasoning.encrypted_content"]; + if (model.api !== "openai-codex-responses") { + params.include = ["reasoning.encrypted_content"]; + } } return params; } +function getRequestServiceTier( + options?: SimpleStreamOptions, +): ResponseCreateParamsStreaming["service_tier"] | undefined { + return (options as { serviceTier?: ResponseCreateParamsStreaming["service_tier"] } | undefined) + ?.serviceTier; +} + +async function applyPayloadHook( + params: ResponseCreateParamsStreaming, + model: Model<"openai-responses" | "openai-codex-responses">, + options?: SimpleStreamOptions, +): Promise { + const payloadHook = options?.onPayload as + | ((payload: unknown, model: Model) => unknown | undefined | Promise) + | undefined; + const nextParams = await payloadHook?.(params, model); + return nextParams ?? params; +} + +function resolveResponsesBaseUrl(model: Model<"openai-responses" | "openai-codex-responses">) { + if (model.api !== "openai-codex-responses") { + return model.baseUrl; + } + + const normalized = model.baseUrl.replace(/\/+$/, ""); + if (normalized.endsWith("/codex/responses")) { + return normalized.slice(0, -"/responses".length); + } + if (normalized.endsWith("/codex")) { + return normalized; + } + return `${normalized}/codex`; +} + +function buildCodexDefaultHeaders( + token: string, + sessionId: string | undefined, +): Record { + const headers: Record = { + "OpenAI-Beta": "responses=experimental", + "chatgpt-account-id": extractCodexAccountId(token), + originator: "pi", + accept: "text/event-stream", + "content-type": "application/json", + }; + if (sessionId) { + headers.session_id = sessionId; + headers["x-client-request-id"] = sessionId; + } + return headers; +} + +function extractCodexAccountId(token: string): string { + try { + const [, payload] = token.split("."); + if (!payload) { + throw new Error("missing payload"); + } + const parsed = JSON.parse(Buffer.from(payload, "base64url").toString("utf8")) as { + [CODEX_JWT_AUTH_CLAIM]?: { chatgpt_account_id?: unknown }; + }; + const accountId = parsed[CODEX_JWT_AUTH_CLAIM]?.chatgpt_account_id; + if (typeof accountId !== "string" || accountId.length === 0) { + throw new Error("missing account id"); + } + return accountId; + } catch { + throw new Error("Failed to extract accountId from token"); + } +} + +function clampCodexReasoningEffort( + modelId: string, + effort: NonNullable, +): NonNullable { + const id = modelId.includes("/") ? (modelId.split("/").at(-1) ?? modelId) : modelId; + if ( + (id.startsWith("gpt-5.2") || + id.startsWith("gpt-5.3") || + id.startsWith("gpt-5.4") || + id.startsWith("gpt-5.5")) && + effort === "minimal" + ) { + return "low"; + } + if (id === "gpt-5.1" && effort === "xhigh") { + return "high"; + } + if (id === "gpt-5.1-codex-mini") { + return effort === "high" || effort === "xhigh" ? "high" : "medium"; + } + return effort; +} + +async function* mapCodexResponsesEvents( + events: AsyncIterable>, +): AsyncIterable { + for await (const event of events) { + const type = typeof event.type === "string" ? event.type : undefined; + if (!type) continue; + if (type === "error") { + const code = typeof event.code === "string" ? event.code : ""; + const message = typeof event.message === "string" ? event.message : ""; + throw new Error(`Codex error: ${message || code || JSON.stringify(event)}`); + } + if ( + type === "response.done" || + type === "response.completed" || + type === "response.incomplete" + ) { + const response = isRecord(event.response) + ? { + ...event.response, + status: + typeof event.response.status === "string" && + CODEX_RESPONSE_STATUSES.has(event.response.status) + ? event.response.status + : undefined, + } + : event.response; + yield { ...event, type: "response.completed", response } as unknown as ResponseStreamEvent; + return; + } + yield event as unknown as ResponseStreamEvent; + } +} + /** * Resolves the effective compat for completions-API calls. * diff --git a/tests/agent/llm/pi-bridge.codex-service-tier.test.ts b/tests/agent/llm/pi-bridge.codex-service-tier.test.ts index b3f4b53..d18e503 100644 --- a/tests/agent/llm/pi-bridge.codex-service-tier.test.ts +++ b/tests/agent/llm/pi-bridge.codex-service-tier.test.ts @@ -62,6 +62,68 @@ function createStoredUserMessage(): Message { }; } +function createStoredDeepSeekAssistantMessage(): Message { + return { + id: "msg_deepseek_assistant", + sessionId: "sess_1", + seq: 2, + role: "assistant", + messageType: "text", + visibility: "user_visible", + channelMessageId: null, + channelParentMessageId: null, + channelThreadId: null, + provider: "deepseek", + model: "deepseek-v4-flash", + modelApi: "openai-completions", + stopReason: "stop", + errorMessage: null, + payloadJson: JSON.stringify({ + content: [ + { + type: "thinking", + thinking: "Private reasoning from a different provider.", + thinkingSignature: "reasoning_content", + }, + { + type: "text", + text: "Visible answer from the same assistant message.", + }, + ], + }), + tokenInput: null, + tokenOutput: null, + tokenCacheRead: null, + tokenCacheWrite: null, + tokenTotal: null, + usageJson: JSON.stringify({ + input: 10, + output: 5, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 15, + cost: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + total: 0, + }, + }), + createdAt: "2026-03-22T00:00:02.000Z", + }; +} + +function createStoredFollowUpUserMessage(): Message { + return { + ...createStoredUserMessage(), + id: "msg_user_2", + seq: 3, + payloadJson: JSON.stringify({ content: "Continue." }), + createdAt: "2026-03-22T00:00:03.000Z", + }; +} + function createSseResponse(): Response { const messageItem = { type: "message", @@ -148,4 +210,40 @@ describe("pi bridge Codex service tier", () => { service_tier: requestTier, }); }); + + test("sends unique fallback response item ids for cross-provider assistant history", async () => { + const requests: Array<{ url: string; body: unknown }> = []; + vi.stubGlobal( + "fetch", + vi.fn(async (input: Parameters[0], init?: Parameters[1]) => { + requests.push({ + url: String(input), + body: typeof init?.body === "string" ? JSON.parse(init.body) : init?.body, + }); + return createSseResponse(); + }), + ); + + const bridge = new PiBridge(); + await bridge.streamTurn({ + model: createCodexModel(), + systemPrompt: "You are concise.", + compactSummary: null, + messages: [ + createStoredUserMessage(), + createStoredDeepSeekAssistantMessage(), + createStoredFollowUpUserMessage(), + ], + tools: new ToolRegistry(), + signal: new AbortController().signal, + }); + + const body = requests[0]?.body as { input?: Array<{ id?: string; type?: string }> }; + const itemIds = (body.input ?? []) + .filter((entry) => entry.type === "message") + .map((entry) => entry.id); + + expect(itemIds).toHaveLength(2); + expect(new Set(itemIds).size).toBe(itemIds.length); + }); }); From 1069300c18c00a23bea9c88ea72b09025c38f21e Mon Sep 17 00:00:00 2001 From: danielwpz <5139554+danielwpz@users.noreply.github.com> Date: Fri, 29 May 2026 11:56:50 +0800 Subject: [PATCH 2/4] Harden Codex Responses adapter --- src/agent/llm/pi-bridge.ts | 92 ++- src/agent/llm/providers/codex/resolver.ts | 12 +- src/agent/llm/providers/codex/stream.ts | 545 ++++++++++++++++++ src/agent/llm/upstream-openai.ts | 196 +------ .../llm/pi-bridge.codex-service-tier.test.ts | 191 +++++- .../llm/providers.codex-resolver.test.ts | 20 +- .../agent/llm/providers.codex-stream.test.ts | 122 ++++ 7 files changed, 961 insertions(+), 217 deletions(-) create mode 100644 src/agent/llm/providers/codex/stream.ts create mode 100644 tests/agent/llm/providers.codex-stream.test.ts diff --git a/src/agent/llm/pi-bridge.ts b/src/agent/llm/pi-bridge.ts index e7516cd..bdcaf71 100644 --- a/src/agent/llm/pi-bridge.ts +++ b/src/agent/llm/pi-bridge.ts @@ -10,8 +10,10 @@ import { type Api, type AssistantMessage, type AssistantMessageEvent, + type Context, completeSimple, type Model, + type SimpleStreamOptions, type Tool, Type, } from "@mariozechner/pi-ai"; @@ -90,8 +92,15 @@ export interface PiBridgeRunTurnResult { errorMessage?: string; } +export interface ResolvedProviderApiCredential { + apiKey: string; + accountId?: string; +} + +export type ResolvedProviderApiKey = string | ResolvedProviderApiCredential; + export interface ProviderApiKeyResolver { - resolveApiKey(provider: ResolvedProvider): Promise; + resolveApiKey(provider: ResolvedProvider): Promise; } export interface PiBridgeOptions { @@ -251,13 +260,18 @@ export class PiBridge { }); try { - const finalMessage = await completeSimple( - model, - context, - await buildPiStreamOptions(this.providerApiKeyResolver, input.model, input.signal, { + const streamOptions = await buildPiStreamOptions( + this.providerApiKeyResolver, + input.model, + input.signal, + { enableReasoning: true, - }), + }, ); + const finalMessage = + model.api === "openai-codex-responses" + ? await consumeStreamToCompletion(model, context, streamOptions) + : await completeSimple(model, context, streamOptions); const contentSummary = summarizeAssistantContent(finalMessage.content); logger.debug("non-stream llm turn finished", { modelId: input.model.id, @@ -286,27 +300,28 @@ export class PiBridge { }); try { - const finalMessage = await completeSimple( - model, - { - systemPrompt: input.systemPrompt, - messages: [ - { - role: "user" as const, - content: [{ type: "text" as const, text: input.prompt }], - timestamp: Date.now(), - }, - ], - }, - await buildPiStreamOptions( - this.providerApiKeyResolver, - input.model, - input.signal ?? new AbortController().signal, + const context = { + systemPrompt: input.systemPrompt, + messages: [ { - enableReasoning: false, + role: "user" as const, + content: [{ type: "text" as const, text: input.prompt }], + timestamp: Date.now(), }, - ), + ], + }; + const streamOptions = await buildPiStreamOptions( + this.providerApiKeyResolver, + input.model, + input.signal ?? new AbortController().signal, + { + enableReasoning: false, + }, ); + const finalMessage = + model.api === "openai-codex-responses" + ? await consumeStreamToCompletion(model, context, streamOptions) + : await completeSimple(model, context, streamOptions); const normalized = normalizeAssistantResult(finalMessage, "complete"); logger.debug("compaction llm call finished", { @@ -365,6 +380,18 @@ export class PiAgentModelRunner implements AgentModelRunner, CompactionModelRunn } } +async function consumeStreamToCompletion( + model: Model, + context: Context, + options: SimpleStreamOptions, +): Promise { + const stream = streamWithNormalizedUpstreamUsage(model, context, options); + for await (const _event of stream) { + // Drain the stream so the adapter can assemble the final AssistantMessage. + } + return await stream.result(); +} + type LlmStreamWatchdogTimeoutKind = "first_response" | "stream_idle"; class LlmStreamWatchdog { @@ -621,6 +648,7 @@ async function buildPiStreamOptions( sessionId: string; maxTokens: number; apiKey?: string; + codexAccountId?: string; reasoning?: "minimal" | "low" | "medium" | "high" | "xhigh"; serviceTier?: RequestServiceTier; onPayload?: ( @@ -633,11 +661,16 @@ async function buildPiStreamOptions( maxTokens: model.maxOutputTokens, }; let resolvedApiKey: string | undefined; + let resolvedCodexAccountId: string | undefined; try { - resolvedApiKey = - (await providerApiKeyResolver?.resolveApiKey(model.provider)) ?? - model.provider.apiKey ?? - undefined; + const resolvedCredential = await providerApiKeyResolver?.resolveApiKey(model.provider); + if (typeof resolvedCredential === "string") { + resolvedApiKey = resolvedCredential; + } else if (resolvedCredential != null) { + resolvedApiKey = resolvedCredential.apiKey; + resolvedCodexAccountId = resolvedCredential.accountId; + } + resolvedApiKey ??= model.provider.apiKey ?? undefined; } catch (error) { throw enrichCodexFetchFailure(error, model, "auth"); } @@ -647,6 +680,9 @@ async function buildPiStreamOptions( apiKey, }); } + if (resolvedCodexAccountId != null && resolvedCodexAccountId.length > 0) { + options.codexAccountId = resolvedCodexAccountId; + } if (input.enableReasoning && model.reasoning?.enabled) { options.reasoning = model.reasoning.effort ?? DEFAULT_REASONING_LEVEL; diff --git a/src/agent/llm/providers/codex/resolver.ts b/src/agent/llm/providers/codex/resolver.ts index 53f6e42..1867525 100644 --- a/src/agent/llm/providers/codex/resolver.ts +++ b/src/agent/llm/providers/codex/resolver.ts @@ -1,6 +1,6 @@ import { refreshOpenAICodexToken } from "@mariozechner/pi-ai/oauth"; import type { ResolvedProvider } from "@/src/agent/llm/models.js"; -import type { ProviderApiKeyResolver } from "@/src/agent/llm/pi-bridge.js"; +import type { ProviderApiKeyResolver, ResolvedProviderApiKey } from "@/src/agent/llm/pi-bridge.js"; import { withFileLock } from "@/src/shared/file-lock.js"; import { CODEX_CREDENTIALS_PATH } from "@/src/shared/paths.js"; import { @@ -14,7 +14,7 @@ import { readStoredCodexCredential, writeStoredCodexCredential } from "./store.j // This prevents accidentally forwarding ChatGPT/Codex bearer tokens to arbitrary // third-party endpoints via a misconfigured provider. export class CodexProviderApiKeyResolver implements ProviderApiKeyResolver { - async resolveApiKey(provider: ResolvedProvider): Promise { + async resolveApiKey(provider: ResolvedProvider): Promise { if (provider.authSource !== "codex-local") { return provider.apiKey; } @@ -25,7 +25,13 @@ export class CodexProviderApiKeyResolver implements ProviderApiKeyResolver { } const credential = await resolveCodexCredential(); - return credential?.accessToken; + if (credential == null) { + return undefined; + } + return { + apiKey: credential.accessToken, + ...(credential.accountId == null ? {} : { accountId: credential.accountId }), + }; } } diff --git a/src/agent/llm/providers/codex/stream.ts b/src/agent/llm/providers/codex/stream.ts new file mode 100644 index 0000000..f581524 --- /dev/null +++ b/src/agent/llm/providers/codex/stream.ts @@ -0,0 +1,545 @@ +import type { + Api, + AssistantMessage, + AssistantMessageEventStream, + Context, + Model, + SimpleStreamOptions, + Tool, + Usage, +} from "@mariozechner/pi-ai"; +import { createAssistantMessageEventStream } from "@mariozechner/pi-ai"; +import OpenAI from "openai"; +import type { + ResponseCreateParamsStreaming, + ResponseStreamEvent, +} from "openai/resources/responses/responses.js"; +import { buildAgentLlmRawErrorPayload } from "@/src/agent/llm/errors.js"; +import { + convertResponsesMessages, + convertResponsesTools, + processResponsesStream, +} from "@/src/agent/llm/pi-ai-openai-responses-shared.js"; + +type ResponsesInput = Exclude; +type RequestServiceTier = ResponseCreateParamsStreaming["service_tier"]; + +export type CodexResponsesStreamOptions = SimpleStreamOptions & { + codexAccountId?: string; + serviceTier?: RequestServiceTier; + onPayload?: ( + payload: unknown, + model: Model, + ) => unknown | undefined | Promise; +}; + +type AssistantMessageWithRawError = AssistantMessage & { + pokoclawRawError?: ReturnType; +}; + +const CODEX_TOOL_CALL_PROVIDERS = new Set(["openai", "openai-codex", "opencode"]); +const CODEX_JWT_AUTH_CLAIM = "https://api.openai.com/auth"; +const CODEX_RESPONSE_STATUSES = new Set([ + "completed", + "incomplete", + "failed", + "cancelled", + "queued", + "in_progress", +]); +const CODEX_FAILED_RESPONSE_STATUSES = new Set(["failed", "cancelled"]); + +export function streamOpenAICodexResponsesWithLocalConverter( + model: Model<"openai-codex-responses">, + context: Context, + options?: CodexResponsesStreamOptions, +): AssistantMessageEventStream { + const stream = createAssistantMessageEventStream(); + + void (async () => { + const output: AssistantMessage = { + role: "assistant", + content: [], + api: model.api, + provider: model.provider, + model: model.id, + usage: zeroUsage(), + stopReason: "stop", + timestamp: Date.now(), + }; + + try { + const apiKey = options?.apiKey; + if (!apiKey) { + throw new Error("OpenAI-compatible API key is missing"); + } + + const requestServiceTier = options?.serviceTier; + const client = new OpenAI({ + apiKey, + baseURL: resolveCodexResponsesBaseUrl(model), + defaultHeaders: buildCodexDefaultHeaders( + apiKey, + options?.sessionId, + options?.codexAccountId, + ), + dangerouslyAllowBrowser: true, + }); + + const params = await applyPayloadHook( + buildOpenAICodexResponsesParams(model, context, options), + model, + options, + ); + const rawStream = await client.responses.create( + params as ResponseCreateParamsStreaming, + options?.signal ? { signal: options.signal } : undefined, + ); + + stream.push({ type: "start", partial: output }); + await processResponsesStream( + withResolvedCodexServiceTier( + mapCodexResponsesEvents(rawStream as AsyncIterable>), + requestServiceTier, + ), + output, + stream, + model, + { + serviceTier: requestServiceTier, + applyServiceTierPricing: (usage, serviceTier) => + applyCodexServiceTierPricing(model, usage, serviceTier), + }, + ); + + if (options?.signal?.aborted) { + throw new Error("Request was aborted"); + } + + if (output.stopReason === "aborted" || output.stopReason === "error") { + throw new Error("An unknown error occurred"); + } + + stream.push({ type: "done", reason: output.stopReason, message: output }); + stream.end(); + } catch (error) { + output.stopReason = options?.signal?.aborted ? "aborted" : "error"; + output.errorMessage = error instanceof Error ? error.message : JSON.stringify(error); + (output as AssistantMessageWithRawError).pokoclawRawError = + buildAgentLlmRawErrorPayload(error); + stream.push({ type: "error", reason: output.stopReason, error: output }); + stream.end(); + } + })(); + + return stream; +} + +export function buildOpenAICodexResponsesParams( + model: Model<"openai-codex-responses">, + context: Context, + options?: CodexResponsesStreamOptions, +): ResponseCreateParamsStreaming { + const input: ResponsesInput = normalizeResponsesInputRoles( + convertResponsesMessages(model, context, CODEX_TOOL_CALL_PROVIDERS, { + includeSystemPrompt: false, + }), + ); + const params: ResponseCreateParamsStreaming = { + model: model.id, + input, + stream: true, + store: false, + instructions: context.systemPrompt, + text: { verbosity: "medium" }, + include: ["reasoning.encrypted_content"], + max_output_tokens: model.maxTokens, + } as ResponseCreateParamsStreaming; + + if (options?.sessionId) { + params.prompt_cache_key = options.sessionId; + } + + if (options?.serviceTier) { + params.service_tier = options.serviceTier; + } + + if (context.tools != null && context.tools.length > 0) { + params.tools = convertResponsesTools(context.tools as Tool[], { strict: null }); + params.tool_choice = "auto"; + params.parallel_tool_calls = true; + } + + if (model.reasoning && options?.reasoning) { + params.reasoning = { + effort: clampCodexReasoningEffort(model.id, options.reasoning), + summary: "auto", + }; + } + + return params; +} + +export async function* mapCodexResponsesEvents( + events: AsyncIterable>, +): AsyncIterable { + for await (const event of events) { + const type = typeof event.type === "string" ? event.type : undefined; + if (!type) continue; + + switch (type) { + case "error": + throw new Error(`Codex error: ${readCodexErrorMessage(event)}`); + case "response.done": + case "response.completed": + case "response.incomplete": + yield normalizeCodexTerminalEvent(event, type); + continue; + case "response.output_item.added": + case "response.output_item.done": + validateCodexOutputItemEvent(event, type); + yield event as unknown as ResponseStreamEvent; + continue; + case "response.content_part.added": + validateCodexContentPartEvent(event, type); + yield event as unknown as ResponseStreamEvent; + continue; + case "response.output_text.delta": + case "response.refusal.delta": + case "response.function_call_arguments.delta": + case "response.reasoning_summary_text.delta": + requireStringField(event, "delta", type, "delta"); + yield event as unknown as ResponseStreamEvent; + continue; + case "response.function_call_arguments.done": + requireStringField(event, "arguments", type, "arguments"); + yield event as unknown as ResponseStreamEvent; + continue; + case "response.reasoning_summary_part.added": + validateReasoningSummaryPartEvent(event, type); + yield event as unknown as ResponseStreamEvent; + continue; + default: + yield event as unknown as ResponseStreamEvent; + continue; + } + } +} + +async function* withResolvedCodexServiceTier( + events: AsyncIterable, + requestServiceTier: RequestServiceTier | undefined, +): AsyncIterable { + for await (const event of events) { + if ( + event.type !== "response.completed" || + event.response == null || + requestServiceTier == null + ) { + yield event; + continue; + } + + const responseServiceTier = event.response.service_tier; + const serviceTier = + responseServiceTier == null || responseServiceTier === "default" + ? requestServiceTier + : responseServiceTier; + yield { + ...event, + response: { + ...event.response, + service_tier: serviceTier, + }, + }; + } +} + +async function applyPayloadHook( + params: ResponseCreateParamsStreaming, + model: Model<"openai-codex-responses">, + options?: CodexResponsesStreamOptions, +): Promise { + const nextParams = await options?.onPayload?.(params, model); + return nextParams ?? params; +} + +function resolveCodexResponsesBaseUrl(model: Model<"openai-codex-responses">): string { + const normalized = model.baseUrl.replace(/\/+$/, ""); + if (normalized.endsWith("/codex/responses")) { + return normalized.slice(0, -"/responses".length); + } + if (normalized.endsWith("/codex")) { + return normalized; + } + return `${normalized}/codex`; +} + +function buildCodexDefaultHeaders( + token: string, + sessionId: string | undefined, + accountId: string | undefined, +): Record { + const resolvedAccountId = + accountId != null && accountId.length > 0 ? accountId : extractCodexAccountId(token); + const headers: Record = { + "OpenAI-Beta": "responses=experimental", + "chatgpt-account-id": resolvedAccountId, + originator: "pi", + accept: "text/event-stream", + "content-type": "application/json", + }; + if (sessionId) { + headers.session_id = sessionId; + headers["x-client-request-id"] = sessionId; + } + return headers; +} + +function extractCodexAccountId(token: string): string { + try { + const [, payload] = token.split("."); + if (!payload) { + throw new Error("missing payload"); + } + const parsed = JSON.parse(Buffer.from(payload, "base64url").toString("utf8")) as { + [CODEX_JWT_AUTH_CLAIM]?: { chatgpt_account_id?: unknown }; + }; + const accountId = parsed[CODEX_JWT_AUTH_CLAIM]?.chatgpt_account_id; + if (typeof accountId !== "string" || accountId.length === 0) { + throw new Error("missing account id"); + } + return accountId; + } catch { + throw new Error("Failed to extract accountId from token"); + } +} + +function clampCodexReasoningEffort( + modelId: string, + effort: NonNullable, +): NonNullable { + const id = modelId.includes("/") ? (modelId.split("/").at(-1) ?? modelId) : modelId; + if ( + (id.startsWith("gpt-5.2") || + id.startsWith("gpt-5.3") || + id.startsWith("gpt-5.4") || + id.startsWith("gpt-5.5")) && + effort === "minimal" + ) { + return "low"; + } + if (id === "gpt-5.1" && effort === "xhigh") { + return "high"; + } + if (id === "gpt-5.1-codex-mini") { + return effort === "high" || effort === "xhigh" ? "high" : "medium"; + } + return effort; +} + +function normalizeCodexTerminalEvent( + event: Record, + type: "response.done" | "response.completed" | "response.incomplete", +): ResponseStreamEvent { + const response = requireRecordField(event, "response", type, "response"); + const fallbackStatus = + type === "response.completed" + ? "completed" + : type === "response.incomplete" + ? "incomplete" + : undefined; + const status = normalizeCodexResponseStatus(response.status, type) ?? fallbackStatus; + const normalizedResponse = { + ...response, + ...(status == null ? {} : { status }), + }; + const normalizedType = + status != null && CODEX_FAILED_RESPONSE_STATUSES.has(status) + ? "response.failed" + : "response.completed"; + return { + ...event, + type: normalizedType, + response: normalizedResponse, + } as unknown as ResponseStreamEvent; +} + +function normalizeCodexResponseStatus(value: unknown, eventType: string): string | undefined { + if (value == null) { + return undefined; + } + if (typeof value !== "string" || !CODEX_RESPONSE_STATUSES.has(value)) { + throw invalidCodexEvent(eventType, "response.status must be a known Codex status"); + } + return value; +} + +function validateCodexOutputItemEvent(event: Record, eventType: string): void { + const item = requireRecordField(event, "item", eventType, "item"); + const itemType = requireStringField(item, "type", eventType, "item.type"); + + if (itemType === "message") { + requireStringField(item, "id", eventType, "item.id"); + validateMessageContentParts(item.content, eventType, "item.content"); + return; + } + + if (itemType === "function_call") { + requireStringField(item, "id", eventType, "item.id"); + requireStringField(item, "call_id", eventType, "item.call_id"); + requireStringField(item, "name", eventType, "item.name"); + if (item.arguments != null && typeof item.arguments !== "string") { + throw invalidCodexEvent(eventType, "item.arguments must be a string"); + } + return; + } + + if (itemType === "reasoning") { + if (item.id != null && typeof item.id !== "string") { + throw invalidCodexEvent(eventType, "item.id must be a string"); + } + if (item.summary != null && !Array.isArray(item.summary)) { + throw invalidCodexEvent(eventType, "item.summary must be an array"); + } + } +} + +function validateCodexContentPartEvent(event: Record, eventType: string): void { + const part = requireRecordField(event, "part", eventType, "part"); + validateMessagePart(part, eventType, "part"); +} + +function validateReasoningSummaryPartEvent( + event: Record, + eventType: string, +): void { + const part = requireRecordField(event, "part", eventType, "part"); + requireStringField(part, "text", eventType, "part.text"); +} + +function validateMessageContentParts(value: unknown, eventType: string, path: string): void { + if (value == null) { + return; + } + if (!Array.isArray(value)) { + throw invalidCodexEvent(eventType, `${path} must be an array`); + } + value.forEach((part, index) => { + if (!isRecord(part)) { + throw invalidCodexEvent(eventType, `${path}[${index}] must be an object`); + } + validateMessagePart(part, eventType, `${path}[${index}]`); + }); +} + +function validateMessagePart(part: Record, eventType: string, path: string): void { + const partType = requireStringField(part, "type", eventType, `${path}.type`); + if (partType === "output_text") { + requireStringField(part, "text", eventType, `${path}.text`); + } else if (partType === "refusal") { + requireStringField(part, "refusal", eventType, `${path}.refusal`); + } +} + +function readCodexErrorMessage(event: Record): string { + const code = typeof event.code === "string" ? event.code : ""; + const message = typeof event.message === "string" ? event.message : ""; + return message || code || JSON.stringify(event); +} + +function requireRecordField( + value: Record, + key: string, + eventType: string, + path: string, +): Record { + const candidate = value[key]; + if (!isRecord(candidate)) { + throw invalidCodexEvent(eventType, `${path} must be an object`); + } + return candidate; +} + +function requireStringField( + value: Record, + key: string, + eventType: string, + path: string, +): string { + const candidate = value[key]; + if (typeof candidate !== "string") { + throw invalidCodexEvent(eventType, `${path} must be a string`); + } + return candidate; +} + +function invalidCodexEvent(eventType: string, message: string): Error { + return new Error(`Invalid Codex ${eventType} event: ${message}`); +} + +function isRecord(value: unknown): value is Record & { role?: unknown } { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function normalizeResponsesInputRoles(input: ResponsesInput): ResponsesInput { + if (!Array.isArray(input)) { + return input; + } + + let changed = false; + const normalized = input.map((item) => { + if (!isRecord(item) || item.role !== "developer") { + return item; + } + + changed = true; + return { + ...item, + role: "system" as const, + }; + }); + + return changed ? normalized : input; +} + +function applyCodexServiceTierPricing( + model: Model<"openai-codex-responses">, + usage: Usage, + serviceTier: RequestServiceTier | undefined, +): void { + const multiplier = getCodexServiceTierCostMultiplier(model.id, serviceTier); + if (multiplier === 1) return; + + usage.cost.input *= multiplier; + usage.cost.output *= multiplier; + usage.cost.cacheRead *= multiplier; + usage.cost.cacheWrite *= multiplier; + usage.cost.total = + usage.cost.input + usage.cost.output + usage.cost.cacheRead + usage.cost.cacheWrite; +} + +function getCodexServiceTierCostMultiplier( + modelId: string, + serviceTier: RequestServiceTier | undefined, +): number { + switch (serviceTier) { + case "flex": + return 0.5; + case "priority": + return modelId === "gpt-5.5" ? 2.5 : 2; + default: + return 1; + } +} + +function zeroUsage(): Usage { + return { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }; +} diff --git a/src/agent/llm/upstream-openai.ts b/src/agent/llm/upstream-openai.ts index b311dc0..0ae42ac 100644 --- a/src/agent/llm/upstream-openai.ts +++ b/src/agent/llm/upstream-openai.ts @@ -33,6 +33,10 @@ import { convertResponsesTools, processResponsesStream, } from "@/src/agent/llm/pi-ai-openai-responses-shared.js"; +import { + type CodexResponsesStreamOptions, + streamOpenAICodexResponsesWithLocalConverter, +} from "@/src/agent/llm/providers/codex/stream.js"; import { appendCappedTextTail } from "@/src/shared/capped-text.js"; type CostedModel = Pick, "api" | "baseUrl" | "cost">; @@ -156,15 +160,6 @@ const DEFAULT_COMPAT: ResolvedOpenAICompletionsCompat = { const OPENAI_TOOL_CALL_PROVIDERS = new Set(["openai", "openai-codex", "opencode"]); const STREAM_REASONING_CONTENT_MAX_CHARS = 100_000; const STREAM_REASONING_TRUNCATION_PREFIX = "...[earlier reasoning truncated]\n"; -const CODEX_RESPONSE_STATUSES = new Set([ - "completed", - "incomplete", - "failed", - "cancelled", - "queued", - "in_progress", -]); -const CODEX_JWT_AUTH_CLAIM = "https://api.openai.com/auth"; export function supportsUpstreamCostParser(model: Pick, "baseUrl">): boolean { return COST_PARSERS.some((parser) => parser.supports(model)); @@ -179,9 +174,6 @@ export function shouldUseCustomOpenAICompletionsStream( export function shouldUseCustomOpenAIResponsesStream( model: Pick, "api" | "baseUrl">, ): boolean { - if (model.api === "openai-codex-responses") { - return true; - } return model.api === "openai-responses" && supportsUpstreamCostParser(model); } @@ -232,6 +224,14 @@ export function streamWithNormalizedUpstreamUsage( ); } + if (model.api === "openai-codex-responses") { + return streamOpenAICodexResponsesWithLocalConverter( + model as Model<"openai-codex-responses">, + context, + options as CodexResponsesStreamOptions | undefined, + ); + } + if (shouldUseCustomOpenAIResponsesStream(model)) { return streamOpenAIResponsesWithUpstreamUsage( model as Model<"openai-responses">, @@ -523,7 +523,7 @@ function appendStreamReasoning(existing: string, delta: string): string { } function streamOpenAIResponsesWithUpstreamUsage( - model: Model<"openai-responses" | "openai-codex-responses">, + model: Model<"openai-responses">, context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream { @@ -549,10 +549,7 @@ function streamOpenAIResponsesWithUpstreamUsage( const client = new OpenAI({ apiKey, - baseURL: resolveResponsesBaseUrl(model), - ...(model.api === "openai-codex-responses" - ? { defaultHeaders: buildCodexDefaultHeaders(apiKey, options?.sessionId) } - : {}), + baseURL: model.baseUrl, dangerouslyAllowBrowser: true, }); @@ -569,10 +566,7 @@ function streamOpenAIResponsesWithUpstreamUsage( let completedUsage: unknown = null; let completedServiceTier: ResponseCreateParamsStreaming["service_tier"] | undefined; async function* tappedStream(): AsyncIterable { - const events = - model.api === "openai-codex-responses" - ? mapCodexResponsesEvents(rawStream as AsyncIterable>) - : (rawStream as AsyncIterable); + const events = rawStream as AsyncIterable; for await (const event of events) { if (event.type === "response.completed") { completedUsage = event.response?.usage ?? null; @@ -658,36 +652,20 @@ export function buildOpenAICompletionsParams( } export function buildOpenAIResponsesParams( - model: Model<"openai-responses" | "openai-codex-responses">, + model: Model<"openai-responses">, context: Context, options?: SimpleStreamOptions, ): ResponseCreateParamsStreaming { const messages: ResponsesInput = normalizeResponsesInputRoles( - convertResponsesMessages(model, context, OPENAI_TOOL_CALL_PROVIDERS, { - includeSystemPrompt: model.api !== "openai-codex-responses", - }), + convertResponsesMessages(model, context, OPENAI_TOOL_CALL_PROVIDERS), ); - const params: ResponseCreateParamsStreaming = - model.api === "openai-codex-responses" - ? ({ - model: model.id, - input: messages, - stream: true, - store: false, - instructions: context.systemPrompt, - text: { verbosity: "medium" }, - include: ["reasoning.encrypted_content"], - prompt_cache_key: options?.sessionId, - tool_choice: "auto", - parallel_tool_calls: true, - } as ResponseCreateParamsStreaming) - : { - model: model.id, - input: messages, - stream: true, - store: false, - max_output_tokens: model.maxTokens, - }; + const params: ResponseCreateParamsStreaming = { + model: model.id, + input: messages, + stream: true, + store: false, + max_output_tokens: model.maxTokens, + }; if (options?.sessionId) { params.prompt_cache_key = options.sessionId; @@ -699,23 +677,15 @@ export function buildOpenAIResponsesParams( } if (context.tools) { - params.tools = convertResponsesTools( - context.tools, - model.api === "openai-codex-responses" ? { strict: null } : undefined, - ); + params.tools = convertResponsesTools(context.tools); } if (model.reasoning && options?.reasoning) { params.reasoning = { - effort: - model.api === "openai-codex-responses" - ? clampCodexReasoningEffort(model.id, options.reasoning) - : options.reasoning, - summary: model.api === "openai-codex-responses" ? "auto" : "detailed", + effort: options.reasoning, + summary: "detailed", }; - if (model.api !== "openai-codex-responses") { - params.include = ["reasoning.encrypted_content"]; - } + params.include = ["reasoning.encrypted_content"]; } return params; @@ -730,7 +700,7 @@ function getRequestServiceTier( async function applyPayloadHook( params: ResponseCreateParamsStreaming, - model: Model<"openai-responses" | "openai-codex-responses">, + model: Model<"openai-responses">, options?: SimpleStreamOptions, ): Promise { const payloadHook = options?.onPayload as @@ -740,114 +710,6 @@ async function applyPayloadHook( return nextParams ?? params; } -function resolveResponsesBaseUrl(model: Model<"openai-responses" | "openai-codex-responses">) { - if (model.api !== "openai-codex-responses") { - return model.baseUrl; - } - - const normalized = model.baseUrl.replace(/\/+$/, ""); - if (normalized.endsWith("/codex/responses")) { - return normalized.slice(0, -"/responses".length); - } - if (normalized.endsWith("/codex")) { - return normalized; - } - return `${normalized}/codex`; -} - -function buildCodexDefaultHeaders( - token: string, - sessionId: string | undefined, -): Record { - const headers: Record = { - "OpenAI-Beta": "responses=experimental", - "chatgpt-account-id": extractCodexAccountId(token), - originator: "pi", - accept: "text/event-stream", - "content-type": "application/json", - }; - if (sessionId) { - headers.session_id = sessionId; - headers["x-client-request-id"] = sessionId; - } - return headers; -} - -function extractCodexAccountId(token: string): string { - try { - const [, payload] = token.split("."); - if (!payload) { - throw new Error("missing payload"); - } - const parsed = JSON.parse(Buffer.from(payload, "base64url").toString("utf8")) as { - [CODEX_JWT_AUTH_CLAIM]?: { chatgpt_account_id?: unknown }; - }; - const accountId = parsed[CODEX_JWT_AUTH_CLAIM]?.chatgpt_account_id; - if (typeof accountId !== "string" || accountId.length === 0) { - throw new Error("missing account id"); - } - return accountId; - } catch { - throw new Error("Failed to extract accountId from token"); - } -} - -function clampCodexReasoningEffort( - modelId: string, - effort: NonNullable, -): NonNullable { - const id = modelId.includes("/") ? (modelId.split("/").at(-1) ?? modelId) : modelId; - if ( - (id.startsWith("gpt-5.2") || - id.startsWith("gpt-5.3") || - id.startsWith("gpt-5.4") || - id.startsWith("gpt-5.5")) && - effort === "minimal" - ) { - return "low"; - } - if (id === "gpt-5.1" && effort === "xhigh") { - return "high"; - } - if (id === "gpt-5.1-codex-mini") { - return effort === "high" || effort === "xhigh" ? "high" : "medium"; - } - return effort; -} - -async function* mapCodexResponsesEvents( - events: AsyncIterable>, -): AsyncIterable { - for await (const event of events) { - const type = typeof event.type === "string" ? event.type : undefined; - if (!type) continue; - if (type === "error") { - const code = typeof event.code === "string" ? event.code : ""; - const message = typeof event.message === "string" ? event.message : ""; - throw new Error(`Codex error: ${message || code || JSON.stringify(event)}`); - } - if ( - type === "response.done" || - type === "response.completed" || - type === "response.incomplete" - ) { - const response = isRecord(event.response) - ? { - ...event.response, - status: - typeof event.response.status === "string" && - CODEX_RESPONSE_STATUSES.has(event.response.status) - ? event.response.status - : undefined, - } - : event.response; - yield { ...event, type: "response.completed", response } as unknown as ResponseStreamEvent; - return; - } - yield event as unknown as ResponseStreamEvent; - } -} - /** * Resolves the effective compat for completions-API calls. * diff --git a/tests/agent/llm/pi-bridge.codex-service-tier.test.ts b/tests/agent/llm/pi-bridge.codex-service-tier.test.ts index d18e503..f7ab1dd 100644 --- a/tests/agent/llm/pi-bridge.codex-service-tier.test.ts +++ b/tests/agent/llm/pi-bridge.codex-service-tier.test.ts @@ -16,6 +16,20 @@ function createCodexAccessToken(): string { return `${header}.${payload}.signature`; } +interface CapturedRequest { + url: string; + body: Record; + headers: Headers; +} + +function captureRequest(input: Parameters[0], init?: Parameters[1]) { + return { + url: String(input), + body: typeof init?.body === "string" ? (JSON.parse(init.body) as Record) : {}, + headers: new Headers(init?.headers), + }; +} + function createCodexModel(overrides?: Partial): ResolvedModel { return { id: "codex-gpt5.5-fast", @@ -124,6 +138,13 @@ function createStoredFollowUpUserMessage(): Message { }; } +function createSseResponseFromEvents(events: Array>): Response { + return new Response(events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join(""), { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); +} + function createSseResponse(): Response { const messageItem = { type: "message", @@ -150,7 +171,7 @@ function createSseResponse(): Response { item: messageItem, }, { - type: "response.completed", + type: "response.done", response: { id: "resp_test", status: "completed", @@ -165,10 +186,7 @@ function createSseResponse(): Response { }, ]; - return new Response(events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join(""), { - status: 200, - headers: { "content-type": "text/event-stream" }, - }); + return createSseResponseFromEvents(events); } describe("pi bridge Codex service tier", () => { @@ -180,14 +198,11 @@ describe("pi bridge Codex service tier", () => { ["fast", "priority"], ["flex", "flex"], ] as const)("sends configured %s service tier as %s in the final streaming Codex HTTP request body", async (configuredTier, requestTier) => { - const requests: Array<{ url: string; body: unknown }> = []; + const requests: CapturedRequest[] = []; vi.stubGlobal( "fetch", vi.fn(async (input: Parameters[0], init?: Parameters[1]) => { - requests.push({ - url: String(input), - body: typeof init?.body === "string" ? JSON.parse(init.body) : init?.body, - }); + requests.push(captureRequest(input, init)); return createSseResponse(); }), ); @@ -208,18 +223,19 @@ describe("pi bridge Codex service tier", () => { expect(requests[0]?.body).toMatchObject({ model: "gpt-5.5", service_tier: requestTier, + max_output_tokens: 16_384, }); + expect(requests[0]?.body).not.toHaveProperty("tool_choice"); + expect(requests[0]?.body).not.toHaveProperty("parallel_tool_calls"); + expect(requests[0]?.body).not.toHaveProperty("tools"); }); test("sends unique fallback response item ids for cross-provider assistant history", async () => { - const requests: Array<{ url: string; body: unknown }> = []; + const requests: CapturedRequest[] = []; vi.stubGlobal( "fetch", vi.fn(async (input: Parameters[0], init?: Parameters[1]) => { - requests.push({ - url: String(input), - body: typeof init?.body === "string" ? JSON.parse(init.body) : init?.body, - }); + requests.push(captureRequest(input, init)); return createSseResponse(); }), ); @@ -246,4 +262,149 @@ describe("pi bridge Codex service tier", () => { expect(itemIds).toHaveLength(2); expect(new Set(itemIds).size).toBe(itemIds.length); }); + + test("uses the local Codex converter for non-streaming bridge turns", async () => { + const requests: CapturedRequest[] = []; + vi.stubGlobal( + "fetch", + vi.fn(async (input: Parameters[0], init?: Parameters[1]) => { + requests.push(captureRequest(input, init)); + return createSseResponse(); + }), + ); + + const bridge = new PiBridge(); + const result = await bridge.completeTurn({ + model: createCodexModel(), + systemPrompt: "You are concise.", + compactSummary: null, + messages: [ + createStoredUserMessage(), + createStoredDeepSeekAssistantMessage(), + createStoredFollowUpUserMessage(), + ], + tools: new ToolRegistry(), + signal: new AbortController().signal, + }); + + expect(result.content).toMatchObject([{ type: "text", text: "ok" }]); + expect(requests).toHaveLength(1); + expect(requests[0]?.body).toMatchObject({ + model: "gpt-5.5", + stream: true, + max_output_tokens: 16_384, + }); + const input = requests[0]?.body.input; + expect(Array.isArray(input)).toBe(true); + if (!Array.isArray(input)) { + throw new Error("expected responses input array"); + } + const itemIds = input + .flatMap((item) => (typeof item === "object" && item != null ? [item] : [])) + .map((item) => (item as { id?: unknown }).id) + .filter((id): id is string => typeof id === "string"); + expect(new Set(itemIds).size).toBe(itemIds.length); + }); + + test("uses resolver-provided Codex account id when the bearer token is opaque", async () => { + const requests: CapturedRequest[] = []; + vi.stubGlobal( + "fetch", + vi.fn(async (input: Parameters[0], init?: Parameters[1]) => { + requests.push(captureRequest(input, init)); + return createSseResponse(); + }), + ); + + const bridge = new PiBridge({ + async resolveApiKey() { + return { apiKey: "opaque-codex-access-token", accountId: "acct_from_store" }; + }, + }); + + await bridge.streamTurn({ + model: createCodexModel({ + provider: { + id: "openai_codex", + api: "openai-codex-responses", + authSource: "codex-local", + }, + }), + systemPrompt: "You are concise.", + compactSummary: null, + messages: [createStoredUserMessage()], + tools: new ToolRegistry(), + signal: new AbortController().signal, + }); + + expect(requests).toHaveLength(1); + expect(requests[0]?.headers.get("authorization")).toBe("Bearer opaque-codex-access-token"); + expect(requests[0]?.headers.get("chatgpt-account-id")).toBe("acct_from_store"); + }); + + test("preserves Codex failed done event details", async () => { + vi.stubGlobal( + "fetch", + vi.fn(async () => + createSseResponseFromEvents([ + { + type: "response.done", + response: { + id: "resp_failed", + status: "failed", + error: { + code: "invalid_request_error", + message: "Codex rejected duplicate item ids", + }, + }, + }, + ]), + ), + ); + + const bridge = new PiBridge(); + await expect( + bridge.streamTurn({ + model: createCodexModel(), + systemPrompt: "You are concise.", + compactSummary: null, + messages: [createStoredUserMessage()], + tools: new ToolRegistry(), + signal: new AbortController().signal, + }), + ).rejects.toThrow("invalid_request_error: Codex rejected duplicate item ids"); + }); + + test("rejects malformed Codex stream events before response processing", async () => { + vi.stubGlobal( + "fetch", + vi.fn(async () => + createSseResponseFromEvents([ + { + type: "response.output_text.delta", + delta: 123, + }, + { + type: "response.done", + response: { + id: "resp_test", + status: "completed", + }, + }, + ]), + ), + ); + + const bridge = new PiBridge(); + await expect( + bridge.streamTurn({ + model: createCodexModel(), + systemPrompt: "You are concise.", + compactSummary: null, + messages: [createStoredUserMessage()], + tools: new ToolRegistry(), + signal: new AbortController().signal, + }), + ).rejects.toThrow("Invalid Codex response.output_text.delta event"); + }); }); diff --git a/tests/agent/llm/providers.codex-resolver.test.ts b/tests/agent/llm/providers.codex-resolver.test.ts index f1be200..96f3811 100644 --- a/tests/agent/llm/providers.codex-resolver.test.ts +++ b/tests/agent/llm/providers.codex-resolver.test.ts @@ -89,7 +89,10 @@ describe("codex provider api key resolver", () => { authSource: "codex-local", }; - await expect(resolver.resolveApiKey(provider)).resolves.toBe("external-access"); + await expect(resolver.resolveApiKey(provider)).resolves.toEqual({ + apiKey: "external-access", + accountId: "acct_1", + }); expect(writeStoredCodexCredentialMock).toHaveBeenCalledOnce(); }); @@ -111,7 +114,10 @@ describe("codex provider api key resolver", () => { authSource: "codex-local", }; - await expect(resolver.resolveApiKey(provider)).resolves.toBe("external-access"); + await expect(resolver.resolveApiKey(provider)).resolves.toEqual({ + apiKey: "external-access", + accountId: "acct_1", + }); expect(writeStoredCodexCredentialMock).toHaveBeenCalledWith( expect.objectContaining({ sourceFingerprint: "keychain:cli|abcd1234" }), ); @@ -141,7 +147,10 @@ describe("codex provider api key resolver", () => { authSource: "codex-local", }; - await expect(resolver.resolveApiKey(provider)).resolves.toBe("external-access"); + await expect(resolver.resolveApiKey(provider)).resolves.toEqual({ + apiKey: "external-access", + accountId: "acct_new", + }); expect(writeStoredCodexCredentialMock).toHaveBeenCalledOnce(); }); @@ -197,7 +206,10 @@ describe("codex provider api key resolver", () => { authSource: "codex-local", }; - await expect(resolver.resolveApiKey(provider)).resolves.toBe("refreshed-access"); + await expect(resolver.resolveApiKey(provider)).resolves.toEqual({ + apiKey: "refreshed-access", + accountId: "acct_2", + }); expect(withFileLockMock).toHaveBeenCalledOnce(); expect(refreshOpenAICodexTokenMock).toHaveBeenCalledWith("stored-refresh"); expect(writeStoredCodexCredentialMock).toHaveBeenCalled(); diff --git a/tests/agent/llm/providers.codex-stream.test.ts b/tests/agent/llm/providers.codex-stream.test.ts new file mode 100644 index 0000000..3d20431 --- /dev/null +++ b/tests/agent/llm/providers.codex-stream.test.ts @@ -0,0 +1,122 @@ +import { describe, expect, test } from "vitest"; +import { mapCodexResponsesEvents } from "@/src/agent/llm/providers/codex/stream.js"; + +async function* eventsFrom(events: Array>) { + yield* events; +} + +async function collectMappedEvents(events: Array>) { + const mapped = []; + for await (const event of mapCodexResponsesEvents(eventsFrom(events))) { + mapped.push(event); + } + return mapped; +} + +describe("codex responses stream adapter", () => { + test("normalizes representative Codex response events for the shared Responses processor", async () => { + const messageItem = { + type: "message", + id: "msg_test", + content: [{ type: "output_text", text: "ok" }], + }; + + const mapped = await collectMappedEvents([ + { + type: "response.output_item.added", + item: { ...messageItem, content: [] }, + }, + { + type: "response.content_part.added", + part: { type: "output_text", text: "" }, + }, + { + type: "response.output_text.delta", + delta: "ok", + }, + { + type: "response.output_item.done", + item: messageItem, + }, + { + type: "response.done", + response: { + id: "resp_test", + status: "completed", + }, + }, + ]); + + expect(mapped.map((event) => event.type)).toEqual([ + "response.output_item.added", + "response.content_part.added", + "response.output_text.delta", + "response.output_item.done", + "response.completed", + ]); + expect(mapped.at(-1)).toMatchObject({ + type: "response.completed", + response: { status: "completed" }, + }); + }); + + test("maps failed Codex done events to the shared failure path", async () => { + const mapped = await collectMappedEvents([ + { + type: "response.done", + response: { + id: "resp_failed", + status: "failed", + error: { + code: "invalid_request_error", + message: "duplicate item id", + }, + }, + }, + ]); + + expect(mapped).toHaveLength(1); + expect(mapped[0]).toMatchObject({ + type: "response.failed", + response: { + status: "failed", + error: { + code: "invalid_request_error", + message: "duplicate item id", + }, + }, + }); + }); + + test("continues consuming events after a terminal Codex event", async () => { + const mapped = await collectMappedEvents([ + { + type: "response.done", + response: { + id: "resp_test", + status: "completed", + }, + }, + { + type: "response.output_text.delta", + delta: "late metadata", + }, + ]); + + expect(mapped.map((event) => event.type)).toEqual([ + "response.completed", + "response.output_text.delta", + ]); + }); + + test("rejects malformed critical Codex events at the adapter boundary", async () => { + await expect( + collectMappedEvents([ + { + type: "response.output_text.delta", + delta: 123, + }, + ]), + ).rejects.toThrow("Invalid Codex response.output_text.delta event"); + }); +}); From 12e6010ee40827cccb991ef6002dd18d915dfb6f Mon Sep 17 00:00:00 2001 From: danielwpz <5139554+danielwpz@users.noreply.github.com> Date: Fri, 29 May 2026 12:23:16 +0800 Subject: [PATCH 3/4] Address Codex adapter review gaps --- .../llm/pi-ai-openai-responses-shared.ts | 35 ++++++ src/agent/llm/pi-bridge.ts | 12 +- src/agent/llm/providers/codex/stream.ts | 89 +++++++++------ src/agent/llm/upstream-openai.ts | 106 +++++++++--------- .../llm/pi-bridge.codex-service-tier.test.ts | 33 ++++++ .../agent/llm/providers.codex-stream.test.ts | 55 +++++++++ .../agent/llm/upstream-openai.stream.test.ts | 99 ++++++++++++++++ 7 files changed, 338 insertions(+), 91 deletions(-) diff --git a/src/agent/llm/pi-ai-openai-responses-shared.ts b/src/agent/llm/pi-ai-openai-responses-shared.ts index 773ea94..f4c39fe 100644 --- a/src/agent/llm/pi-ai-openai-responses-shared.ts +++ b/src/agent/llm/pi-ai-openai-responses-shared.ts @@ -43,6 +43,8 @@ import type { } from "openai/resources/responses/responses.js"; import { appendCappedTextTail, capTextTail } from "@/src/shared/capped-text.js"; +export type ResponsesInput = Exclude; + export interface OpenAIResponsesStreamOptions { serviceTier?: ResponseCreateParamsStreaming["service_tier"]; applyServiceTierPricing?: ( @@ -74,6 +76,39 @@ type StreamBlock = StreamThinkingBlock | StreamTextBlock | StreamToolCallBlock; const STREAM_REASONING_CONTENT_MAX_CHARS = 100_000; const STREAM_REASONING_TRUNCATION_PREFIX = "...[earlier reasoning truncated]\n"; +/** + * Normalizes developer role to system for the Responses API. + * + * This is a hard invariant: the Responses API must never emit developer-role + * input items, even if a future upstream change allows `supportsDeveloperRole` + * in a compat layer. The responses format treats developer and system as + * equivalent, and pokoclaw normalizes to system everywhere. + */ +export function normalizeResponsesInputRoles(input: ResponsesInput): ResponsesInput { + if (!Array.isArray(input)) { + return input; + } + + let changed = false; + const normalized = input.map((item) => { + if (!isRecord(item) || item.role !== "developer") { + return item; + } + + changed = true; + return { + ...item, + role: "system" as const, + }; + }); + + return changed ? normalized : input; +} + +export function isRecord(value: unknown): value is Record & { role?: unknown } { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + interface ResponsesReasoningSummaryPart { text: string; } diff --git a/src/agent/llm/pi-bridge.ts b/src/agent/llm/pi-bridge.ts index bdcaf71..ccc0b1c 100644 --- a/src/agent/llm/pi-bridge.ts +++ b/src/agent/llm/pi-bridge.ts @@ -691,7 +691,9 @@ async function buildPiStreamOptions( const serviceTier = resolveRequestServiceTier(model.serviceTier); if (serviceTier != null && shouldApplyOpenAIServiceTier(model)) { options.serviceTier = serviceTier; - options.onPayload = createServiceTierPayloadPatch(serviceTier); + if (shouldInjectServiceTierViaPayloadHook(model)) { + options.onPayload = createServiceTierPayloadPatch(serviceTier); + } } return options; @@ -707,8 +709,8 @@ function resolveRequestServiceTier( } function createServiceTierPayloadPatch(serviceTier: RequestServiceTier) { - // pi-ai calls `onPayload` with the live request payload before JSON serialization. - // Mutating in place is intentional here and covered by the final-fetch bridge tests. + // Used for pi-ai streamSimple adapters that do not read Pokoclaw's + // `options.serviceTier` extension. Custom adapters set service_tier directly. return (payload: unknown): unknown | undefined => { if (!isPlainObjectRecord(payload) || payload.service_tier !== undefined) { return undefined; @@ -718,6 +720,10 @@ function createServiceTierPayloadPatch(serviceTier: RequestServiceTier) { }; } +function shouldInjectServiceTierViaPayloadHook(model: ResolvedModel): boolean { + return resolvePiApi(model) === "openai-responses"; +} + function isPlainObjectRecord(value: unknown): value is Record { return Boolean(value && typeof value === "object" && !Array.isArray(value)); } diff --git a/src/agent/llm/providers/codex/stream.ts b/src/agent/llm/providers/codex/stream.ts index f581524..ed3f00e 100644 --- a/src/agent/llm/providers/codex/stream.ts +++ b/src/agent/llm/providers/codex/stream.ts @@ -18,10 +18,12 @@ import { buildAgentLlmRawErrorPayload } from "@/src/agent/llm/errors.js"; import { convertResponsesMessages, convertResponsesTools, + isRecord, + normalizeResponsesInputRoles, processResponsesStream, + type ResponsesInput, } from "@/src/agent/llm/pi-ai-openai-responses-shared.js"; -type ResponsesInput = Exclude; type RequestServiceTier = ResponseCreateParamsStreaming["service_tier"]; export type CodexResponsesStreamOptions = SimpleStreamOptions & { @@ -349,7 +351,9 @@ function normalizeCodexTerminalEvent( : type === "response.incomplete" ? "incomplete" : undefined; - const status = normalizeCodexResponseStatus(response.status, type) ?? fallbackStatus; + const status = + normalizeCodexResponseStatus(response.status, type, { required: type === "response.done" }) ?? + fallbackStatus; const normalizedResponse = { ...response, ...(status == null ? {} : { status }), @@ -365,8 +369,15 @@ function normalizeCodexTerminalEvent( } as unknown as ResponseStreamEvent; } -function normalizeCodexResponseStatus(value: unknown, eventType: string): string | undefined { +function normalizeCodexResponseStatus( + value: unknown, + eventType: string, + options?: { required?: boolean }, +): string | undefined { if (value == null) { + if (options?.required === true) { + throw invalidCodexEvent(eventType, "response.status is required"); + } return undefined; } if (typeof value !== "string" || !CODEX_RESPONSE_STATUSES.has(value)) { @@ -378,10 +389,13 @@ function normalizeCodexResponseStatus(value: unknown, eventType: string): string function validateCodexOutputItemEvent(event: Record, eventType: string): void { const item = requireRecordField(event, "item", eventType, "item"); const itemType = requireStringField(item, "type", eventType, "item.type"); + const isDoneEvent = eventType === "response.output_item.done"; if (itemType === "message") { requireStringField(item, "id", eventType, "item.id"); - validateMessageContentParts(item.content, eventType, "item.content"); + validateMessageContentParts(item.content, eventType, "item.content", { + required: isDoneEvent, + }); return; } @@ -399,10 +413,13 @@ function validateCodexOutputItemEvent(event: Record, eventType: if (item.id != null && typeof item.id !== "string") { throw invalidCodexEvent(eventType, "item.id must be a string"); } - if (item.summary != null && !Array.isArray(item.summary)) { - throw invalidCodexEvent(eventType, "item.summary must be an array"); - } + validateReasoningSummaryParts(item.summary, eventType, "item.summary", { + required: isDoneEvent, + }); + return; } + + requireStringField(item, "id", eventType, "item.id"); } function validateCodexContentPartEvent(event: Record, eventType: string): void { @@ -418,8 +435,16 @@ function validateReasoningSummaryPartEvent( requireStringField(part, "text", eventType, "part.text"); } -function validateMessageContentParts(value: unknown, eventType: string, path: string): void { +function validateMessageContentParts( + value: unknown, + eventType: string, + path: string, + options?: { required?: boolean }, +): void { if (value == null) { + if (options?.required === true) { + throw invalidCodexEvent(eventType, `${path} is required`); + } return; } if (!Array.isArray(value)) { @@ -442,6 +467,29 @@ function validateMessagePart(part: Record, eventType: string, p } } +function validateReasoningSummaryParts( + value: unknown, + eventType: string, + path: string, + options?: { required?: boolean }, +): void { + if (value == null) { + if (options?.required === true) { + throw invalidCodexEvent(eventType, `${path} is required`); + } + return; + } + if (!Array.isArray(value)) { + throw invalidCodexEvent(eventType, `${path} must be an array`); + } + value.forEach((part, index) => { + if (!isRecord(part)) { + throw invalidCodexEvent(eventType, `${path}[${index}] must be an object`); + } + requireStringField(part, "text", eventType, `${path}[${index}].text`); + }); +} + function readCodexErrorMessage(event: Record): string { const code = typeof event.code === "string" ? event.code : ""; const message = typeof event.message === "string" ? event.message : ""; @@ -478,31 +526,6 @@ function invalidCodexEvent(eventType: string, message: string): Error { return new Error(`Invalid Codex ${eventType} event: ${message}`); } -function isRecord(value: unknown): value is Record & { role?: unknown } { - return typeof value === "object" && value !== null && !Array.isArray(value); -} - -function normalizeResponsesInputRoles(input: ResponsesInput): ResponsesInput { - if (!Array.isArray(input)) { - return input; - } - - let changed = false; - const normalized = input.map((item) => { - if (!isRecord(item) || item.role !== "developer") { - return item; - } - - changed = true; - return { - ...item, - role: "system" as const, - }; - }); - - return changed ? normalized : input; -} - function applyCodexServiceTierPricing( model: Model<"openai-codex-responses">, usage: Usage, diff --git a/src/agent/llm/upstream-openai.ts b/src/agent/llm/upstream-openai.ts index 0ae42ac..7184d07 100644 --- a/src/agent/llm/upstream-openai.ts +++ b/src/agent/llm/upstream-openai.ts @@ -31,7 +31,9 @@ import { buildAgentLlmRawErrorPayload } from "@/src/agent/llm/errors.js"; import { convertResponsesMessages, convertResponsesTools, + normalizeResponsesInputRoles, processResponsesStream, + type ResponsesInput, } from "@/src/agent/llm/pi-ai-openai-responses-shared.js"; import { type CodexResponsesStreamOptions, @@ -86,7 +88,6 @@ interface ParsedActualCost { total?: number; } -type ResponsesInput = Exclude; type AssistantContentItem = AssistantMessage["content"][number]; type AssistantToolCallContent = Extract; type AssistantMessageWithRawError = AssistantMessage & { @@ -99,6 +100,14 @@ type ResolvedOpenAICompletionsCompat = Omit< cacheControlFormat?: OpenAICompletionsCompat["cacheControlFormat"]; }; +type OpenAIResponsesAdapterOptions = SimpleStreamOptions & { + serviceTier?: ResponseCreateParamsStreaming["service_tier"]; + onPayload?: ( + payload: unknown, + model: Model, + ) => unknown | undefined | Promise; +}; + interface UpstreamCostParser { supports(model: Pick, "baseUrl">): boolean; parse(rawUsage: OpenAICompatibleUsageRaw): ParsedActualCost | null; @@ -236,7 +245,7 @@ export function streamWithNormalizedUpstreamUsage( return streamOpenAIResponsesWithUpstreamUsage( model as Model<"openai-responses">, context, - options, + options as OpenAIResponsesAdapterOptions | undefined, ); } @@ -525,7 +534,7 @@ function appendStreamReasoning(existing: string, delta: string): string { function streamOpenAIResponsesWithUpstreamUsage( model: Model<"openai-responses">, context: Context, - options?: SimpleStreamOptions, + options?: OpenAIResponsesAdapterOptions, ): AssistantMessageEventStream { const stream = createAssistantMessageEventStream(); @@ -565,12 +574,30 @@ function streamOpenAIResponsesWithUpstreamUsage( let completedUsage: unknown = null; let completedServiceTier: ResponseCreateParamsStreaming["service_tier"] | undefined; + const requestServiceTier = options?.serviceTier; async function* tappedStream(): AsyncIterable { const events = rawStream as AsyncIterable; for await (const event of events) { if (event.type === "response.completed") { completedUsage = event.response?.usage ?? null; - completedServiceTier = event.response?.service_tier; + completedServiceTier = resolveResponseServiceTier( + event.response?.service_tier, + requestServiceTier, + ); + if ( + event.response != null && + completedServiceTier !== undefined && + completedServiceTier !== event.response.service_tier + ) { + yield { + ...event, + response: { + ...event.response, + service_tier: completedServiceTier, + }, + }; + continue; + } } yield event; } @@ -578,7 +605,7 @@ function streamOpenAIResponsesWithUpstreamUsage( stream.push({ type: "start", partial: output }); await processResponsesStream(tappedStream(), output, stream, model, { - serviceTier: completedServiceTier, + serviceTier: requestServiceTier, applyServiceTierPricing, }); @@ -587,7 +614,7 @@ function streamOpenAIResponsesWithUpstreamUsage( if (normalizedUsage) { output.usage = normalizedUsage.usage; if (normalizedUsage.costSource === "estimated") { - applyServiceTierPricing(output.usage, completedServiceTier); + applyServiceTierPricing(output.usage, completedServiceTier ?? requestServiceTier); } } } @@ -654,7 +681,7 @@ export function buildOpenAICompletionsParams( export function buildOpenAIResponsesParams( model: Model<"openai-responses">, context: Context, - options?: SimpleStreamOptions, + options?: OpenAIResponsesAdapterOptions, ): ResponseCreateParamsStreaming { const messages: ResponsesInput = normalizeResponsesInputRoles( convertResponsesMessages(model, context, OPENAI_TOOL_CALL_PROVIDERS), @@ -671,9 +698,8 @@ export function buildOpenAIResponsesParams( params.prompt_cache_key = options.sessionId; } - const requestServiceTier = getRequestServiceTier(options); - if (requestServiceTier) { - params.service_tier = requestServiceTier; + if (options?.serviceTier) { + params.service_tier = options.serviceTier; } if (context.tools) { @@ -691,25 +717,28 @@ export function buildOpenAIResponsesParams( return params; } -function getRequestServiceTier( - options?: SimpleStreamOptions, -): ResponseCreateParamsStreaming["service_tier"] | undefined { - return (options as { serviceTier?: ResponseCreateParamsStreaming["service_tier"] } | undefined) - ?.serviceTier; -} - async function applyPayloadHook( params: ResponseCreateParamsStreaming, model: Model<"openai-responses">, - options?: SimpleStreamOptions, + options?: OpenAIResponsesAdapterOptions, ): Promise { - const payloadHook = options?.onPayload as - | ((payload: unknown, model: Model) => unknown | undefined | Promise) - | undefined; - const nextParams = await payloadHook?.(params, model); + const nextParams = await options?.onPayload?.(params, model); return nextParams ?? params; } +function resolveResponseServiceTier( + responseServiceTier: ResponseCreateParamsStreaming["service_tier"] | undefined, + requestServiceTier: ResponseCreateParamsStreaming["service_tier"] | undefined, +): ResponseCreateParamsStreaming["service_tier"] | undefined { + if ( + requestServiceTier != null && + (responseServiceTier == null || responseServiceTier === "default") + ) { + return requestServiceTier; + } + return responseServiceTier; +} + /** * Resolves the effective compat for completions-API calls. * @@ -727,35 +756,6 @@ function resolveCompat(model: Model<"openai-completions">): ResolvedOpenAIComple }; } -/** - * Normalizes developer role to system for the Responses API. - * - * This is a hard invariant: the Responses API must never emit developer-role - * input items, even if a future upstream change allows `supportsDeveloperRole` - * in a compat layer. The responses format treats developer and system as - * equivalent, and pokoclaw normalizes to system everywhere. - */ -function normalizeResponsesInputRoles(input: ResponsesInput): ResponsesInput { - if (!Array.isArray(input)) { - return input; - } - - let changed = false; - const normalized = input.map((item) => { - if (!isRecord(item) || item.role !== "developer") { - return item; - } - - changed = true; - return { - ...item, - role: "system" as const, - }; - }); - - return changed ? normalized : input; -} - function convertCompletionTools( tools: Tool[] | undefined, ): OpenAI.Chat.ChatCompletionTool[] | undefined { @@ -771,10 +771,6 @@ function convertCompletionTools( })); } -function isRecord(value: unknown): value is Record & { role?: unknown } { - return typeof value === "object" && value !== null; -} - function parseRawUsage(rawUsage: unknown): OpenAICompatibleUsageRaw | null { if (typeof rawUsage !== "object" || rawUsage == null) { return null; diff --git a/tests/agent/llm/pi-bridge.codex-service-tier.test.ts b/tests/agent/llm/pi-bridge.codex-service-tier.test.ts index f7ab1dd..b4057de 100644 --- a/tests/agent/llm/pi-bridge.codex-service-tier.test.ts +++ b/tests/agent/llm/pi-bridge.codex-service-tier.test.ts @@ -375,6 +375,39 @@ describe("pi bridge Codex service tier", () => { ).rejects.toThrow("invalid_request_error: Codex rejected duplicate item ids"); }); + test("preserves Codex failed done event details for non-streaming bridge turns", async () => { + vi.stubGlobal( + "fetch", + vi.fn(async () => + createSseResponseFromEvents([ + { + type: "response.done", + response: { + id: "resp_failed", + status: "failed", + error: { + code: "invalid_request_error", + message: "Codex rejected duplicate item ids", + }, + }, + }, + ]), + ), + ); + + const bridge = new PiBridge(); + await expect( + bridge.completeTurn({ + model: createCodexModel(), + systemPrompt: "You are concise.", + compactSummary: null, + messages: [createStoredUserMessage()], + tools: new ToolRegistry(), + signal: new AbortController().signal, + }), + ).rejects.toThrow("invalid_request_error: Codex rejected duplicate item ids"); + }); + test("rejects malformed Codex stream events before response processing", async () => { vi.stubGlobal( "fetch", diff --git a/tests/agent/llm/providers.codex-stream.test.ts b/tests/agent/llm/providers.codex-stream.test.ts index 3d20431..6a56f68 100644 --- a/tests/agent/llm/providers.codex-stream.test.ts +++ b/tests/agent/llm/providers.codex-stream.test.ts @@ -119,4 +119,59 @@ describe("codex responses stream adapter", () => { ]), ).rejects.toThrow("Invalid Codex response.output_text.delta event"); }); + + test("requires a known status on Codex done events", async () => { + await expect( + collectMappedEvents([ + { + type: "response.done", + response: { + id: "resp_missing_status", + }, + }, + ]), + ).rejects.toThrow("Invalid Codex response.done event: response.status is required"); + }); + + test("requires finalized message content on Codex output item done events", async () => { + await expect( + collectMappedEvents([ + { + type: "response.output_item.done", + item: { + type: "message", + id: "msg_missing_content", + }, + }, + ]), + ).rejects.toThrow("Invalid Codex response.output_item.done event: item.content is required"); + }); + + test("requires finalized reasoning summary on Codex output item done events", async () => { + await expect( + collectMappedEvents([ + { + type: "response.output_item.done", + item: { + type: "reasoning", + id: "rs_missing_summary", + }, + }, + ]), + ).rejects.toThrow("Invalid Codex response.output_item.done event: item.summary is required"); + }); + + test("validates shared fields on unknown Codex output item types", async () => { + await expect( + collectMappedEvents([ + { + type: "response.output_item.added", + item: { + type: "custom_item", + id: 123, + }, + }, + ]), + ).rejects.toThrow("Invalid Codex response.output_item.added event: item.id must be a string"); + }); }); diff --git a/tests/agent/llm/upstream-openai.stream.test.ts b/tests/agent/llm/upstream-openai.stream.test.ts index 355495e..9f09c72 100644 --- a/tests/agent/llm/upstream-openai.stream.test.ts +++ b/tests/agent/llm/upstream-openai.stream.test.ts @@ -672,4 +672,103 @@ describe("upstream openai responses streaming", () => { rawMessage: "provider_error: upstream exploded", }); }); + + test("uses requested service tier for estimated Responses pricing when upstream reports default tier", async () => { + responsesCreateMock.mockResolvedValue( + createAsyncIterable([ + { + type: "response.output_item.added", + item: { + type: "message", + id: "msg_123", + role: "assistant", + content: [], + }, + }, + { + type: "response.content_part.added", + part: { + type: "output_text", + text: "", + annotations: [], + }, + }, + { + type: "response.output_text.delta", + delta: "OK", + }, + { + type: "response.output_item.done", + item: { + type: "message", + id: "msg_123", + role: "assistant", + status: "completed", + content: [ + { + type: "output_text", + text: "OK", + annotations: [], + }, + ], + }, + }, + { + type: "response.completed", + response: { + status: "completed", + service_tier: "default", + usage: { + input_tokens: 100, + output_tokens: 50, + total_tokens: 150, + input_tokens_details: { + cached_tokens: 0, + }, + }, + }, + }, + ]), + ); + + const options = { + apiKey: "secret", + serviceTier: "priority" as const, + }; + + const stream = streamWithNormalizedUpstreamUsage( + { + api: "openai-responses", + id: "openai/gpt-5.4", + name: "gpt-5.4", + provider: "openrouter", + baseUrl: "https://openrouter.ai/api/v1", + reasoning: true, + input: ["text"], + contextWindow: 200_000, + maxTokens: 16_384, + cost: { + input: 1, + output: 1, + cacheRead: 0, + cacheWrite: 0, + }, + }, + { + systemPrompt: "You are a helpful assistant.", + messages: [], + }, + options, + ); + + for await (const _event of stream) { + // Fully drain the stream before reading the final result. + } + + const result = await stream.result(); + expect(result.stopReason).toBe("stop"); + expect(result.usage.cost.input).toBeCloseTo(0.0002, 12); + expect(result.usage.cost.output).toBeCloseTo(0.0001, 12); + expect(result.usage.cost.total).toBeCloseTo(0.0003, 12); + }); }); From 00b87fc16d6124b9590d14ab530cfbcba4f95cd9 Mon Sep 17 00:00:00 2001 From: danielwpz <5139554+danielwpz@users.noreply.github.com> Date: Fri, 29 May 2026 12:48:36 +0800 Subject: [PATCH 4/4] Clarify Codex adapter review notes --- src/agent/llm/pi-bridge.ts | 2 ++ src/agent/llm/upstream-openai.ts | 2 ++ 2 files changed, 4 insertions(+) diff --git a/src/agent/llm/pi-bridge.ts b/src/agent/llm/pi-bridge.ts index ccc0b1c..eb3e680 100644 --- a/src/agent/llm/pi-bridge.ts +++ b/src/agent/llm/pi-bridge.ts @@ -385,6 +385,8 @@ async function consumeStreamToCompletion( context: Context, options: SimpleStreamOptions, ): Promise { + // Reuses the Codex streaming adapter for converter parity only. Non-streaming + // calls still rely on the caller-provided AbortSignal, not LlmStreamWatchdog. const stream = streamWithNormalizedUpstreamUsage(model, context, options); for await (const _event of stream) { // Drain the stream so the adapter can assemble the final AssistantMessage. diff --git a/src/agent/llm/upstream-openai.ts b/src/agent/llm/upstream-openai.ts index 7184d07..302cad8 100644 --- a/src/agent/llm/upstream-openai.ts +++ b/src/agent/llm/upstream-openai.ts @@ -610,6 +610,8 @@ function streamOpenAIResponsesWithUpstreamUsage( }); if (completedUsage != null) { + // Prefer the upstream-compatible usage payload when present; this + // intentionally replaces usage calculated by the shared stream parser. const normalizedUsage = normalizeUsageFromOpenAICompatible(model, completedUsage); if (normalizedUsage) { output.usage = normalizedUsage.usage;