Repository Public
1/**2 * OpenAI adapter — Responses API through the official SDK, stateless.3 *4 * Request shape (every call), verified against `openai` 7.54 typings and developers.openai.com/api/docs5 * (guides: migrate-to-responses, function-calling, reasoning, prompt-caching; read 2026-11-10):6 * client.responses.create({ model, instructions, input, tools:[{type:'function', name, description, parameters, strict:true}],7 * parallel_tool_calls:true, store:false, stream:true, include:['reasoning.encrypted_content'],8 * reasoning:{effort, summary:'auto'}, max_output_tokens })9 * Never sent: `previous_response_id`, `conversation`, `tool_choice`, sampling parameters.10 *11 * Native transcript entries are Responses input items, append-only: user/developer `message` items, the12 * model's output items verbatim (reasoning items with `encrypted_content`, `message`, `function_call`) and13 * `function_call_output` items. Prompt caching is implicit (stable prefix: tools sorted by name, fixed14 * instructions, append-only input).15 */16import OpenAI from 'openai';17import type { ContentBlock, ModelInfo, StopReason, ToolResult, Usage } from '@aos/shared';18import { z } from 'zod';19import { CATALOGUE, catalogueEntry, priceRow, priceUsage, type TokenUsage } from './catalogue.js';20import { ProviderError, parseRetryAfter } from './errors.js';21import { DEFAULT_MAX_ATTEMPTS, abortableSleep, withRetries } from './retry.js';22import { markInvalidToolInput, stripNullOptionals, toStrictSchema, validateToolInput } from './schema.js';23import type { AdapterOptions, ModelRequest, ProviderAdapter, ProviderResponse, ToolDef } from './types.js';2425const PROVIDER = 'openai' as const;26export const OPENAI_DEFAULT_MAX_OUTPUT_TOKENS = 64_000;27export const TOOL_ERROR_PREFIX = '[tool_error] ';2829type Item = Record<string, unknown>;30const ItemSchema = z.looseObject({}).refine((v) => typeof v.type === 'string' || typeof v.role === 'string');3132/** Accepts `[...t, ...native]` and `[...t, native]`: nested arrays are flattened one level, then validated. */33export function openaiItems(transcript: unknown[]): Item[] {34 const out: Item[] = [];35 for (const entry of transcript) {36 for (const item of Array.isArray(entry) ? entry : [entry]) {37 const parsed = ItemSchema.safeParse(item);38 if (!parsed.success) {39 throw new ProviderError('Invalid OpenAI transcript entry', { provider: PROVIDER, retryable: false, code: 'invalid_transcript' });40 }41 out.push(parsed.data);42 }43 }44 return out;45}4647export interface OpenAIRequest {48 params: Record<string, unknown>;49 strictSchemas: Map<string, Record<string, unknown>>;50 originals: Map<string, Record<string, unknown>>;51}5253function openaiTools(tools: ToolDef[]): Pick<OpenAIRequest, 'strictSchemas' | 'originals'> & { tools: Item[] } {54 const strictSchemas = new Map<string, Record<string, unknown>>();55 const originals = new Map<string, Record<string, unknown>>();56 // Deterministic order keeps the cached prefix stable.57 const sorted = [...tools].sort((a, b) => (a.name < b.name ? -1 : a.name > b.name ? 1 : 0));58 const out = sorted.map((t) => {59 const strict = toStrictSchema(t.inputSchema);60 strictSchemas.set(t.name, strict);61 originals.set(t.name, t.inputSchema);62 return { type: 'function', name: t.name, description: t.description, parameters: strict, strict: true };63 });64 return { tools: out, strictSchemas, originals };65}6667/** Pure request builder (exported for tests and for auditing the exact payload). */68export function buildOpenAIRequest(req: ModelRequest, items: Item[] = openaiItems(req.transcript)): OpenAIRequest {69 const info = catalogueEntry(PROVIDER, req.model);70 const { tools, strictSchemas, originals } = openaiTools(req.tools);71 const cap = info?.maxOutputTokens ?? Number.POSITIVE_INFINITY;72 const params: Record<string, unknown> = {73 model: req.model,74 input: items,75 store: false,76 stream: true,77 // Encrypted reasoning items are returned and replayed unchanged (stateless multi-turn reasoning).78 include: ['reasoning.encrypted_content'],79 // `effort` maps 1:1 from our Effort (low…max); omitted ⇒ the model's documented default.80 reasoning: { ...(req.effort ? { effort: req.effort } : {}), summary: 'auto' },81 max_output_tokens: Math.min(req.maxOutputTokens ?? OPENAI_DEFAULT_MAX_OUTPUT_TOKENS, cap),82 };83 if (req.system.length > 0) params.instructions = req.system;84 if (tools.length > 0) {85 params.tools = tools;86 params.parallel_tool_calls = true;87 }88 return { params, strictSchemas, originals };89}9091function num(v: unknown): number {92 return typeof v === 'number' && Number.isFinite(v) ? v : 0;93}9495/** `input_tokens` INCLUDES cached and cache-write tokens on OpenAI; we normalise to the uncached remainder. */96export function openaiUsage(usage: unknown): TokenUsage {97 const u = (usage ?? {}) as Record<string, unknown>;98 const details = (u.input_tokens_details ?? {}) as Record<string, unknown>;99 const cacheReadTokens = num(details.cached_tokens);100 const cacheWriteTokens = num(details.cache_write_tokens);101 return {102 inputTokens: Math.max(0, num(u.input_tokens) - cacheReadTokens - cacheWriteTokens),103 outputTokens: num(u.output_tokens), // reasoning tokens are part of output_tokens104 cacheReadTokens, cacheWriteTokens,105 };106}107108/** `response.failed` codes that are a policy decision, not an outage. */109const POLICY_FAILURE_CODES = new Set(['invalid_prompt', 'bio_policy', 'misalignment_policy_violation', 'image_content_policy_violation']);110const RETRYABLE_FAILURE_CODES = new Set(['server_error', 'rate_limit_exceeded', 'vector_store_timeout']);111112interface Normalised {113 blocks: ContentBlock[];114 native: Item[];115 stop: StopReason;116 nativeStop: string | null;117 refusal?: { category: string | null; explanation: string | null };118}119120/** Normalises a final Response and its completed output items. Exported for fixture tests. */121export function normaliseOpenAIResponse(122 response: Item,123 outputItems: Item[],124 strictSchemas: Map<string, Record<string, unknown>>,125 originals: Map<string, Record<string, unknown>>,126): Normalised {127 const status = typeof response.status === 'string' ? response.status : null;128 const incomplete = (response.incomplete_details ?? null) as Record<string, unknown> | null;129 const incompleteReason = typeof incomplete?.reason === 'string' ? incomplete.reason : null;130 const nativeStop = status === null ? null : incompleteReason ? `${status}:${incompleteReason}` : status;131132 if (status === 'failed') {133 const error = (response.error ?? {}) as Record<string, unknown>;134 const code = typeof error.code === 'string' ? error.code : null;135 const message = typeof error.message === 'string' ? error.message : 'The response failed';136 if (code !== null && POLICY_FAILURE_CODES.has(code)) {137 return { blocks: [], native: [], stop: 'refusal', nativeStop: `failed:${code}`, refusal: { category: code, explanation: message } };138 }139 throw new ProviderError(message, { provider: PROVIDER, retryable: code === null || RETRYABLE_FAILURE_CODES.has(code), code: code ?? 'response_failed' });140 }141142 const truncated = status === 'incomplete' && incompleteReason === 'max_output_tokens';143 const blocks: ContentBlock[] = [];144 const refusalTexts: string[] = [];145 const lastIndex = outputItems.length - 1;146 outputItems.forEach((item, i) => {147 if (item.type === 'reasoning') {148 const parts = Array.isArray(item.summary) ? (item.summary as Item[]) : [];149 blocks.push({ type: 'reasoning', summary: parts.map((p) => (typeof p.text === 'string' ? p.text : '')).filter(Boolean).join('\n\n') });150 } else if (item.type === 'message') {151 for (const part of Array.isArray(item.content) ? (item.content as Item[]) : []) {152 if (part.type === 'output_text' && typeof part.text === 'string' && part.text.length > 0) blocks.push({ type: 'text', text: part.text });153 else if (part.type === 'refusal') refusalTexts.push(typeof part.refusal === 'string' ? part.refusal : '');154 }155 } else if (item.type === 'function_call' && typeof item.call_id === 'string' && typeof item.name === 'string') {156 const raw = typeof item.arguments === 'string' ? item.arguments : '';157 let input: unknown;158 if (truncated && (item.status === 'incomplete' || i === lastIndex)) {159 input = markInvalidToolInput({ reason: 'truncated', detail: 'output-token limit reached inside the tool arguments', raw });160 } else {161 try {162 const parsed: unknown = JSON.parse(raw);163 const strict = strictSchemas.get(item.name);164 const problem = strict ? validateToolInput(strict, parsed) : null;165 input = problem166 ? markInvalidToolInput({ reason: 'schema', detail: problem, raw })167 : stripNullOptionals(originals.get(item.name), parsed);168 } catch {169 input = markInvalidToolInput({ reason: 'invalid_json', detail: 'arguments are not valid JSON', raw });170 }171 }172 blocks.push({ type: 'tool_call', id: item.call_id, name: item.name, input });173 }174 });175176 const contentFiltered = status === 'incomplete' && incompleteReason === 'content_filter';177 if (refusalTexts.length > 0 || contentFiltered) {178 // Never run the tool calls of a refused turn; keep only the assistant message items in the transcript179 // (a reasoning item without its following item, or an unanswered function_call, would be rejected on replay).180 return {181 blocks: blocks.filter((b) => b.type !== 'tool_call'),182 native: outputItems.filter((it) => it.type === 'message'),183 stop: 'refusal', nativeStop,184 refusal: { category: contentFiltered ? 'content_filter' : null, explanation: refusalTexts.filter(Boolean).join('\n') || null },185 };186 }187188 // A trailing reasoning item (output cut off while reasoning) has no following item and cannot be replayed.189 const native = [...outputItems];190 while (native.length > 0 && native[native.length - 1]?.type === 'reasoning') native.pop();191192 let stop: StopReason;193 // Every function_call in the transcript needs a function_call_output, whatever ended the turn.194 if (blocks.some((b) => b.type === 'tool_call')) stop = 'tool_use';195 else if (status === 'completed') stop = 'end_turn';196 else if (truncated) stop = 'max_tokens';197 else stop = 'error';198 return { blocks, native, stop, nativeStop };199}200201const RETRYABLE_STREAM_CODES = new Set(['server_error', 'rate_limit_exceeded', 'overloaded', 'service_unavailable', 'timeout', 'internal_error']);202203/** Typed error chain, most specific first. Anything that is not an SDK error is returned as null. */204export function mapOpenAIError(err: unknown): ProviderError | null {205 if (err instanceof ProviderError) return err;206 if (err instanceof OpenAI.APIUserAbortError) {207 return new ProviderError('Request aborted', { provider: PROVIDER, retryable: false, code: 'aborted', cause: err });208 }209 if (err instanceof OpenAI.APIConnectionTimeoutError) {210 return new ProviderError('OpenAI request timed out', { provider: PROVIDER, retryable: true, code: 'timeout', cause: err });211 }212 if (err instanceof OpenAI.APIConnectionError) {213 return new ProviderError('Could not reach the OpenAI API', { provider: PROVIDER, retryable: true, code: 'connection_error', cause: err });214 }215 if (err instanceof OpenAI.APIError) {216 const status = typeof err.status === 'number' ? err.status : null;217 const code = err.code ?? err.type ?? null;218 let retryable: boolean;219 if (err instanceof OpenAI.RateLimitError) retryable = code !== 'insufficient_quota'; // an exhausted quota never recovers by waiting220 else if (status === null) retryable = code === null || RETRYABLE_STREAM_CODES.has(code); // error event inside an open stream221 else retryable = status === 408 || status === 409 || status >= 500;222 return new ProviderError(err.message, {223 provider: PROVIDER, retryable, status, code: code ?? (status === null ? 'api_error' : `http_${status}`),224 requestId: err.requestID ?? null, retryAfterMs: parseRetryAfter(err.headers), cause: err,225 });226 }227 return null;228}229230export function createOpenAIAdapter(options: AdapterOptions = {}): ProviderAdapter {231 let client: OpenAI | null = null;232 const getClient = (): OpenAI => {233 if (client) return client;234 const apiKey = options.apiKey ?? process.env.OPENAI_API_KEY;235 if (!apiKey) {236 throw new ProviderError('OPENAI_API_KEY is not set', { provider: PROVIDER, retryable: false, code: 'missing_api_key' });237 }238 client = new OpenAI({239 apiKey,240 maxRetries: 0, // retries are handled here so that attempts, backoff and retry-after are observable241 timeout: options.timeoutMs ?? 600_000,242 ...(options.baseURL ? { baseURL: options.baseURL } : {}),243 ...(options.fetch ? { fetch: options.fetch } : {}),244 });245 return client;246 };247 const retry = {248 maxAttempts: options.maxAttempts ?? DEFAULT_MAX_ATTEMPTS,249 sleep: options.sleep ?? abortableSleep,250 random: options.random ?? Math.random,251 };252 const aborted = (cause: unknown, attempts: number): ProviderError =>253 new ProviderError('Request aborted', { provider: PROVIDER, retryable: false, code: 'aborted', attempts, cause });254255 /** One streamed request → final Response + completed output items. */256 async function streamOnce(built: OpenAIRequest, req: ModelRequest): Promise<{ response: Item; items: Item[] }> {257 const emit = (type: 'text' | 'reasoning', text: string): void => {258 if (!req.onDelta || text.length === 0) return;259 try { req.onDelta({ type, text }); } catch { /* a console listener must never break the model call */ }260 };261 // Plain-JSON params (the native transcript is opaque JSON by contract), cast once at the SDK boundary.262 const stream = await getClient().responses.create(263 built.params as unknown as OpenAI.Responses.ResponseCreateParamsStreaming,264 req.signal ? { signal: req.signal } : {},265 );266 const done = new Map<number, Item>();267 let final: Item | null = null;268 let sawText = false;269 let lastTextItem: string | null = null;270 for await (const event of stream) {271 switch (event.type) {272 case 'response.output_text.delta':273 if (sawText && lastTextItem !== null && lastTextItem !== event.item_id) emit('text', '\n\n');274 emit('text', event.delta);275 sawText = true;276 lastTextItem = event.item_id;277 break;278 case 'response.reasoning_summary_part.added':279 if (event.summary_index > 0) emit('reasoning', '\n\n');280 break;281 case 'response.reasoning_summary_text.delta':282 emit('reasoning', event.delta);283 break;284 case 'response.output_item.done':285 // The completed item is the one to replay (its encrypted_content is complete only here).286 done.set(event.output_index, event.item as unknown as Item);287 break;288 case 'response.completed':289 case 'response.incomplete':290 case 'response.failed':291 final = event.response as unknown as Item;292 break;293 case 'error':294 throw new ProviderError(event.message, {295 provider: PROVIDER, code: event.code ?? 'stream_error',296 retryable: event.code === null || RETRYABLE_STREAM_CODES.has(event.code),297 });298 default:299 break;300 }301 }302 if (!final) {303 throw new ProviderError('The response stream ended before a terminal event', { provider: PROVIDER, retryable: true, code: 'stream_truncated' });304 }305 const items = done.size > 0306 ? [...done.entries()].sort((a, b) => a[0] - b[0]).map(([, item]) => item)307 : Array.isArray(final.output) ? (final.output as Item[]) : [];308 return { response: final, items };309 }310311 const adapter: ProviderAdapter = {312 id: PROVIDER,313314 initTranscript(userText) {315 return [{ type: 'message', role: 'user', content: [{ type: 'input_text', text: userText }] }];316 },317 appendAssistant(transcript, native) {318 return [...transcript, ...(Array.isArray(native) ? native : [native])];319 },320 appendToolResults(transcript, results: ToolResult[]) {321 // One function_call_output per call, appended together. The Responses API has no error flag on tool322 // output, so a failure is marked in the text itself.323 return [...transcript, ...results.map((r) => ({324 type: 'function_call_output', call_id: r.callId, output: r.ok ? r.content : `${TOOL_ERROR_PREFIX}${r.content}`,325 }))];326 },327 appendUser(transcript, text) {328 return [...transcript, { type: 'message', role: 'user', content: [{ type: 'input_text', text }] }];329 },330 appendOperatorNote(transcript, text) {331 // `developer` is the operator-authority role of the Responses API and is valid anywhere in `input`.332 return [...transcript, { type: 'message', role: 'developer', content: [{ type: 'input_text', text }] }];333 },334335 async call(req) {336 const built = buildOpenAIRequest(req);337 const { value, attempts } = await withRetries(async () => {338 try {339 const { response, items } = await streamOnce(built, req);340 // A failed response with a transient code throws a retryable ProviderError from here.341 return { response, norm: normaliseOpenAIResponse(response, items, built.strictSchemas, built.originals) };342 } catch (err) {343 const mapped = mapOpenAIError(err);344 if (mapped) throw mapped;345 if (req.signal?.aborted) throw aborted(err, 1);346 throw new ProviderError(err instanceof Error ? err.message : 'Unexpected error while reading the response stream', {347 provider: PROVIDER, retryable: false, code: 'unexpected_error', cause: err,348 });349 }350 }, { ...retry, signal: req.signal }, aborted);351 const { response, norm } = value;352 const tokens = openaiUsage(response.usage);353 const servedBy = typeof response.model === 'string' ? response.model : req.model;354 // The API may answer with a dated snapshot ID; prices are keyed by the requested ID.355 const usd = priceUsage(PROVIDER, priceRow(PROVIDER, req.model) ? req.model : servedBy, tokens);356 const usage: Usage = { usd, ...tokens, turns: 1, toolCalls: 0 };357 return {358 blocks: norm.blocks, stop: norm.stop, usage, model: servedBy, native: norm.native,359 ...(norm.refusal ? { refusal: norm.refusal } : {}),360 nativeStop: norm.nativeStop, attempts,361 } satisfies ProviderResponse;362 },363364 async listLiveModelIds() {365 try {366 const ids: string[] = [];367 for await (const m of getClient().models.list()) ids.push(m.id);368 return ids;369 } catch (err) {370 throw mapOpenAIError(err) ?? new ProviderError('Could not list OpenAI models', { provider: PROVIDER, retryable: false, code: 'list_models_failed', cause: err });371 }372 },373374 async listModels(): Promise<ModelInfo[]> {375 const mine = CATALOGUE.filter((m) => m.provider === PROVIDER);376 let live: Set<string>;377 try {378 live = new Set(await adapter.listLiveModelIds());379 } catch {380 return mine.map((m) => ({ ...m, available: false }));381 }382 // The OpenAI Models API reports no limits or prices; those stay as documented in the catalogue.383 return mine.map((m) => ({ ...m, available: live.has(m.id) }));384 },385386 price(model, usage) {387 return priceUsage(PROVIDER, model, usage);388 },389 };390 return adapter;391}392