import OpenAI from "openai"; import type { ResponseCreateParamsStreaming } from "openai/resources/responses/responses.js"; import { clampThinkingLevel } from "../models.ts"; import type { Api, AssistantMessage, CacheRetention, Context, Model, OpenAIResponsesCompat, SimpleStreamOptions, StreamFunction, StreamOptions, Usage, } from "../types.ts"; import { AssistantMessageEventStream } from "../utils/event-stream.ts"; import { headersToRecord } from "../utils/headers.ts"; import { isCloudflareProvider, resolveCloudflareBaseUrl } from "./cloudflare.ts"; import { buildCopilotDynamicHeaders, hasCopilotVisionInput } from "./github-copilot-headers.ts"; import { clampOpenAIPromptCacheKey } from "./openai-prompt-cache.ts"; import { convertResponsesMessages, convertResponsesTools, processResponsesStream } from "./openai-responses-shared.ts"; import { buildBaseOptions } from "./simple-options.ts"; const OPENAI_TOOL_CALL_PROVIDERS = new Set(["openai", "openai-codex", "opencode"]); /** * Resolve cache retention preference. * Defaults to "short" and uses PI_CACHE_RETENTION for backward compatibility. */ function resolveCacheRetention(cacheRetention?: CacheRetention): CacheRetention { if (cacheRetention) { return cacheRetention; } if (typeof process !== "undefined" && process.env.PI_CACHE_RETENTION === "long") { return "long"; } return "short"; } function getCompat(model: Model<"openai-responses">): Required { return { supportsDeveloperRole: model.compat?.supportsDeveloperRole ?? true, sendSessionIdHeader: model.compat?.sendSessionIdHeader ?? true, supportsLongCacheRetention: model.compat?.supportsLongCacheRetention ?? true, }; } function getPromptCacheRetention( compat: Required, cacheRetention: CacheRetention, ): "24h" | undefined { return cacheRetention === "long" && compat.supportsLongCacheRetention ? "24h" : undefined; } function formatOpenAIResponsesError(error: unknown): string { if (error instanceof Error) { const status = (error as Error & { status?: unknown }).status; const statusCode = typeof status === "number" ? status : undefined; if (statusCode !== undefined) { return `OpenAI API error (${statusCode}): ${error.message}`; } return error.message; } try { return JSON.stringify(error); } catch { return String(error); } } // OpenAI Responses-specific options export interface OpenAIResponsesOptions extends StreamOptions { reasoningEffort?: "minimal" | "low" | "medium" | "high" | "xhigh"; reasoningSummary?: "auto" | "detailed" | "concise" | null; serviceTier?: ResponseCreateParamsStreaming["service_tier"]; } /** * Generate function for OpenAI Responses API */ export const streamOpenAIResponses: StreamFunction<"openai-responses", OpenAIResponsesOptions> = ( model: Model<"openai-responses">, context: Context, options?: OpenAIResponsesOptions, ): AssistantMessageEventStream => { const stream = new AssistantMessageEventStream(); // Start async processing (async () => { const output: AssistantMessage = { role: "assistant", content: [], api: model.api as Api, provider: model.provider, model: model.id, usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, stopReason: "stop", timestamp: Date.now(), }; try { // Create OpenAI client const apiKey = options?.apiKey; if (!apiKey) { throw new Error(`No API key for provider: ${model.provider}`); } const cacheRetention = resolveCacheRetention(options?.cacheRetention); const cacheSessionId = cacheRetention === "none" ? undefined : options?.sessionId; const client = createClient(model, context, apiKey, options?.headers, cacheSessionId); let params = buildParams(model, context, options); const nextParams = await options?.onPayload?.(params, model); if (nextParams !== undefined) { params = nextParams as ResponseCreateParamsStreaming; } const requestOptions = { ...(options?.signal ? { signal: options.signal } : {}), ...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}), maxRetries: options?.maxRetries ?? 0, }; const { data: openaiStream, response } = await client.responses.create(params, requestOptions).withResponse(); await options?.onResponse?.({ status: response.status, headers: headersToRecord(response.headers) }, model); stream.push({ type: "start", partial: output }); await processResponsesStream(openaiStream, output, stream, model, { serviceTier: options?.serviceTier, applyServiceTierPricing: (usage, serviceTier) => applyServiceTierPricing(usage, serviceTier, model), }); 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) { for (const block of output.content) { delete (block as { index?: number }).index; // partialJson is only a streaming scratch buffer; never persist it. delete (block as { partialJson?: string }).partialJson; } output.stopReason = options?.signal?.aborted ? "aborted" : "error"; output.errorMessage = formatOpenAIResponsesError(error); stream.push({ type: "error", reason: output.stopReason, error: output }); stream.end(); } })(); return stream; }; export const streamSimpleOpenAIResponses: StreamFunction<"openai-responses", SimpleStreamOptions> = ( model: Model<"openai-responses">, context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream => { const apiKey = options?.apiKey; if (!apiKey) { throw new Error(`No API key for provider: ${model.provider}`); } const base = buildBaseOptions(model, options, apiKey); const clampedReasoning = options?.reasoning ? clampThinkingLevel(model, options.reasoning) : undefined; const reasoningEffort = clampedReasoning === "off" ? undefined : clampedReasoning; return streamOpenAIResponses(model, context, { ...base, reasoningEffort, } satisfies OpenAIResponsesOptions); }; function createClient( model: Model<"openai-responses">, context: Context, apiKey: string, optionsHeaders?: Record, sessionId?: string, ) { const compat = getCompat(model); const headers = { ...model.headers }; if (model.provider === "github-copilot") { const hasImages = hasCopilotVisionInput(context.messages); const copilotHeaders = buildCopilotDynamicHeaders({ messages: context.messages, hasImages, }); Object.assign(headers, copilotHeaders); } if (sessionId) { if (compat.sendSessionIdHeader) { headers.session_id = sessionId; } headers["x-client-request-id"] = sessionId; } // Merge options headers last so they can override defaults if (optionsHeaders) { Object.assign(headers, optionsHeaders); } const defaultHeaders = model.provider === "cloudflare-ai-gateway" ? { ...headers, Authorization: headers.Authorization ?? null, "cf-aig-authorization": `Bearer ${apiKey}`, } : headers; return new OpenAI({ apiKey, baseURL: isCloudflareProvider(model.provider) ? resolveCloudflareBaseUrl(model) : model.baseUrl, dangerouslyAllowBrowser: true, defaultHeaders, }); } function buildParams(model: Model<"openai-responses">, context: Context, options?: OpenAIResponsesOptions) { const messages = convertResponsesMessages(model, context, OPENAI_TOOL_CALL_PROVIDERS); const cacheRetention = resolveCacheRetention(options?.cacheRetention); const compat = getCompat(model); const params: ResponseCreateParamsStreaming = { model: model.id, input: messages, stream: true, prompt_cache_key: cacheRetention === "none" ? undefined : clampOpenAIPromptCacheKey(options?.sessionId), prompt_cache_retention: getPromptCacheRetention(compat, cacheRetention), store: false, }; if (options?.maxTokens) { params.max_output_tokens = options?.maxTokens; } if (options?.temperature !== undefined) { params.temperature = options?.temperature; } if (options?.serviceTier !== undefined) { params.service_tier = options.serviceTier; } if (context.tools && context.tools.length > 0) { params.tools = convertResponsesTools(context.tools); } if (model.reasoning) { if (options?.reasoningEffort || options?.reasoningSummary) { const effort = options?.reasoningEffort ? (model.thinkingLevelMap?.[options.reasoningEffort] ?? options.reasoningEffort) : "medium"; params.reasoning = { effort: effort as NonNullable["effort"], summary: options?.reasoningSummary || "auto", }; params.include = ["reasoning.encrypted_content"]; } else if (model.provider !== "github-copilot" && model.thinkingLevelMap?.off !== null) { params.reasoning = { effort: (model.thinkingLevelMap?.off ?? "none") as NonNullable["effort"], }; } } return params; } function getServiceTierCostMultiplier( model: Pick, "id">, serviceTier: ResponseCreateParamsStreaming["service_tier"] | undefined, ): number { switch (serviceTier) { case "flex": return 0.5; case "priority": return model.id === "gpt-5.5" ? 2.5 : 2; default: return 1; } } function applyServiceTierPricing( usage: Usage, serviceTier: ResponseCreateParamsStreaming["service_tier"] | undefined, model: Pick, "id">, ) { const multiplier = getServiceTierCostMultiplier(model, 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; }