apps/api/src/ai/client.ts

// apps/api/src/ai/client.ts
// Server-side AI client. Ported from apps/web-legacy/scripts/ai-client.js with
// these adjustments:
//   - No browser CORS proxy. The backend talks directly to model endpoints.
//   - Optional outbound proxy via UPSTREAM_PROXY (socks5:// or http://) using
//     undici's ProxyAgent. Replaces the legacy proxy.mjs CONNECT tunnel.
//   - Streaming support is preserved but defaults to non-streaming since the
//     analyzer pipeline always wants the full payload before parsing.
//   - All API keys arrive via constructor / per-call config, never logged.

import { ProxyAgent, fetch as undiciFetch, type Dispatcher } from "undici";

// ── Public types ───────────────────────────────────────────────────────────
export interface AiClientConfig {
  apiKey: string;
  baseURL?: string;
  apiMode?: "chat" | "responses";
  model?: string;
  modelPreset?: string;
  customModel?: string;
  temperature?: number;
  topP?: number;
  maxTokens?: number;
  enableThinking?: boolean;
  clearThinking?: boolean;
}

export interface ChatMessage {
  role: "system" | "user" | "assistant" | "tool" | string;
  content: string;
}

export interface RequestOptions {
  stream?: boolean;
  jsonMode?: boolean;
  jsonModeFallback?: boolean;
  pluginCompat?: boolean;
  disableBetaParameterFallback?: boolean;
  schema?: unknown;
  schemaName?: string;
  retries?: number;
  temperature?: number;
  topP?: number;
  maxTokens?: number;
  endpoint?: string; // computed internally; provider hint for special headers
  onRetry?: (err: AiError, retryDelayMs: number, attempt: number) => void;
  onTextDelta?: (collected: string, delta: string) => void;
  onReasoningDelta?: (collected: string, delta: string) => void;
}

export interface AiResponse {
  text: string;
  endpoint: string;
  fetchURL: string;
  apiMode: "chat" | "responses";
}

export class AiError extends Error {
  status: number;
  isAuthError = false;
  retryable = false;
  isTruncated = false;
  isJsonParseError = false;
  rawPreview?: string;
  endpoint?: string;
  apiMode?: string;
  original?: unknown;

  constructor(message: string, status = 0) {
    super(message);
    this.status = status;
  }
}

// ── Helpers ────────────────────────────────────────────────────────────────
function clamp(value: number, min: number, max: number): number {
  return Math.min(max, Math.max(min, value));
}

function numberOrDefault(value: unknown, fallback: number): number {
  const parsed = Number(value);
  return Number.isFinite(parsed) ? parsed : fallback;
}

function optionOrConfig<T>(
  options: Record<string, unknown> | undefined,
  config: Record<string, unknown> | undefined,
  key: string,
  fallback: T
): T {
  if (options && options[key] !== undefined) return options[key] as T;
  if (config && config[key] !== undefined) return config[key] as T;
  return fallback;
}

export function resolveApiMode(
  baseURL: string | undefined,
  fallback: "chat" | "responses" | undefined
): "chat" | "responses" {
  const value = String(baseURL || "");
  if (/\/chat\/completions(\/?|$)/i.test(value)) return "chat";
  if (/\/responses(\/?|$)/i.test(value)) return "responses";
  return fallback === "chat" ? "chat" : "responses";
}

export function resolveEndpoint(
  baseURL: string | undefined,
  apiMode: "chat" | "responses"
): string {
  const targetSuffix = apiMode === "chat" ? "/chat/completions" : "/responses";
  const value = String(baseURL || "").trim();
  if (!value) return "https://api.openai.com/v1" + targetSuffix;
  let trimmed = value.replace(/\/+$/g, "");
  trimmed = trimmed.replace(/\/(responses|chat\/completions)$/i, "");
  if (/\/v\d+$/i.test(trimmed) || /\/openai\/v\d+$/i.test(trimmed)) {
    return trimmed + targetSuffix;
  }
  return trimmed + "/v1" + targetSuffix;
}

function normalizeModel(config: AiClientConfig): string {
  if (config.model) return config.model;
  if (config.modelPreset === "custom") return config.customModel || "";
  return config.modelPreset || "";
}

function normalizeMessagesForResponses(messages: ChatMessage[]) {
  return (messages || []).map((m) => ({
    role: m.role,
    content: [{ type: "input_text", text: String(m.content || "") }],
  }));
}

function buildHeaders(
  config: AiClientConfig,
  endpoint: string
): Record<string, string> {
  if (!config.apiKey) {
    throw new AiError("缺少 API Key（请在后端 .env 中配置 DEFAULT_API_KEY）。", 0);
  }
  const headers: Record<string, string> = { "Content-Type": "application/json" };
  if (/azure\.com/i.test(endpoint)) headers["api-key"] = config.apiKey;
  else headers.Authorization = "Bearer " + config.apiKey;
  return headers;
}

function buildRequestBody(
  config: AiClientConfig,
  messages: ChatMessage[],
  options: RequestOptions
): Record<string, unknown> {
  const apiMode: "chat" | "responses" = config.apiMode === "chat" ? "chat" : "responses";
  const model = normalizeModel(config);
  if (!model) throw new AiError("缺少模型名（model）。", 0);

  if (apiMode === "chat") {
    if (options.pluginCompat) {
      return {
        model,
        messages: messages || [],
        temperature: clamp(
          numberOrDefault(
            optionOrConfig(
              options as Record<string, unknown>,
              config as unknown as Record<string, unknown>,
              "temperature",
              0.7
            ),
            0.7
          ),
          0,
          2
        ),
      };
    }
    const body: Record<string, unknown> = {
      model,
      messages: messages || [],
      temperature: clamp(
        numberOrDefault(
          optionOrConfig(options as Record<string, unknown>, config as unknown as Record<string, unknown>, "temperature", 0.7),
          0.7
        ),
        0,
        2
      ),
      top_p: clamp(
        numberOrDefault(
          optionOrConfig(options as Record<string, unknown>, config as unknown as Record<string, unknown>, "topP", 1),
          1
        ),
        0,
        1
      ),
      max_tokens: clamp(
        numberOrDefault(
          optionOrConfig(options as Record<string, unknown>, config as unknown as Record<string, unknown>, "maxTokens", 16384),
          16384
        ),
        1,
        131072
      ),
      stream: Boolean(options.stream),
    };
    if (options.jsonMode !== false) {
      body.response_format = { type: "json_object" };
    }
    const ep = String(options.endpoint || "");
    if (/bigmodel\.cn|zhipuai/i.test(ep)) {
      body.thinking = {
        type: config.enableThinking === false ? "disabled" : "enabled",
      };
    } else if (
      (config.enableThinking || config.clearThinking) &&
      /dashscope|aliyun|qwen|modelscope/i.test(ep)
    ) {
      body.chat_template_kwargs = {
        enable_thinking: Boolean(config.enableThinking),
        clear_thinking: Boolean(config.clearThinking),
      };
    }
    return body;
  }

  // Responses API
  const responsesBody: Record<string, unknown> = {
    model,
    store: false,
    temperature: clamp(
      numberOrDefault(
        optionOrConfig(options as Record<string, unknown>, config as unknown as Record<string, unknown>, "temperature", 0.7),
        0.7
      ),
      0,
      2
    ),
    input: normalizeMessagesForResponses(messages),
  };
  if (options.schema) {
    responsesBody.text = {
      format: {
        type: "json_schema",
        name: options.schemaName || "structured_result",
        strict: true,
        schema: options.schema,
      },
    };
  }
  if (options.stream) responsesBody.stream = true;
  return responsesBody;
}

// ── Response parsing ───────────────────────────────────────────────────────
function safePreview(text: string): string {
  return String(text || "")
    .replace(/sk-[A-Za-z0-9_-]{8,}/g, "sk-***")
    .replace(/nvapi-[A-Za-z0-9_-]{8,}/g, "nvapi-***")
    .replace(/Bearer\s+[A-Za-z0-9._-]+/gi, "Bearer ***")
    .slice(0, 1200);
}

function normalizeText(value: unknown): string {
  if (typeof value === "string") return value;
  if (Array.isArray(value)) {
    return value
      .map((part) => {
        if (typeof part === "string") return part;
        if (part && typeof (part as Record<string, unknown>).text === "string")
          return (part as Record<string, unknown>).text as string;
        if (part && typeof (part as Record<string, unknown>).content === "string")
          return (part as Record<string, unknown>).content as string;
        return "";
      })
      .join("");
  }
  return "";
}

function extractChatText(payload: unknown): string {
  const p = payload as
    | { choices?: Array<{ finish_reason?: string; message?: { content?: unknown } }> }
    | undefined;
  const choice = p?.choices?.[0];
  if (!choice) return "";
  const finish = choice.finish_reason || "";
  if (finish === "length" || finish === "max_tokens") {
    const err = new AiError(
      "模型输出被截断（token 超限）。请减少录制文件大小，或在设置中提高 Max Tokens。",
      0
    );
    err.isTruncated = true;
    err.rawPreview = safePreview(normalizeText(choice.message?.content));
    throw err;
  }
  return choice.message ? normalizeText(choice.message.content) : "";
}

function extractResponsesText(payload: unknown): string {
  const p = payload as
    | { output_text?: string; output?: Array<{ content?: Array<{ type?: string; text?: string }> }> }
    | undefined;
  if (!p) return "";
  if (typeof p.output_text === "string") return p.output_text;
  if (!Array.isArray(p.output)) return "";
  const chunks: string[] = [];
  p.output.forEach((item) => {
    (item.content || []).forEach((part) => {
      if (part && part.type === "output_text" && typeof part.text === "string")
        chunks.push(part.text);
      else if (part && typeof part.text === "string") chunks.push(part.text);
    });
  });
  return chunks.join("");
}

// ── JSON helpers ───────────────────────────────────────────────────────────
function stripJsonNoise(text: string): string {
  return String(text || "")
    .replace(/<think[\s\S]*?<\/think>/gi, "")
    .replace(/^```(?:json)?\s*/i, "")
    .replace(/```\s*$/i, "")
    .trim();
}

function extractJsonObject(text: string): string {
  const start = text.indexOf("{");
  const end = text.lastIndexOf("}");
  return start >= 0 && end > start ? text.slice(start, end + 1) : "";
}

function extractBalancedJsonObjects(text: string): string[] {
  const out: string[] = [];
  let start = -1;
  let depth = 0;
  let inString = false;
  let escaped = false;
  for (let i = 0; i < text.length; i += 1) {
    const ch = text[i];
    if (inString) {
      if (escaped) escaped = false;
      else if (ch === "\\") escaped = true;
      else if (ch === '"') inString = false;
      continue;
    }
    if (ch === '"') {
      inString = true;
      continue;
    }
    if (ch === "{") {
      if (depth === 0) start = i;
      depth += 1;
      continue;
    }
    if (ch === "}" && depth > 0) {
      depth -= 1;
      if (depth === 0 && start >= 0) {
        out.push(text.slice(start, i + 1));
        start = -1;
      }
    }
  }
  return out;
}

export function parseJsonText(text: string): unknown {
  const raw = String(text || "").trim();
  if (!raw) throw new Error("模型返回为空，无法解析 JSON。");
  const cleaned = stripJsonNoise(raw);
  const candidates = [raw, cleaned, ...extractBalancedJsonObjects(cleaned).reverse(), extractJsonObject(cleaned)].filter(
    Boolean
  );
  const seen: Record<string, true> = {};
  for (const cand of candidates) {
    if (seen[cand]) continue;
    seen[cand] = true;
    try {
      return JSON.parse(cand);
    } catch (_) {
      // try next
    }
  }
  throw new Error("模型返回的内容不是合法 JSON。");
}

// ── HTTP error helpers ─────────────────────────────────────────────────────
function shouldFallbackJsonMode(
  error: AiError,
  body: Record<string, unknown>,
  options: RequestOptions,
  tried: boolean
): boolean {
  if (tried || !body || !body.response_format || options.jsonModeFallback === false)
    return false;
  return Number(error?.status) === 400;
}

function shouldFallbackBetaParameters(
  error: AiError,
  tried: boolean,
  options: RequestOptions
): boolean {
  if (options.disableBetaParameterFallback) return false;
  if (tried || Number(error?.status) !== 400) return false;
  return /beta[-\s]?limitations|temperature[\s\S]*top_p[\s\S]*(?:fixed|1)|presence_penalty|frequency_penalty/i.test(
    error?.message || ""
  );
}

function applyBetaParameterLimits(
  body: Record<string, unknown>,
  apiMode: "chat" | "responses"
): Record<string, unknown> {
  const next: Record<string, unknown> = { ...body, temperature: 1 };
  if (apiMode === "chat") {
    next.top_p = 1;
    next.n = 1;
  }
  delete next.presence_penalty;
  delete next.frequency_penalty;
  return next;
}

async function parseErrorResponse(response: Response): Promise<AiError> {
  const status = response.status;
  let message = "HTTP " + status;
  try {
    const payload = await response.json();
    const p = payload as { error?: { message?: string; code?: string }; message?: string };
    if (p && p.error) {
      message +=
        " - " + (p.error.message || p.error.code || JSON.stringify(p.error));
    } else if (p && p.message) {
      message += " - " + p.message;
    }
  } catch (_) {
    try {
      const text = await response.text();
      if (text) message += " - " + text.slice(0, 300);
    } catch (_ignore) {
      // ignore
    }
  }
  const err = new AiError(message, status);
  err.isAuthError = status === 401 || status === 403;
  err.retryable = status === 429 || (status >= 500 && status < 600);
  return err;
}

function isRetryable(error: AiError | Error | undefined): boolean {
  if (!error) return false;
  if (error instanceof AiError && typeof error.retryable === "boolean")
    return error.retryable;
  const status = Number((error as AiError).status) || 0;
  return status === 429 || (status >= 500 && status < 600);
}

function delay(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

// ── Main client ────────────────────────────────────────────────────────────
export interface AiClientOptions {
  /** Optional UPSTREAM_PROXY URI (socks5:// or http://). */
  upstreamProxy?: string;
}

export class AiClient {
  private dispatcher?: Dispatcher;

  constructor(opts: AiClientOptions = {}) {
    if (opts.upstreamProxy) {
      this.dispatcher = new ProxyAgent(opts.upstreamProxy);
    }
  }

  private fetchImpl: typeof undiciFetch = (url, init) =>
    undiciFetch(url, this.dispatcher ? { ...init, dispatcher: this.dispatcher } : init);

  /** Call model and return parsed JSON. Recovers from common malformed outputs. */
  async requestJson(
    config: AiClientConfig,
    messages: ChatMessage[],
    options: RequestOptions = {}
  ): Promise<unknown> {
    const raw = await this.requestText(config, messages, options);
    try {
      return parseJsonText(raw.text);
    } catch (error) {
      if ((error as AiError).isTruncated) throw error;
      const next = new AiError(
        "模型返回非 JSON，已截取原始返回用于诊断。可切换 Responses API 或关闭 thinking 后重试。",
        0
      );
      next.retryable = false;
      next.isJsonParseError = true;
      next.rawPreview = safePreview(raw.text);
      next.endpoint = raw.endpoint;
      next.apiMode = raw.apiMode;
      next.original = error;
      throw next;
    }
  }

  /** Call model and return raw text. Handles retries + JSON-mode fallback + streaming. */
  async requestText(
    config: AiClientConfig,
    messages: ChatMessage[],
    options: RequestOptions = {}
  ): Promise<AiResponse> {
    const apiMode = resolveApiMode(config.baseURL, config.apiMode);
    const endpoint = resolveEndpoint(config.baseURL, apiMode);
    const effectiveConfig: AiClientConfig = { ...config, apiMode };
    let body = buildRequestBody(effectiveConfig, messages, { ...options, endpoint });
    const headers = buildHeaders(effectiveConfig, endpoint);

    const retries = options.retries == null ? 3 : Number(options.retries);
    let attempt = 0;
    let lastError: AiError | Error | undefined;
    let jsonModeFallbackTried = false;
    let betaParameterFallbackTried = false;

    while (attempt <= retries) {
      try {
        const response = await this.fetchImpl(endpoint, {
          method: "POST",
          headers,
          body: JSON.stringify(body),
        });
        if (!response.ok) {
          const httpError = await parseErrorResponse(response as unknown as Response);
          if (shouldFallbackBetaParameters(httpError, betaParameterFallbackTried, options)) {
            betaParameterFallbackTried = true;
            body = applyBetaParameterLimits(body, apiMode);
            attempt = 0;
            continue;
          }
          if (
            shouldFallbackJsonMode(httpError, body, options, jsonModeFallbackTried)
          ) {
            jsonModeFallbackTried = true;
            body = { ...body };
            delete body.response_format;
            attempt = 0;
            continue;
          }
          throw httpError;
        }
        const text =
          options.stream && response.body
            ? await readStream(response as unknown as Response, apiMode, options)
            : await readJsonPayload(response as unknown as Response, apiMode);
        return { text, endpoint, fetchURL: endpoint, apiMode };
      } catch (error) {
        lastError = error as AiError;
        if (!isRetryable(lastError) || attempt >= retries) throw lastError;
        attempt += 1;
        const retryDelay = Math.pow(2, attempt - 1) * 1000;
        if (typeof options.onRetry === "function") {
          options.onRetry(lastError as AiError, retryDelay, attempt);
        }
        await delay(retryDelay);
      }
    }
    throw lastError || new AiError("请求失败。");
  }
}

// ── Streaming reader (SSE) ─────────────────────────────────────────────────
async function readStream(
  response: Response,
  apiMode: "chat" | "responses",
  options: RequestOptions
): Promise<string> {
  if (!response.body) return "";
  const reader = (response.body as ReadableStream<Uint8Array>).getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  let collected = "";
  let reasoning = "";

  const consumeEvent = (rawEvent: string) => {
    const parsed = parseSSEEvent(rawEvent);
    if (!parsed) return;
    if (apiMode === "chat") {
      const choice = (parsed.choices && parsed.choices[0]) as
        | { delta?: { reasoning_content?: unknown; content?: unknown } }
        | undefined;
      const delta = choice?.delta;
      if (!delta) return;
      const reasoningDelta = normalizeText(delta.reasoning_content);
      if (reasoningDelta) {
        reasoning += reasoningDelta;
        options.onReasoningDelta?.(reasoning, reasoningDelta);
      }
      const contentDelta = normalizeText(delta.content);
      if (contentDelta) {
        collected += contentDelta;
        options.onTextDelta?.(collected, contentDelta);
      }
      return;
    }
    if (
      parsed.type === "response.output_text.delta" &&
      typeof parsed.delta === "string"
    ) {
      collected += parsed.delta;
      options.onTextDelta?.(collected, parsed.delta);
    }
    if (
      parsed.type === "response.refusal.delta" &&
      typeof parsed.delta === "string"
    ) {
      collected += parsed.delta;
      options.onTextDelta?.(collected, parsed.delta);
    }
    if (parsed.type === "error") {
      throw new AiError(
        (parsed.error && parsed.error.message) || "流式响应出错。",
        0
      );
    }
  };

  while (true) {
    const step = await reader.read();
    if (step.done) break;
    buffer += decoder.decode(step.value, { stream: true });
    let boundary = buffer.indexOf("\n\n");
    while (boundary !== -1) {
      const rawEvent = buffer.slice(0, boundary);
      buffer = buffer.slice(boundary + 2);
      consumeEvent(rawEvent);
      boundary = buffer.indexOf("\n\n");
    }
  }
  buffer += decoder.decode();
  if (buffer.trim()) consumeEvent(buffer);
  return collected;
}

interface SSEEvent {
  choices?: Array<unknown>;
  type?: string;
  delta?: unknown;
  error?: { message?: string };
  [key: string]: unknown;
}

function parseSSEEvent(rawEvent: string): SSEEvent | null {
  const dataLines: string[] = [];
  rawEvent.split(/\r?\n/).forEach((line) => {
    if (line.indexOf("data:") === 0) dataLines.push(line.slice(5).trimStart());
  });
  const payload = dataLines.join("\n").trim();
  if (!payload || payload === "[DONE]") return null;
  try {
    return JSON.parse(payload) as SSEEvent;
  } catch (_) {
    return null;
  }
}

async function readJsonPayload(
  response: Response,
  apiMode: "chat" | "responses"
): Promise<string> {
  const payload = await response.json();
  return apiMode === "chat"
    ? extractChatText(payload)
    : extractResponsesText(payload);
}

// Re-export safePreview for use by Judge / runner when persisting error previews.
export { safePreview };