宜博 / vokoworld

/**
 * OpenAI Responses API 子类 —— POST /responses 协议，OpenAI 厂商默认适配器。
 *
 * 与 Chat Completions（llm.openai.js）的差异仅在：端点（/responses）、请求体
 * （`input` items / `instructions` / `store` / `prompt_cache_key`）与流事件类型。
 * 复用父类 OpenAIProvider 的 Bearer header（chatCompletionsHeaders）与 defaultBaseUrl
 * （https://api.openai.com/v1）；chat / messageToValue / 消息映射整体覆写。
 *
 * 网关无状态（store:false，不依赖服务端会话历史）：req.messages 全量映射为 input items。
 *
 * 请求体字段与流事件类型以 OpenAI Responses API 当前官方文档为准：
 *   请求：POST /responses，body { model, input, instructions?, stream, store, prompt_cache_key?,
 *          temperature?, max_output_tokens? }。
 *   流事件（SSE `event: <type>\ndata: {json}`，data 内 `type` 即事件类型）：
 *     response.output_text.delta            → text 增量（取 delta）
 *     response.reasoning_summary_text.delta → reasoning 增量（取 delta）
 *     response.output_item.done(function_call) → tool_call
 *     response.completed                    → response.usage 用量
 *   usage 字段：input_tokens / output_tokens / total_tokens /
 *     input_tokens_details.cached_tokens / input_tokens_details.cache_write_tokens。
 *
 * 流解析沿用 LLMBase 的 iterateSse（标准 SSE `data:` 分帧，帧内 type 承载事件类型），不覆写。
 */

import { LLMError } from '../errors.js'
import { mapResponsesFailure } from '../map-error.js'
import OpenAIProvider, {
  readOptionalUsageInteger,
  readUsageInteger,
} from './llm.openai.js'

// 成功终态：携 usage、走成功路径。completed=完整、incomplete=受控部分结果（如 max_output_tokens
// 耗尽，语义等同 chat 的 finish_reason=length）。response.failed 是明确失败，不在此列，单独抛错。
const TERMINAL_FRAMES = new Set(['response.completed', 'response.incomplete'])

export default class OpenAIResponsesProvider extends OpenAIProvider {
  /**
   * @param {import('../index.js').ChatRequest} req
   * @param {import('../index.js').ChatOpts} [opts]
  * @returns {Promise<import('../base.js').StreamResult>}
  */
  async chat(req, opts = {}) {
    const model = req.model ?? this.model
    return this.request({
      req,
      opts,
      affinity_key: opts.trace_ctx?.session_id,
      doFetch: (lease, { signal }) => {
        const base = lease.base_url ?? this.defaultBaseUrl()
        const headers = this.chatCompletionsHeaders(lease)
        const { instructions, input } = this._toResponsesInput(req)
        /** @type {Record<string, any>} */
        const body = { model, input, stream: true, store: false }
        if (instructions) body.instructions = instructions
        if (opts.prompt_cache_key) body.prompt_cache_key = opts.prompt_cache_key
        if (req.prompt_cache_options) body.prompt_cache_options = req.prompt_cache_options
        if (req.temperature != null) body.temperature = req.temperature
        if (req.max_tokens != null) body.max_output_tokens = req.max_tokens
        this.applyResponseFormat(body, headers, req.response_format)
        const request = this.responsesRequest(lease, req, opts, headers, body)
        return this.httpFetch(`${base}/responses`, {
          method: 'POST',
          signal,
          headers: request.headers,
          body: JSON.stringify(request.body),
        })
      },
    })
  }

  /**
   * Responses 请求扩展点。公共端点原样返回；内部协议适配器在这里一次性改请求头和请求体。
   * @protected
   * @param {import('../index.js').KeyLease} lease
   * @param {import('../index.js').ChatRequest} _req
   * @param {import('../index.js').ChatOpts} _opts
   * @param {Record<string, string>} headers
   * @param {Record<string, any>} body
   */
  responsesRequest(_lease, _req, _opts, headers, body) {
    return { headers, body }
  }

  /**
   * Responses API structured output 分派表：以 text.format 表达（非 Chat Completions 的 response_format）。
   *   json_object → text.format.type='json_object'。
   *   json_schema → text.format.json_schema（name+schema+strict）。
   * @protected
   * @returns {Record<string, (body:Record<string,any>, headers:Record<string,string>, rf:{schema?:object, name?:string}) => void>}
   */
  responseFormatHandlers() {
    return {
      json_object: (body) => {
        body.text = { ...(body.text ?? {}), format: { type: 'json_object' } }
      },
      json_schema: (body, _headers, rf) => {
        body.text = {
          ...(body.text ?? {}),
          format: { type: 'json_schema', name: rf.name ?? 'response', schema: rf.schema, strict: true },
        }
      },
    }
  }

  /**
   * messages → Responses 的 { instructions?, input }。system 累积为顶层 instructions，
   * 其余转 input item（assistant→output_text、其余→input_text）。全量传入（store:false 无状态）。
   * @private
   * @param {import('../index.js').ChatRequest} req
   * @returns {{ instructions?:string, input:Array<any> }}
   */
  _toResponsesInput(req) {
    /** @type {string[]} */
    const systemTexts = []
    /** @type {Array<any>} */
    const input = []
    for (const m of req.messages) {
      if (m.role === 'system') {
        systemTexts.push(String(m.content))
        continue
      }
      const content = {
        type: m.role === 'assistant' ? 'output_text' : 'input_text',
        text: String(m.content),
      }
      if (req.prompt_cache_options && m.prompt_cache_breakpoint === true) {
        content.prompt_cache_breakpoint = { mode: 'explicit' }
      }
      input.push({
        role: m.role,
        content: [content],
      })
    }
    const images = this.buildMediaParts(req.images, 'image')
    if (images.length && input.length) {
      const last = input[input.length - 1]
      for (const img of images) {
        const url = img.url ?? `data:${img.mime};base64,${img.data}`
        last.content.push({ type: 'input_image', image_url: url })
      }
    }
    const out = { input }
    if (systemTexts.length) out.instructions = systemTexts.join('\n')
    return out
  }

  /**
   * @param {any} frame Responses SSE data 帧（已解析 JSON），frame.type 为事件类型。
   * @returns {import('../base.js').ParsedValue | null}
   */
  messageToValue(frame) {
    if (!frame || typeof frame !== 'object') return null
    if (frame.type === 'response.output_text.delta') return { type: 'text', text: frame.delta }
    if (frame.type === 'response.reasoning_summary_text.delta') {
      return { type: 'reasoning', text: frame.delta }
    }
    if (frame.type === 'response.output_item.done' && frame.item?.type === 'function_call') {
      // Responses 的 function_call 参数经 response.function_call_arguments.delta 分片，最终在 output_item.done
      // 携全量 arguments 字符串。归一为 base 的分片机制：发单条 tool_call_delta（整帧即全量片段），
      // 由 base 按 index 聚合 + 一次 JSON.parse 得中性 { id, name, args(object) }，不在 provider 拼装 wire 形态。
      return {
        type: 'tool_call_delta',
        index: frame.output_index ?? 0,
        id: frame.item.call_id,
        name: frame.item.name,
        args_fragment: frame.item.arguments ?? '',
      }
    }
    // 明确失败终态：从 response.error（code/message）构造 LLMError 抛出（沿用 azure_foundry
    // 在 provider 内抛 LLMError 的既有约定）。经 base 的 _consumeStream→request 重试循环，reason
    // 决定重试与否；该次调用以抛错结束，createLLM 不进成功计费路径，审计落 status='error'。
    if (frame.type === 'response.failed') {
      const error = frame.response?.error ?? {}
      const { reason, retryable } = mapResponsesFailure(error.code)
      throw new LLMError({
        provider: this.provider,
        reason,
        retryable,
        cause: error,
        message: `[${this.provider}] response.failed (${error.code ?? 'unknown'}): ${error.message ?? ''}`,
        detail: { type: 'provider_response', response: error },
      })
    }
    // completed/incomplete 两终态是成功路径——incomplete（如思维链耗尽 max_output_tokens）时上游
    // 已对 reasoning token 计费，漏读即漏计费，故与 completed 同走成功 + usage 路径。
    //
    // 终态必带 usage：Responses 规范允许 `usage: null`（见 API reference 的 incomplete 示例），
    // 但零 usage 会让本次调用以 amount=0 静默跳账 + 审计 status='ok' 落地——上游已产生费用却无账可
    // 对，违反「运行期不存在免费调用路径」。故缺 usage 即显式失败，绝不降级为零成本成功。
    if (TERMINAL_FRAMES.has(frame.type)) {
      const u = frame.response?.usage
      if (!u) {
        throw new LLMError({
          provider: this.provider,
          reason: 'unknown',
          retryable: false,
          message: `[${this.provider}] ${frame.type} 未携带 usage（上游已计费但本地无法计价，拒绝记为零成本成功）`,
        })
      }
      const input_details = u.input_tokens_details
      const prompt_tokens = readUsageInteger(u.input_tokens, 'input_tokens', this.provider)
      const completion_tokens = readUsageInteger(u.output_tokens, 'output_tokens', this.provider)
      const total_tokens = readUsageInteger(u.total_tokens, 'total_tokens', this.provider)
      const cached_tokens = readOptionalUsageInteger(
        input_details?.cached_tokens,
        'input_tokens_details.cached_tokens',
        this.provider,
      )
      const cache_write_tokens = readOptionalUsageInteger(
        input_details?.cache_write_tokens,
        'input_tokens_details.cache_write_tokens',
        this.provider,
      )
      const reasoning_tokens = readOptionalUsageInteger(
        u.output_tokens_details?.reasoning_tokens,
        'output_tokens_details.reasoning_tokens',
        this.provider,
      )
      if (cached_tokens + cache_write_tokens > prompt_tokens) {
        throw new Error(
          `[${this.provider}] cached_tokens + cache_write_tokens 不能大于 input_tokens`,
        )
      }
      if (total_tokens !== prompt_tokens + completion_tokens) {
        throw new Error(`[${this.provider}] total_tokens 必须等于 input_tokens + output_tokens`)
      }
      return [
        {
          type: 'metadata',
          provider_request_id: typeof frame.response.id === 'string' ? frame.response.id : null,
          service_tier: typeof frame.response.service_tier === 'string' ? frame.response.service_tier : null,
        },
        {
          type: 'usage',
          usage: {
            prompt_tokens,
            completion_tokens,
            total_tokens,
            cached_tokens,
            cache_write_tokens,
            reasoning_tokens,
          },
        },
      ]
    }
    return null
  }
}