宜博 / vokoworld

/**
 * LLMBase —— 适配器三段式的「公共一处」。
 *
 * 公共能力全在这里、且只在这里（生命周期编排）：
 *   - attempt 生命周期（acquire→fetch→消费流→release）、每 attempt 独立超时与组合 signal
 *   - SSE 帧切分外移至 sse.js、mime 魔数外移至 media.js，本文件一行 import 接入、不含其逻辑
 *   - 增量回调编排（text→on_content、reasoning→on_thinking、tool_call→on_tool_call）
 *   - usage 汇总与 reasoning/text 累积
 *   - 重试退避（瞬态错误重试，经 @drama/service-call 纯函数判定）
 *   - trace（经注入的 tracePort，缺省 no-op；本包不写盘、不读全局上下文）
 *   - 多模态消息中间态构建（images/videos → 统一中间态，交由子类落各厂商形态）
 *
 * 子类（providers/llm.<apiType>.js）只覆写两件事：
 *   - async chat(req, opts)：拼本厂商 body/headers，调 this.request(...)
 *   - messageToValue(rawFrame)：把本厂商 SSE 帧解析成统一 { type, ... }
 *
 * 边界纪律：构造里不建连接、不 mkdir、不 appendFile；apiKey/baseUrl 不静态持有，
 * 每次 request 从注入的 keyProvider.acquire() 取、finally release()。
 */

import { LLMError } from './errors.js'
import { mapHttpStatus, mapThrownError, isAbortError, mapChatFinishReason } from './map-error.js'
import { iterateSse } from './sse.js'
import { sniffMime, parseDataUrl } from './media.js'
import { normalizeResponseFormat } from './response-format.js'
import {
  DEFAULT_RETRY_POLICY,
  shouldRetry,
  backoffDelayMs,
  isRetryableReason,
} from '@drama/service-call'

/** 外部 abort reason 常量（与 map-error.js 的 'aborted' 对齐，消除魔法字符串）。 */
const ABORTED_REASON = 'aborted'
export const EMPTY_USAGE = {
  prompt_tokens: 0,
  completion_tokens: 0,
  total_tokens: 0,
  cached_tokens: 0,
  cache_write_tokens: 0,
  reasoning_tokens: 0,
}

/**
 * @typedef {import('./map-error.js').LLMErrorReason} LLMErrorReason
 * @typedef {import('@drama/service-call').RetryPolicy} RetryPolicy
 */

/**
 * 多模态归一后的单一形态：恰含 (mime+data) 或 url 其一。
 * @typedef {object} MediaPart
 * @property {'image'|'video'} kind
 * @property {string} [mime]   本地字节的 mime（sniff/dataUrl 得），与 data 同时存在。
 * @property {string} [data]   base64 负载。
 * @property {string} [url]    远端直传 URL（无本地 mime）。
 */

/**
 * 统一中间帧（messageToValue 的产物）。messageToValue 可返回单个 ParsedValue、其数组、或 null；
 * 同帧多产物（如 Gemini 同帧 text+usage、OpenAI 同帧多路 tool_call 分片）以数组返回，base 展开消费。
 * @typedef {object} ParsedValue
 * @property {'text'|'reasoning'|'tool_call'|'tool_call_delta'|'usage'|'metadata'|'finish'} type
 * @property {string} [text]                          type==='text'|'reasoning' 时的增量。
 * @property {Array<object>} [tool_calls]             type==='tool_call' 时的完整（中性）调用列表。
 * @property {number} [index]                         type==='tool_call_delta' 时的分片归属序号（同一 index 聚合为一个调用）。
 * @property {string} [id]                            type==='tool_call_delta' 时的调用 id（首帧给出）。
 * @property {string} [name]                          type==='tool_call_delta' 时的工具名（首帧给出）。
 * @property {string} [args_fragment]                 type==='tool_call_delta' 时的 arguments JSON 分片（跨帧拼接）。
 * @property {Partial<import('./index.js').ChatUsage>} [usage]
 * @property {string} [provider_request_id]
 * @property {string} [service_tier]
 * @property {string|null} [finish_reason]
 */

/**
 * request() 返回的中性聚合结果（不含任何厂商专有字段）。
 * @typedef {object} StreamResult
 * @property {string} content
 * @property {string} [reasoning]
 * @property {Array<object>} [tool_calls]
 * @property {import('./index.js').ChatUsage} usage
 * @property {string|null} provider_request_id
 * @property {string|null} service_tier
 * @property {string|null} [finish_reason]
 * @property {{ ttft_ms:number|null, duration_ms:number }} [timings]
 *        计时（仅成功路径有）：ttft_ms 为首个 content/reasoning 帧相对 startedAt 的偏移（无帧则 null），
 *        duration_ms 为流消费完成相对 startedAt 的偏移。
 */

export class LLMBase {
  /**
   * @param {object} cfg
   * @param {import('./index.js').KeyProvider} cfg.keyProvider
   * @param {string} cfg.provider                  api_type，决定协议子类与 LLMError.provider。
   * @param {string} [cfg.vendor]                  keyProvider 隔离维度（厂商身份）；缺省回落 provider。
   * @param {string} [cfg.model]                   默认模型。
   * @param {import('./index.js').TracePort} cfg.tracePort
   * @param {RetryPolicy} [cfg.retryPolicy]
   * @param {() => number} [cfg.nowMs]              注入时钟（计时用）；缺省 Date.now。
   */
  constructor({ keyProvider, provider, vendor, model, tracePort, retryPolicy, nowMs }) {
    this.keyProvider = keyProvider
    this.provider = provider
    this.vendor = vendor ?? provider
    this.model = model
    this.tracePort = tracePort
    this.retryPolicy = retryPolicy ?? DEFAULT_RETRY_POLICY
    // 全局 fetch，无注入缝：测试打真实 HTTP 上游（base_url 指向本地真服务），不再注入假 fetch。
    this.httpFetch = fetch
    this.nowMs = nowMs ?? Date.now
  }

  /**
   * 子类必须覆写：拼本厂商请求并返回中性 StreamResult。
   * @param {import('./index.js').ChatRequest} _req
   * @param {import('./index.js').ChatOpts} [_opts]
   * @returns {Promise<StreamResult>}
   */
  async chat(_req, _opts) {
    throw new Error('LLMBase.chat must be overridden by a provider subclass')
  }

  /**
   * 子类必须覆写：把一帧厂商 SSE data 的已解析 JSON 转成统一 ParsedValue（或 null 跳过）。
   * @param {any} _rawFrame
   * @returns {ParsedValue | null}
   */
  messageToValue(_rawFrame) {
    throw new Error('LLMBase.messageToValue must be overridden by a provider subclass')
  }

  /**
   * 能力声明钩子：本 provider 是否支持视频输入。缺省 false（openai/anthropic/moonshot/azure_foundry），
   * 视频厂商（gemini/volcengine）覆写返回 true。与 supportsThinkingParam() 同为能力钩子，
   * 非厂商 if 分支——video 护栏在 request() 单点消费它，基类零厂商分支。
   * @protected
   * @returns {boolean}
   */
  supportsVideo() {
    return false
  }

  /**
   * 多模态归一入口（唯一 mime 归一点）：把 images/videos（Buffer|string）按 MECE 四态归一为
   * 单一 MediaPart 形态（恰含 mime+data 或 url 其一），交由子类落各厂商消息形态。
   * 落地处零 mime 回退：未识别字节在此一次性 badRequest，子类信任 part.mime 非空。
   * @param {Array<Buffer|string>|undefined} items
   * @param {'image'|'video'} kind
   * @returns {Array<MediaPart>}
   */
  buildMediaParts(items, kind) {
    if (!items || items.length === 0) return []
    return items.map((it) => {
      if (Buffer.isBuffer(it)) {
        const mime = sniffMime(it)
        if (mime == null) throw this._mediaError(kind)
        return { kind, mime, data: it.toString('base64') }
      }
      if (it.startsWith('data:')) {
        const parsed = parseDataUrl(it)
        if (parsed == null) throw this._mediaError(kind)
        return { kind, mime: parsed.mime, data: parsed.data }
      }
      if (it.startsWith('http')) return { kind, url: it }
      const mime = sniffMime(Buffer.from(it, 'base64'))
      if (mime == null) throw this._mediaError(kind)
      return { kind, mime, data: it }
    })
  }

  /**
   * @private
   * @param {'image'|'video'} kind
   * @returns {LLMError}
   */
  _mediaError(kind) {
    return new LLMError({
      provider: this.provider,
      reason: 'badRequest',
      retryable: false,
      message: `[${this.provider}] unrecognized ${kind} bytes (no mime magic match)`,
    })
  }

  /**
   * responseFormat 分派表钩子：provider 覆写返回「归一 type → 落地函数」表。
   * 缺省空表——任何 responseFormat 都由 applyResponseFormat 统一 badRequest（不支持结构化输出）。
   * 支持厂商只声明自己能表达的 type（如 moonshot 仅 json_object），不能表达的 type 天然落 badRequest，
   * 无需各写 if-else。
   * @protected
   * @returns {Record<string, (body:Record<string,any>, headers:Record<string,string>, rf:{schema?:object, name?:string}) => void>}
   */
  responseFormatHandlers() {
    return {}
  }

  /**
   * responseFormat（structured output）唯一落地入口：入口一次归一 → 按归一 type 查 provider 分派表 →
   * 命中即落地，未命中显式 badRequest。子类零 if-else、零静默忽略；每个 provider 的 chat 调它一次（null 时 no-op）。
   * @protected
   * @param {Record<string, any>} body
   * @param {Record<string, string>} headers
   * @param {object|undefined} responseFormat
   * @returns {void}
   */
  applyResponseFormat(body, headers, responseFormat) {
    if (responseFormat == null) return
    const normalized = normalizeResponseFormat(responseFormat, this.provider)
    const handler = this.responseFormatHandlers()[normalized.type]
    if (!handler) {
      throw new LLMError({
        provider: this.provider,
        reason: 'badRequest',
        retryable: false,
        message: `[${this.provider}] does not support responseFormat type '${normalized.type}'`,
      })
    }
    handler(body, headers, normalized)
  }

  /**
   * 公共请求执行：acquire key → 带重试地发请求并解析流 → reportUsage/reportRateLimited → release。
   * 子类调它，传入已拼好的「发请求闭包」`doFetch(lease)`，本方法负责其余全部编排。
   *
   * @param {object} args
   * @param {(lease: import('./index.js').KeyLease) => Promise<Response>} args.doFetch
   *        子类用 lease.apiKey/baseUrl 拼 body/headers 并发 fetch，返回原始 Response。
   * @param {import('./index.js').ChatRequest} args.req
   * @param {import('./index.js').ChatOpts} [args.opts]
   * @param {string} [args.affinity_key]
   * @returns {Promise<StreamResult>}
   */
  async request({ doFetch, req, opts = {}, affinity_key }) {
    // 视频护栏（唯一点）：所有子类 chat 都经此。缺省 supportsVideo()===false 覆盖
    // openai/anthropic/moonshot/azure_foundry；视频厂商各自覆写返回 true。基类零厂商分支。
    if (req.videos?.length && !this.supportsVideo()) {
      throw new LLMError({
        provider: this.provider,
        reason: 'badRequest',
        retryable: false,
        message: `[${this.provider}] video input is not supported by this provider`,
      })
    }
    const span = this.tracePort.startSpan(`llm.chat:${this.provider}`)
    span.setAttr('provider', this.provider)
    span.setAttr('model', req.model ?? this.model ?? '')
    // sessionId 一次入口归一：有值才落 span 属性（对齐 gen_ai 语义约定），无值不写占位。
    const sessionId = opts.trace_ctx?.session_id
    if (sessionId) span.setAttr('gen_ai.conversation.id', sessionId)

    // 计时起点：整个 request（含重试）相对它测 ttft / duration，交由 _consumeStream 采样。
    const startedAt = this.nowMs()
    // 单一重试循环：attempt 从 0（首次）起算；每次失败按 @drama/service-call 纯函数判定是否再试。
    let attempt = 0
    try {
      for (;;) {
        if (opts.signal?.aborted) {
          throw new LLMError({ provider: this.provider, reason: ABORTED_REASON, retryable: false })
        }
        try {
          return await this._attempt({ doFetch, req, opts, affinity_key, span, startedAt })
        } catch (err) {
          const mapped = this._toLLMError(err)
          // 外部 abort 不算可重试错误、不重试、不告警；内部 timeout 不落此短路（reason 不同），正常进重试。
          if (isAbortError(err) || mapped.reason === ABORTED_REASON) {
            span.addEvent('aborted', {})
            throw mapped
          }
          attempt += 1
          span.addEvent('error', { reason: mapped.reason, attempt })
          if (!shouldRetry(attempt, mapped.retryable && isRetryableReason(mapped.reason), this.retryPolicy)) {
            throw mapped
          }
          await sleep(backoffDelayMs(attempt, this.retryPolicy), opts.signal)
        }
      }
    } finally {
      span.end()
    }
  }

  /**
   * 单次尝试：acquire key → doFetch → 非 2xx 归一并上报 → 解析流 → reportUsage → 必 release。
   * 任何失败都抛（由 request 的循环决定是否重试）。
   * @private
   * @param {object} a
   * @param {(lease: import('./index.js').KeyLease) => Promise<Response>} a.doFetch
   * @param {import('./index.js').ChatRequest} a.req
   * @param {import('./index.js').ChatOpts} a.opts
   * @param {string} [a.affinity_key]
   * @param {import('./index.js').Span} a.span
   * @param {number} a.startedAt
   * @returns {Promise<StreamResult>}
   */
  async _attempt({ doFetch, req, opts, affinity_key, span, startedAt }) {
    const lease = await this.keyProvider.acquire({
      vendor: `text:${this.vendor}`,
      model: req.model ?? this.model,
      affinity_key,
    })
    // 这次 attempt 是哪把 key 服务的——审计要归因(认证失败时尤其),故 lease.keyId 必须离开这里。
    // 本包不认识 keyrouter:keyId 只是 lease 上的一个值,经回调原样交给上层,不做解释。
    opts.on_key_lease?.(lease.key_id)
    // 每 attempt 独立计时：内部 AbortController 用 TimeoutError 中止，AbortSignal.any 组合外部 opts.signal。
    const ac = new AbortController()
    const signals = opts.signal ? [ac.signal, opts.signal] : [ac.signal]
    const attemptSignal = AbortSignal.any(signals)
    const timer = opts.timeout_ms
      ? setTimeout(() => ac.abort(new DOMException('llm attempt timeout', 'TimeoutError')), opts.timeout_ms)
      : null
    try {
      const res = await doFetch(lease, { signal: attemptSignal })
      if (!res.ok) {
        const { reason, retryable } = mapHttpStatus(res.status)
        const bodyText = await safeReadText(res)
        const body = httpErrorBody(res, bodyText)
        const errorInfo = this.httpErrorInfo(res, body, reason)
        if (reason === 'rateLimit') this.keyProvider.report_rate_limited(lease.key_id, retryAfterMs(res))
        else if (reason === 'auth') this.keyProvider.report_error(lease.key_id, 'auth')
        else if (reason === 'transient') this.keyProvider.report_error(lease.key_id, 'transient')
        throw new LLMError({
          provider: this.provider,
          reason,
          retryable,
          cause: body,
          ...errorInfo,
          detail: {
            type: 'http',
            status: res.status,
            headers: Object.fromEntries(res.headers.entries()),
            body,
          },
        })
      }
      const result = await this._consumeStream(res, attemptSignal, opts, span, startedAt)
      this.keyProvider.report_usage(lease.key_id, result.usage)
      span.addEvent('usage', result.usage)
      return result
    } finally {
      if (timer) clearTimeout(timer)
      // release 必被 await（兼容同步/异步）；失败不覆盖返回值，仅 span 落痕。
      try {
        await lease.release()
      } catch (e) {
        span.addEvent('release_error', { message: String(e?.message ?? e) })
      }
    }
  }

  /**
   * 消费 SSE 流：分帧 → messageToValue → 累积 text/reasoning/tool_calls/usage + 触发回调。
   * @private
   * @param {Response} res
   * @param {AbortSignal} attemptSignal       每 attempt 的组合信号（内部超时 + 外部 abort）；中途可中断读循环。
   * @param {import('./index.js').ChatOpts} opts
   * @param {import('./index.js').Span} span
   * @param {number} startedAt                计时起点（this.nowMs() 域）；ttft/duration 相对它测。
   * @returns {Promise<StreamResult>}
   */
  async _consumeStream(res, attemptSignal, opts, span, startedAt) {
    let content = ''
    let reasoning = ''
    /** @type {number | undefined} 首个 content/reasoning 帧的到达时刻（TTFT 采样点，只采一次）。 */
    let firstFrameAt
    /** @type {Array<object>} */
    const tool_calls = []
    /**
     * 分片 tool_call 聚合缓冲（协议无关，base 独有）：index → { id?, name?, argsText }。
     * 真实流的 function.arguments 跨多帧分片，provider 逐帧发 tool_call_delta，base 按 index 聚合，
     * 流结束时对每个完整调用一次 JSON.parse 得中性 { id, name, args(object) }。每次 _consumeStream 独立、不跨 attempt。
     * @type {Map<number, { id?:string, name?:string, argsText:string }>}
     */
    const toolCallBuffers = new Map()
    /** @type {StreamResult['usage']} */
    let usage = { ...EMPTY_USAGE }
    let provider_request_id = null
    let service_tier = null
    let finish_reason = null
    // 只有输出已经通过回调交付给调用方，才存在不可撤销的流式副作用；无回调时结果尚未离开
    // Promise，失败后仍可安全重试。已交付后禁止 retry/fallback，避免混入两个 attempt 的内容。
    let emittedOutput = false

    try {
      for await (const frame of iterateSse(res, attemptSignal)) {
        opts.on_raw_frame?.(frame)
        // messageToValue 单值/数组统一归一为数组，在入口一处展开（同帧多产物：text+usage、多路 tool_call 分片）。
        for (const value of toValues(this.messageToValue(frame))) {
          switch (value.type) {
            case 'text':
              if (value.text) {
                content += value.text
                firstFrameAt ??= this.nowMs()
                if (opts.on_content) {
                  emittedOutput = true
                  opts.on_content(value.text)
                }
              }
              break
            case 'reasoning':
              if (value.text) {
                reasoning += value.text
                firstFrameAt ??= this.nowMs()
                if (opts.on_thinking) {
                  emittedOutput = true
                  opts.on_thinking(value.text)
                }
              }
              break
            case 'tool_call':
              // 完整（非分片）中性调用：provider 已给 { id, name, args(object) }（整帧下发的厂商）。
              if (value.tool_calls?.length) {
                tool_calls.push(...value.tool_calls)
                if (opts.on_tool_call) {
                  emittedOutput = true
                  opts.on_tool_call(value.tool_calls)
                }
              }
              break
            case 'tool_call_delta':
              bufferToolCallDelta(toolCallBuffers, value)
              break
            case 'usage':
              if (value.usage) usage = { ...usage, ...value.usage }
              break
            case 'metadata':
              provider_request_id = value.provider_request_id
              service_tier = value.service_tier
              break
            case 'finish':
              finish_reason = normalizeFinishReason(value.finish_reason)
              break
            default:
              throw new Error(`Unhandled parsed value type: ${value.type}`)
          }
        }
      }
    } catch (err) {
      if (!emittedOutput) throw err
      const mapped = this._toLLMError(err)
      throw new LLMError({
        provider: this.provider,
        reason: mapped.reason,
        retryable: false,
        cause: err,
        message: `[${this.provider}] stream failed after output started: ${mapped.reason}`,
        detail: { type: 'stream', error: mapped.detail },
      })
    }
    // 流结束：把聚合完成的分片 tool_call 落成中性 { id, name, args(object) }，一次性回调。
    const aggregated = finalizeToolCalls(toolCallBuffers)
    if (aggregated.length) {
      tool_calls.push(...aggregated)
      opts.on_tool_call?.(aggregated)
    }
    span.addEvent('content', { length: content.length })

    // Anthropic 分帧只给输入和输出分量，流结束时补齐总量。
    if (!usage.total_tokens) {
      usage = {
        ...usage,
        total_tokens: usage.prompt_tokens + usage.completion_tokens,
      }
    }

    const finishError = chatFinishReasonError({
      provider: this.provider,
      finish_reason,
      content_length: content.length,
      usage,
      provider_request_id,
    })
    if (finishError) {
      if (emittedOutput) {
        throw new LLMError({
          provider: this.provider,
          reason: finishError.reason,
          retryable: false,
          cause: finishError,
          message: `[${this.provider}] stream ended with finish_reason=${finish_reason} after output started`,
          detail: {
            type: 'stream_finish_reason',
            finish_reason,
            content_length: content.length,
            usage,
          },
          provider_request_id,
        })
      }
      throw finishError
    }

    /** @type {StreamResult} */
    const out = { content, usage, provider_request_id, service_tier, finish_reason }
    if (reasoning) out.reasoning = reasoning
    if (tool_calls.length) out.tool_calls = tool_calls
    // 计量取整只在出口一次：ttft 仅在有帧时有值，duration 恒为流消费总时长。
    out.timings = {
      ttft_ms: firstFrameAt === undefined ? null : firstFrameAt - startedAt,
      duration_ms: this.nowMs() - startedAt,
    }
    return out
  }

  /**
   * 把任意抛出物归一为 LLMError（已是 LLMError 则原样）。
   * @private
   * @param {unknown} err
   * @returns {LLMError}
   */
  _toLLMError(err) {
    if (err instanceof LLMError) return err
    const { reason, retryable } = mapThrownError(err)
    if (!(err instanceof Error)) throw new Error(`[${this.provider}] provider 抛出了非 Error 值`)
    return new LLMError({
      provider: this.provider,
      reason,
      retryable,
      cause: err,
      message: err.name === 'KeyRouterError' ? `[${this.provider}] ${err.message}` : undefined,
      detail: { type: 'runtime', name: err.name, message: err.message },
    })
  }

  httpErrorInfo(res, _body, reason) {
    return {
      message: `[${this.provider}] HTTP ${res.status}: ${reason}`,
      provider_request_id: res.headers.get('x-request-id'),
    }
  }
}

function normalizeFinishReason(reason) {
  if (reason === null || typeof reason === 'string') return reason
  throw new Error(`finish_reason must be string or null, got ${typeof reason}`)
}

function chatFinishReasonError({ provider, finish_reason, content_length, usage, provider_request_id }) {
  const mapped = mapChatFinishReason(finish_reason)
  if (mapped === null) return null
  return new LLMError({
    provider,
    reason: mapped.reason,
    retryable: mapped.retryable,
    message: `[${provider}] chat completion ended with finish_reason=${finish_reason}`,
    detail: {
      type: 'finish_reason',
      finish_reason,
      content_length,
      usage,
    },
    provider_request_id,
  })
}

/**
 * messageToValue 返回归一为 ParsedValue 数组（单值/数组/null 三态一处收口），消费端零形态判断。
 * @param {import('./base.js').ParsedValue | Array<import('./base.js').ParsedValue> | null | undefined} v
 * @returns {Array<import('./base.js').ParsedValue>}
 */
function toValues(v) {
  if (v == null) return []
  return Array.isArray(v) ? v : [v]
}

/**
 * 把一帧 tool_call_delta 并入按 index 聚合的缓冲：id/name 首帧给出（合并），args_fragment 跨帧拼接。
 * @param {Map<number, { id?:string, name?:string, argsText:string }>} buffers
 * @param {{ index?:number, id?:string, name?:string, args_fragment?:string }} delta
 * @returns {void}
 */
function bufferToolCallDelta(buffers, delta) {
  const index = delta.index ?? 0
  let buf = buffers.get(index)
  if (!buf) {
    buf = { argsText: '' }
    buffers.set(index, buf)
  }
  if (delta.id != null) buf.id = delta.id
  if (delta.name != null) buf.name = delta.name
  if (delta.args_fragment) buf.argsText += delta.args_fragment
}

/**
 * 聚合缓冲 → 中性 tool_call 列表：按 index 升序，每个 argsText 做一次 JSON.parse（args 恒为对象）。
 * args 为空时归一为 {}；非法 JSON 由 JSON.parse 自然抛出（真实上游不会给坏 JSON，坏则是真错误）。
 * @param {Map<number, { id?:string, name?:string, argsText:string }>} buffers
 * @returns {Array<{ id?:string, name?:string, args:object }>}
 */
function finalizeToolCalls(buffers) {
  return [...buffers.entries()]
    .sort((a, b) => a[0] - b[0])
    .map(([, buf]) => ({ id: buf.id, name: buf.name, args: JSON.parse(buf.argsText || '{}') }))
}

/**
 * 安全读取响应文本（错误体），失败返回空串。
 * @param {Response} res
 * @returns {Promise<string>}
 */
async function safeReadText(res) {
  try {
    return await res.text()
  } catch {
    return ''
  }
}

function httpErrorBody(res, bodyText) {
  const contentType = res.headers.get('content-type')
  return contentType !== null && contentType.startsWith('application/json')
    ? JSON.parse(bodyText)
    : bodyText
}

/**
 * 从 429 响应取 Retry-After（秒）→ 毫秒；无则 undefined。
 * @param {Response} res
 * @returns {number | undefined}
 */
function retryAfterMs(res) {
  const h = res.headers?.get?.('retry-after')
  if (!h) return undefined
  const sec = Number(h)
  return Number.isFinite(sec) ? sec * 1000 : undefined
}

/**
 * 可被 abort 的 sleep。
 * @param {number} ms
 * @param {AbortSignal} [signal]
 * @returns {Promise<void>}
 */
function sleep(ms, signal) {
  return new Promise((resolve, reject) => {
    if (signal?.aborted) {
      reject(abortReason(signal))
      return
    }
    const t = setTimeout(() => {
      signal?.removeEventListener?.('abort', onAbort)
      resolve()
    }, ms)
    function onAbort() {
      clearTimeout(t)
      reject(abortReason(signal))
    }
    signal?.addEventListener?.('abort', onAbort, { once: true })
  })
}

/**
 * 中止原因：优先透传 signal.reason（内部超时的 TimeoutError / 外部的 AbortError），缺省造 AbortError。
 * @param {AbortSignal} [signal]
 * @returns {unknown}
 */
function abortReason(signal) {
  return signal?.reason ?? Object.assign(new Error('aborted'), { name: 'AbortError' })
}