ycode/src/main/gateway/index.ts

/**
 * 自建模型网关（OpenAI 兼容）。
 *
 * 职责：
 *  1. 持有上游 baseUrl / apiKey（永不进入渲染进程）
 *  2. 按配置地址选择 Chat Completions / Responses + SSE 流式
 *  3. 把上游的增量 delta 组装成「文本增量」与「完整工具调用」两类输出
 *
 * 设计约束：网关只做协议适配与组装，不含业务策略；
 * 模型路由、上下文裁剪、工具编排由上层（agent/loop.ts）负责。
 */

import type { GatewayConfig, ToolSpec } from '@shared/types'
import { resolveOpenAIEndpoint } from '@shared/openai-endpoint'
import { parseSse } from './sse'
import { buildResponsesRequest, parseResponsesStream } from './responses'

/** 上游 wire 格式的消息 */
export interface WireMessage {
  role: 'system' | 'user' | 'assistant' | 'tool'
  content: string | null | Array<{ type: 'text'; text: string } | { type: 'image_url'; image_url: { url: string } }>
  tool_calls?: WireToolCall[]
  tool_call_id?: string
}

export interface WireToolCall {
  id: string
  type: 'function'
  function: { name: string; arguments: string }
}

/** 网关输出的增量事件 */
export type GatewayDelta =
  | { type: 'text'; text: string }
  | { type: 'reasoning'; text: string }
  | { type: 'tool_calls'; calls: WireToolCall[] }
  | { type: 'finish'; reason: string | null }

export interface GatewayStreamRequest {
  messages: WireMessage[]
  tools?: ToolSpec[]
  signal?: AbortSignal
}

export class GatewayError extends Error {
  constructor(
    message: string,
    readonly status?: number,
    readonly body?: string
  ) {
    super(message)
    this.name = 'GatewayError'
  }
}

export class Gateway {
  constructor(private readonly config: GatewayConfig) {}

  async *stream(req: GatewayStreamRequest): AsyncGenerator<GatewayDelta, void, void> {
    if (!this.config.apiKey && this.config.authMode !== 'none') {
      throw new GatewayError('未配置 API Key，请先在设置中填写')
    }
    if (!this.config.baseUrl) {
      throw new GatewayError('未配置上游地址（baseUrl）')
    }

    let endpoint: ReturnType<typeof resolveOpenAIEndpoint>
    try { endpoint = resolveOpenAIEndpoint(this.config.baseUrl) } catch { throw new GatewayError('上游地址格式无效') }
    const responses = endpoint.api === 'openai-responses'
    const body: Record<string, unknown> = responses ? buildResponsesRequest(this.config, req) : {
      model: this.config.model,
      messages: req.messages,
      stream: true,
      temperature: this.config.temperature
    }
    if (!responses && this.config.maxOutputTokens) body.max_tokens = this.config.maxOutputTokens
    if (!responses && req.tools && req.tools.length > 0) {
      body.tools = req.tools
      body.tool_choice = 'auto'
    }

    let res: Response
    try {
      res = await fetch(endpoint.endpoint, {
        method: 'POST',
        headers: {
          'content-type': 'application/json',
          accept: 'text/event-stream',
          ...(this.config.apiKey ? { authorization: `Bearer ${this.config.apiKey}` } : {})
        },
        body: JSON.stringify(body),
        signal: req.signal
      })
    } catch (err) {
      if ((err as Error).name === 'AbortError') throw err
      throw new GatewayError(responses ? '无法连接 Responses 上游' : `无法连接上游：${(err as Error).message}`)
    }

    if (!res.ok || !res.body) {
      // A Responses provider can echo request credentials in arbitrary error bodies.
      const text = responses ? undefined : await safeReadText(res)
      if (responses) await res.body?.cancel().catch(() => {})
      throw new GatewayError(
        responses ? `Responses 上游返回 HTTP ${res.status}` : `上游返回 ${res.status} ${res.statusText}`.trim(),
        res.status,
        text
      )
    }

    if (responses) {
      yield* parseResponsesStream(res.body, (message) => new GatewayError(message), req.signal)
      return
    }

    // 累加器：按 index 归并流式工具调用分片
    const acc = new Map<number, { id: string; name: string; args: string }>()
    let finishReason: string | null = null

    for await (const evt of parseSse(res.body)) {
      if (evt.data === '[DONE]') break

      let payload: any
      try {
        payload = JSON.parse(evt.data)
      } catch {
        continue // 非 JSON 的心跳/注释行直接跳过
      }

      if (payload?.error) {
        throw new GatewayError(
          `上游错误：${payload.error.message ?? JSON.stringify(payload.error)}`
        )
      }

      const choice = payload?.choices?.[0]
      if (!choice) continue
      const delta = choice.delta ?? {}

      if (typeof delta.content === 'string' && delta.content.length > 0) {
        yield { type: 'text', text: delta.content }
      }
      if (typeof delta.reasoning_content === 'string' && delta.reasoning_content.length > 0) {
        yield { type: 'reasoning', text: delta.reasoning_content }
      }

      if (Array.isArray(delta.tool_calls)) {
        for (const piece of delta.tool_calls) {
          const idx = typeof piece.index === 'number' ? piece.index : 0
          const slot = acc.get(idx) ?? { id: '', name: '', args: '' }
          if (piece.id) slot.id = piece.id
          if (piece.function?.name) slot.name = piece.function.name
          if (piece.function?.arguments) slot.args += piece.function.arguments
          acc.set(idx, slot)
        }
      }

      if (choice.finish_reason) finishReason = choice.finish_reason
    }

    const calls: WireToolCall[] = [...acc.entries()]
      .sort((a, b) => a[0] - b[0])
      .map(([idx, slot]) => ({
        id: slot.id || `call_${idx}_${Date.now()}`,
        type: 'function' as const,
        function: { name: slot.name, arguments: slot.args || '{}' }
      }))

    if (calls.length > 0) yield { type: 'tool_calls', calls }
    yield { type: 'finish', reason: finishReason }
  }
}

async function safeReadText(res: Response): Promise<string> {
  try {
    return (await res.text()).slice(0, 2000)
  } catch {
    return ''
  }
}