AgentServer/apps/server/src/providers/codex/adapter.ts

// Codex 适配器：codex app-server（JSON-RPC）→ 统一事件（docs/design/03-providers.md §3，ADR-0003）
import fs from 'node:fs'
import path from 'node:path'
import type { AppSettings, ProviderStatusDto, UsagePayload } from '@agent-server/shared'
import { checkAccess } from '../../services/access-policy'
import { ProviderRunError, type AgentProvider, type RunContext, type RunResult } from '../types'
import { bundledCodexPath, codexHome } from './cli'
import { buildThreadConfig } from './permissions'
import type {
  AskForApproval, CommandExecutionRequestApprovalParams, FileChangeRequestApprovalParams, GetAccountResponse,
  ItemCompletedNotification, ItemStartedNotification, LoginAccountResponse, ModelListResponse,
  PermissionsRequestApprovalParams, ThreadItem, ThreadStartResponse, ThreadTokenUsageUpdatedNotification,
  ToolRequestUserInputParams, TurnCompletedNotification, TurnStartResponse,
} from './protocol/v2'
import { AppServerClient, RpcRequestError } from './rpc'

const MAX_TOOL_OUTPUT = 64 * 1024
const CLIENT_INFO = { name: 'agent_server', title: 'AgentServer', version: '0.1.0' }

interface Logger { info(o: object, m?: string): void; warn(o: object, m?: string): void; error(o: object, m?: string): void }

export interface CodexLoginState {
  loginId: string
  verificationUrl: string
  userCode: string
  status: 'pending' | 'success' | 'failed' | 'cancelled'
  error: string | null
}

export class CodexProvider implements AgentProvider {
  readonly id = 'codex' as const
  private client: AppServerClient | null = null
  private starting: Promise<AppServerClient> | null = null
  private clientKey = ''
  private runs = new Map<string, ThreadRun>()
  private models: string[] = []
  private login: CodexLoginState | null = null
  private version: string | null = null

  constructor(private log: Logger) {}

  knownModels(): string[] {
    return this.models
  }

  private bin(settings: AppSettings) {
    return settings.providers.codex.cliPath
  }

  /** 启动（或复用）app-server；CLI 路径或配置覆盖变化时重启 */
  private async ensure(settings: AppSettings): Promise<AppServerClient> {
    const s = settings.providers.codex
    const key = JSON.stringify([s.cliPath, s.configOverrides])
    if (this.client?.alive && this.clientKey === key) return this.client
    if (this.starting && this.clientKey === key) return this.starting
    this.client?.stop()
    this.clientKey = key
    this.starting = (async () => {
      const home = codexHome()
      fs.mkdirSync(home, { recursive: true, mode: 0o700 })
      const env: NodeJS.ProcessEnv = { ...process.env, CODEX_HOME: home }
      delete env.OPENAI_API_KEY
      delete env.CODEX_API_KEY
      const client = new AppServerClient({ bin: s.cliPath, env, args: s.configOverrides.flatMap(o => ['-c', o]) })
      client.on('notification', (method, params) => this.onNotification(method, params))
      client.onServerRequest = (method, params) => this.onServerRequest(method, params)
      client.on('stderr', line => {
        if (!/WARN|INFO|DEBUG/.test(line)) this.log.warn({ stderr: line.trim().slice(0, 500) }, 'codex stderr')
      })
      client.on('exit', code => {
        this.log.warn({ code }, 'codex app-server 已退出')
        if (this.client === client) this.client = null
        for (const run of this.runs.values()) run.fail(new ProviderRunError('codex_exited', `codex app-server 意外退出（${code}）`))
      })
      const init = await client.start(CLIENT_INFO) as { userAgent?: string }
      // userAgent 形如 "agent_server/0.164.0 (...)"，斜杠后是 codex 版本
      this.version = /\/([\d.]+)/.exec(init.userAgent ?? '')?.[1] ?? null
      this.client = client
      return client
    })()
    try {
      return await this.starting
    } finally {
      this.starting = null
    }
  }

  async run(ctx: RunContext): Promise<RunResult> {
    const client = await this.ensure(ctx.settings)
    fs.mkdirSync(path.join(ctx.workspaceDir, '.agent', 'tmp'), { recursive: true })
    const config = buildThreadConfig({
      access: ctx.access, permissionMode: ctx.permissionMode, settings: ctx.settings,
      codexBin: this.bin(ctx.settings), codexHome: codexHome(),
    })
    // 需审批：沙箱内的命令自动执行，越过沙箱边界时请求审批；自动/只读：从不请求，越界直接失败
    const approvalPolicy: AskForApproval = ctx.permissionMode === 'ask' ? 'on-request' : 'never'
    const threadParams = {
      cwd: ctx.workspaceDir, model: ctx.model, approvalPolicy, config,
      developerInstructions: ctx.systemPromptAppend,
    }

    let threadId = ctx.providerSessionId
    if (threadId) {
      try {
        await client.request('thread/resume', { ...threadParams, threadId, excludeTurns: true })
      } catch (err) {
        ctx.log.warn({ err: (err as Error).message, threadId }, 'codex 会话恢复失败，以新会话继续')
        threadId = null
      }
    }
    if (!threadId) {
      const r = await client.request<ThreadStartResponse>('thread/start', { ...threadParams, serviceName: 'agent-server' })
      threadId = r.thread.id
      ctx.setProviderSessionId(threadId)
    }

    const run = new ThreadRun(ctx, threadId)
    this.runs.set(threadId, run)
    const onAbort = () => {
      if (run.turnId) void client.request('turn/interrupt', { threadId, turnId: run.turnId }).catch(() => {})
      run.fail(new ProviderRunError('interrupted', '已中断'))
    }
    ctx.signal.addEventListener('abort', onAbort, { once: true })
    try {
      const r = await client.request<TurnStartResponse>('turn/start', { threadId, model: ctx.model, input: buildInput(ctx) })
      run.turnId = r.turn.id
      return await run.done
    } finally {
      ctx.signal.removeEventListener('abort', onAbort)
      this.runs.delete(threadId)
    }
  }

  private onNotification(method: string, params: unknown) {
    if (method === 'account/login/completed') {
      const p = params as { loginId: string | null; success: boolean; error: string | null }
      if (this.login && (!p.loginId || p.loginId === this.login.loginId)) {
        Object.assign(this.login, { status: p.success ? 'success' : 'failed', error: p.error })
      }
      return
    }
    const threadId = (params as { threadId?: string } | null)?.threadId
    if (threadId) this.runs.get(threadId)?.onNotification(method, params)
  }

  private async onServerRequest(method: string, params: unknown): Promise<unknown> {
    const threadId = (params as { threadId?: string } | null)?.threadId
    const run = threadId ? this.runs.get(threadId) : undefined
    if (!run) throw new Error(`没有对应的运行：${method}`)
    return run.onServerRequest(method, params)
  }

  // ---- 账户与登录（设备码方式：在 OpenAI 官方页面输入短码完成登录，AgentServer 不接触凭证） ----

  async startLogin(settings: AppSettings): Promise<CodexLoginState> {
    const client = await this.ensure(settings)
    if (this.login?.status === 'pending') await this.cancelLogin(settings).catch(() => {})
    const r = await client.request<LoginAccountResponse>('account/login/start', { type: 'chatgptDeviceCode' })
    if (r.type !== 'chatgptDeviceCode') throw new Error('codex 没有返回设备码')
    this.login = { loginId: r.loginId, verificationUrl: r.verificationUrl, userCode: r.userCode, status: 'pending', error: null }
    return this.login
  }

  loginState(): CodexLoginState | null {
    return this.login
  }

  async cancelLogin(settings: AppSettings) {
    if (!this.login || this.login.status !== 'pending') return
    const client = await this.ensure(settings)
    await client.request('account/login/cancel', { loginId: this.login.loginId }).catch(() => {})
    this.login.status = 'cancelled'
  }

  async logout(settings: AppSettings) {
    const client = await this.ensure(settings)
    await client.request('account/logout', {})
  }

  async status(settings: AppSettings): Promise<ProviderStatusDto> {
    const s = settings.providers.codex
    const base: ProviderStatusDto = {
      id: 'codex', enabled: s.enabled, available: false, loggedIn: null, account: null,
      cliPath: s.cliPath, version: null, models: s.models, defaultModel: s.defaultModel, message: null,
    }
    try {
      const client = await this.ensure(settings)
      const acc = await client.request<GetAccountResponse>('account/read', {}, 30_000)
      const loggedIn = !!acc.account
      let models = s.models
      if (loggedIn) {
        const list = await client.request<ModelListResponse>('model/list', {}, 30_000).catch(() => null)
        if (list) {
          this.models = list.data.filter(m => !m.hidden).map(m => m.model)
          if (!models.length) models = this.models
        }
      }
      const account = acc.account?.type === 'chatgpt'
        ? [acc.account.email, `(${acc.account.planType})`].filter(Boolean).join(' ')
        : acc.account?.type ?? null
      return {
        ...base, version: this.version, loggedIn, account, models,
        defaultModel: s.defaultModel || models[0] || '',
        available: s.enabled && loggedIn,
        message: loggedIn ? (s.enabled ? null : '已登录，尚未在"设置 → 引擎"中启用') : '未登录：点击"登录"按钮，用 ChatGPT 账号完成设备码登录',
      }
    } catch (err) {
      return { ...base, message: `无法启动 codex app-server：${(err as Error).message}` }
    }
  }
}

export function defaultCodexPath() {
  return bundledCodexPath()
}

function buildInput(ctx: RunContext) {
  let text = ctx.input.text
  if (ctx.input.attachments.length) {
    const list = ctx.input.attachments.map(a => `- ${a.path}（${a.mime}）`).join('\n')
    text += `\n\n<attachments>\n用户上传了以下文件，位于工作区中：\n${list}\n</attachments>`
  }
  const images = ctx.input.attachments
    .filter(a => ctx.input.images.some(i => i.name === a.name))
    .map(a => ({ type: 'localImage' as const, path: path.join(ctx.workspaceDir, a.path) }))
  return [{ type: 'text' as const, text, text_elements: [] }, ...images]
}

/** 一次 Run 对应的一个 turn：把通知映射为统一事件，处理审批与提问 */
class ThreadRun {
  turnId: string | null = null
  readonly done: Promise<RunResult>
  private resolve!: (r: RunResult) => void
  private reject!: (e: Error) => void
  private settled = false
  private items = new Map<string, ThreadItem>()
  private denied = new Set<string>()
  private usageBase: UsagePayload | null = null
  private usageLatest: UsagePayload | null = null

  constructor(private ctx: RunContext, private threadId: string) {
    this.done = new Promise((res, rej) => {
      this.resolve = res
      this.reject = rej
    })
  }

  fail(err: Error) {
    if (this.settled) return
    this.settled = true
    this.reject(err)
  }

  onNotification(method: string, params: unknown) {
    const emit = this.ctx.emit
    switch (method) {
      case 'item/started': {
        const { item } = params as ItemStartedNotification
        this.items.set(item.id, item)
        const tool = describeTool(item, this.ctx.workspaceDir)
        if (tool) emit('tool.started', { toolCallId: item.id, ...tool })
        if (item.type === 'commandExecution') this.ctx.audit('command', item.command.slice(0, 500), { itemId: item.id })
        break
      }
      case 'item/agentMessage/delta': {
        const p = params as { itemId: string; delta: string }
        emit('message.delta', { messageId: p.itemId, text: p.delta })
        break
      }
      case 'item/reasoning/summaryTextDelta': {
        const p = params as { itemId: string; delta: string }
        emit('reasoning.delta', { blockId: p.itemId, text: p.delta })
        break
      }
      case 'item/completed': {
        const { item } = params as ItemCompletedNotification
        this.items.set(item.id, item)
        this.onItemCompleted(item)
        break
      }
      case 'thread/tokenUsage/updated': {
        const { tokenUsage } = params as ThreadTokenUsageUpdatedNotification
        const toUsage = (b: typeof tokenUsage.total): UsagePayload => ({
          inputTokens: b.inputTokens - b.cachedInputTokens, outputTokens: b.outputTokens,
          cacheReadTokens: b.cachedInputTokens, cacheWriteTokens: b.cacheWriteInputTokens,
        })
        // 本次 Run 的用量 = 最新累计 −（第一次通知时的累计 − 那一次的增量）
        if (!this.usageBase) {
          const t = toUsage(tokenUsage.total), l = toUsage(tokenUsage.last)
          this.usageBase = {
            inputTokens: t.inputTokens - l.inputTokens, outputTokens: t.outputTokens - l.outputTokens,
            cacheReadTokens: t.cacheReadTokens - l.cacheReadTokens, cacheWriteTokens: t.cacheWriteTokens - l.cacheWriteTokens,
          }
        }
        this.usageLatest = toUsage(tokenUsage.total)
        break
      }
      case 'turn/completed': {
        const { turn } = params as TurnCompletedNotification
        if (this.settled) return
        if (turn.status === 'completed') {
          this.settled = true
          this.resolve({ usage: this.runUsage() })
        } else if (turn.status === 'interrupted') {
          this.fail(new ProviderRunError('interrupted', '已中断'))
        } else {
          this.fail(new ProviderRunError('codex_failed', turn.error?.message ?? '运行失败'))
        }
        break
      }
      case 'error': {
        const p = params as { error: { message: string }; willRetry: boolean }
        this.ctx.log.warn({ error: p.error.message, willRetry: p.willRetry }, 'codex error')
        break
      }
    }
  }

  private runUsage(): UsagePayload | null {
    if (!this.usageLatest || !this.usageBase) return null
    const b = this.usageBase, l = this.usageLatest
    return {
      inputTokens: l.inputTokens - b.inputTokens, outputTokens: l.outputTokens - b.outputTokens,
      cacheReadTokens: l.cacheReadTokens - b.cacheReadTokens, cacheWriteTokens: l.cacheWriteTokens - b.cacheWriteTokens,
    }
  }

  private onItemCompleted(item: ThreadItem) {
    const emit = this.ctx.emit
    const done = (status: 'ok' | 'error' | 'denied', output: string) => emit('tool.completed', {
      toolCallId: item.id, status: this.denied.has(item.id) ? 'denied' : status,
      output: output.slice(0, MAX_TOOL_OUTPUT), truncated: output.length > MAX_TOOL_OUTPUT,
    })
    switch (item.type) {
      case 'agentMessage':
        if (item.text) emit('message.completed', { messageId: item.id, text: item.text })
        break
      case 'reasoning': {
        const text = item.summary.join('\n').trim()
        if (text) emit('reasoning.completed', { blockId: item.id, text })
        break
      }
      case 'commandExecution':
        done(item.status === 'declined' ? 'denied' : item.status === 'completed' && (item.exitCode ?? 0) === 0 ? 'ok' : 'error',
          `${item.aggregatedOutput ?? ''}${item.exitCode ? `\n(exit ${item.exitCode})` : ''}`.trim())
        break
      case 'fileChange':
        done(item.status === 'declined' ? 'denied' : item.status === 'completed' ? 'ok' : 'error',
          item.changes.map(c => `${c.kind} ${c.path}`).join('\n'))
        break
      case 'mcpToolCall':
        done(item.status === 'completed' ? 'ok' : 'error',
          item.error ? JSON.stringify(item.error) : (item.result?.content ?? []).map(c => (c as { text?: string }).text ?? '').join('\n'))
        break
      case 'webSearch':
      case 'imageView':
        done('ok', '')
        break
    }
  }

  async onServerRequest(method: string, params: unknown): Promise<unknown> {
    const ctx = this.ctx
    switch (method) {
      case 'item/commandExecution/requestApproval': {
        const p = params as CommandExecutionRequestApprovalParams
        // 网络访问不放开：Codex 的命令一旦联网就是无限制的（可访问内网）
        if (p.networkApprovalContext) {
          this.denied.add(p.itemId)
          ctx.audit('network_denied', `拒绝命令联网：${p.command ?? ''}`.slice(0, 500), { networkApprovalContext: p.networkApprovalContext })
          return { decision: 'decline' }
        }
        if (ctx.permissionMode === 'readonly') return { decision: 'decline' }
        const command = p.command ?? (this.items.get(p.itemId) as { command?: string } | undefined)?.command ?? ''
        const outcome = await ctx.interact({
          kind: 'approval', tool: 'Bash', title: commandTitle(command) || '执行命令',
          input: { command, cwd: p.cwd }, detail: { command, reason: p.reason ?? undefined },
          options: ['allow_once', 'allow_session', 'deny'],
        })
        ctx.audit('approval', `codex 命令：${outcome.resolution}`, { command })
        if (outcome.resolution === 'allowed') return { decision: outcome.scope === 'session' ? 'acceptForSession' : 'accept' }
        this.denied.add(p.itemId)
        return { decision: outcome.resolution === 'denied' ? 'decline' : 'cancel' }
      }
      case 'item/fileChange/requestApproval': {
        const p = params as FileChangeRequestApprovalParams
        const item = this.items.get(p.itemId)
        const paths = item?.type === 'fileChange' ? item.changes.map(c => path.resolve(ctx.workspaceDir, c.path)) : []
        if (p.grantRoot) paths.push(path.resolve(ctx.workspaceDir, p.grantRoot))
        const blocked = paths.map(f => ({ f, d: checkAccess(ctx.access, f, 'write') })).find(x => !x.d.allowed)
        if (ctx.permissionMode === 'readonly' || blocked || !paths.length) {
          this.denied.add(p.itemId)
          if (blocked && !blocked.d.allowed) ctx.audit('file_access_denied', blocked.d.reason, { tool: 'codex fileChange', paths })
          return { decision: 'decline' }
        }
        const outcome = await ctx.interact({
          kind: 'approval', tool: 'Edit', title: paths.map(f => path.relative(ctx.workspaceDir, f)).join('、'),
          input: { paths }, detail: { path: paths[0], reason: p.reason ?? undefined }, options: ['allow_once', 'allow_session', 'deny'],
        })
        if (outcome.resolution === 'allowed') return { decision: outcome.scope === 'session' ? 'acceptForSession' : 'accept' }
        this.denied.add(p.itemId)
        return { decision: outcome.resolution === 'denied' ? 'decline' : 'cancel' }
      }
      case 'item/permissions/requestApproval': {
        // 申请额外权限：联网一律拒绝；文件路径必须在 Profile 的访问策略之内，然后由人审批
        const p = params as PermissionsRequestApprovalParams
        const reqFs = p.permissions.fileSystem
        const reads = (reqFs?.read ?? []).map(f => path.resolve(p.cwd, f))
        const writes = (reqFs?.write ?? []).map(f => path.resolve(p.cwd, f))
        const outside = [...reads.map(f => checkAccess(ctx.access, f, 'read')), ...writes.map(f => checkAccess(ctx.access, f, 'write'))].find(d => !d.allowed)
        const wantsNetwork = !!p.permissions.network?.enabled
        if (wantsNetwork) ctx.audit('network_denied', '拒绝 codex 申请联网权限', { reason: p.reason })
        if (outside && !outside.allowed) ctx.audit('file_access_denied', outside.reason, { reads, writes })
        if (wantsNetwork || outside || (!reads.length && !writes.length) || ctx.permissionMode !== 'ask') {
          return { permissions: {}, scope: 'turn' }
        }
        const outcome = await ctx.interact({
          kind: 'approval', tool: 'Permissions', title: `申请访问：${[...reads, ...writes].join('、')}`.slice(0, 300),
          input: p.permissions, detail: { reason: p.reason ?? undefined }, options: ['allow_once', 'allow_session', 'deny'],
        })
        if (outcome.resolution !== 'allowed') return { permissions: {}, scope: 'turn' }
        return { permissions: { fileSystem: { read: reads, write: writes } }, scope: outcome.scope === 'session' ? 'session' : 'turn' }
      }
      case 'item/tool/requestUserInput': {
        const p = params as ToolRequestUserInputParams
        const outcome = await ctx.interact({
          kind: 'question',
          questions: p.questions.map(q => ({
            id: q.id, question: q.question, header: q.header, multiSelect: false,
            options: (q.options ?? []).map(o => ({ label: o.label, description: o.description })),
          })),
        })
        const answers: Record<string, { answers: string[] }> = {}
        if (outcome.resolution === 'answered') {
          for (const q of p.questions) {
            const a = outcome.answers[q.id]
            if (a !== undefined) answers[q.id] = { answers: Array.isArray(a) ? a : [a] }
          }
        }
        return { answers }
      }
      case 'mcpServer/elicitation/request':
        return { action: 'decline', content: null, _meta: null }
      default:
        throw new RpcRequestError(method, { code: -32601, message: `AgentServer 不支持：${method}` })
    }
  }
}

/** Codex 把命令包成 `/usr/bin/bash -lc "..."`，显示时去掉外层，只保留第一行 */
export function commandTitle(command: string): string {
  let c = command.trim()
  const m = /^(?:\/usr)?(?:\/bin\/)?(?:ba|z)?sh\s+-l?c\s+([\s\S]*)$/.exec(c)
  if (m) {
    c = m[1]!.trim()
    if ((c.startsWith('"') && c.endsWith('"')) || (c.startsWith("'") && c.endsWith("'"))) c = c.slice(1, -1)
  }
  return c.split('\n')[0]!.slice(0, 200)
}

function describeTool(item: ThreadItem, cwd: string): { name: string; title: string; input: unknown } | null {
  const rel = (p: string) => {
    const r = path.relative(cwd, path.resolve(cwd, p))
    return r.startsWith('..') ? p : r
  }
  switch (item.type) {
    case 'commandExecution':
      return { name: 'Bash', title: commandTitle(item.command), input: { command: item.command, cwd: item.cwd } }
    case 'fileChange':
      return { name: 'Edit', title: item.changes.map(c => rel(c.path)).join('、').slice(0, 200), input: { changes: item.changes.map(c => ({ path: c.path, kind: c.kind, diff: c.diff })) } }
    case 'mcpToolCall':
      return { name: `mcp__${item.server}__${item.tool}`, title: '', input: item.arguments }
    case 'webSearch':
      return { name: 'WebSearch', title: (item as { query?: string }).query ?? '', input: { query: (item as { query?: string }).query } }
    case 'imageView':
      return { name: 'Read', title: rel(item.path), input: { path: item.path } }
    default:
      return null
  }
}