// Long-lived `codex app-server` client, one process per workspace.
//
// The app-server is a JSON-RPC 2.0 server over stdio (newline-delimited JSON)
// bundled with the Codex CLI — the same protocol the official VS Code
// extension drives. One process serves every thread of one workspace: threads
// carry their own cwd, but env is process-level, so per-workspace env
// injection (moi's `workspaceEnv`) forces one process per workspace — same
// frozen-at-spawn semantics as the Claude Code subprocess (`cc-session.ts`).
//
// Protocol shapes here are hand-written against the bindings generated by
// `codex app-server generate-ts` at CLI 0.144.5, and read defensively — see
// ./NOTES.md.
import type { McpServer, Model, SessionInfo, StreamEvent } from '@/lib/types'

import {
  type CodexModel,
  type CodexThread,
  type SubagentReplay,
  childThreadToSubagentRecord,
  codexModelToModel,
  codexThreadToEvents,
  codexThreadToSessionInfo,
  collectSubagentActivities,
  selectCodexWorkspacePreview
} from './adapter'
import type { WorkspaceActivityPreview } from '../types'
import { findHarnessExecutable, requireHarnessExecutable } from '../executable'
import { debug } from '../../debug'
import { tapWire } from '../debug'
import { resolveWorkspaceEnv } from '../../workspace-env'

const REQUEST_TIMEOUT_MS = 30_000

type Json = Record<string, unknown>
type NotificationListener = (method: string, params: Json) => void

// Every JSON-RPC frame in either direction lands in the shared wire ring
// (server/harness-debug.ts, scoped by workspacePath), kept OUTSIDE the client
// record so it survives process death. Read via /api/.../harness/debug.

export type CodexProcessInfo = {
  running: boolean
  pid?: number
  binary: string | null
}

export type CodexProcessSnapshot = {
  workspacePath: string
  pid: number
  startedAt: number
}

export function getCodexProcessInfo(workspacePath: string): Promise<CodexProcessInfo> {
  const rec = clients.get(workspacePath)
  const binary = findHarnessExecutable('codex')
  if (!rec) return Promise.resolve({ running: false, binary })
  return rec
    .then(r => ({ running: r.client.isAlive(), pid: r.proc.pid, binary }))
    .catch(() => ({ running: false, binary }))
}

export type CodexClient = {
  rpc: <T>(method: string, params?: Json) => Promise<T>
  onNotification: (l: NotificationListener) => () => void
  isAlive: () => boolean
  workspacePath: string
  // Whether this app-server accepts `turn/start.additionalContext` (the native
  // per-turn context channel). Resolved from the initialize handshake.
  supportsAdditionalContext: boolean
}

// `additionalContext` shipped in codex 0.135.0 (openai/codex#24154, May 2026).
// Older servers have no `deny_unknown_fields` on TurnStartParams, so they
// silently DROP the field instead of erroring — probing is useless. The only
// reliable signal is the version embedded in the initialize response's
// userAgent (e.g. `codex_cli_rs/0.144.5 (…)`); an unparsable userAgent gates
// to false and the session path falls back to appending context to the text.
const ADDITIONAL_CONTEXT_MIN_VERSION = [0, 135, 0] as const

export function codexSupportsAdditionalContext(userAgent: string | undefined): boolean {
  const m = userAgent?.match(/\/(\d+)\.(\d+)\.(\d+)/)
  if (!m) return false
  const v = [Number(m[1]), Number(m[2]), Number(m[3])]
  for (let i = 0; i < 3; i++) {
    if (v[i] !== ADDITIONAL_CONTEXT_MIN_VERSION[i]) return v[i] > ADDITIONAL_CONTEXT_MIN_VERSION[i]
  }
  return true
}

type ClientRecord = {
  client: CodexClient
  proc: ReturnType<typeof Bun.spawn>
}

const clients = new Map<string, Promise<ClientRecord>>() // key: workspacePath
const liveProcesses = new Map<
  string,
  { proc: ReturnType<typeof Bun.spawn>; startedAt: number; isAlive: () => boolean }
>()

export function getCodexProcessSnapshot(): CodexProcessSnapshot[] {
  return [...liveProcesses.entries()]
    .filter(([, process]) => process.isAlive())
    .map(([workspacePath, process]) => ({
      workspacePath,
      pid: process.proc.pid,
      startedAt: process.startedAt
    }))
}

async function startClient(workspacePath: string): Promise<ClientRecord> {
  const bin = requireHarnessExecutable('codex')

  const workspaceEnv = await resolveWorkspaceEnv(workspacePath)
  const startedAt = Date.now()
  const proc = Bun.spawn([bin, 'app-server'], {
    cwd: workspacePath,
    stdin: 'pipe',
    stdout: 'pipe',
    stderr: 'pipe',
    // MOI_AGENT marks the agent's shells as agent-driven for the moi CLI
    // (see agent-caller.ts).
    env: { ...process.env, ...workspaceEnv, MOI_AGENT: '1' }
  })

  let alive = true
  let nextId = 1
  const pending = new Map<
    number,
    { resolve: (v: unknown) => void; reject: (e: Error) => void; timer: Timer }
  >()
  const listeners = new Set<NotificationListener>()

  function send(obj: Json) {
    tapWire(workspacePath, 'send', obj)
    proc.stdin.write(JSON.stringify(obj) + '\n')
    proc.stdin.flush()
  }

  function rpc<T>(method: string, params: Json = {}): Promise<T> {
    if (!alive) return Promise.reject(new Error('codex app-server not running'))
    const id = nextId++
    send({ jsonrpc: '2.0', id, method, params })
    return new Promise<T>((resolve, reject) => {
      const timer = setTimeout(() => {
        pending.delete(id)
        reject(new Error(`codex rpc timeout: ${method}`))
      }, REQUEST_TIMEOUT_MS)
      pending.set(id, { resolve: resolve as (v: unknown) => void, reject, timer })
    })
  }

  function fanout(method: string, params: Json) {
    for (const l of listeners) {
      try {
        l(method, params)
      } catch (err) {
        console.error('[codex] notification listener threw', err)
      }
    }
  }

  // Server→client requests MUST be answered or the turn hangs. We run threads
  // with `approvalPolicy: 'never'`, so approval requests should not occur —
  // accept defensively if one does; reject anything else as unsupported.
  function answerServerRequest(msg: Json) {
    const method = msg.method as string
    if (method.endsWith('requestApproval') || method === 'applyPatchApproval') {
      send({ jsonrpc: '2.0', id: msg.id, result: { decision: 'accept' } })
    } else if (method === 'execCommandApproval') {
      send({ jsonrpc: '2.0', id: msg.id, result: { decision: 'approved' } })
    } else {
      send({
        jsonrpc: '2.0',
        id: msg.id,
        error: { code: -32601, message: `moi does not handle ${method}` }
      })
    }
  }

  function handleLine(line: string) {
    if (!line.trim()) return
    let msg: Json
    try {
      msg = JSON.parse(line) as Json
    } catch {
      return
    }
    tapWire(workspacePath, 'recv', msg)
    if ('id' in msg && 'method' in msg) {
      answerServerRequest(msg)
    } else if ('id' in msg) {
      const p = pending.get(msg.id as number)
      if (!p) return
      pending.delete(msg.id as number)
      clearTimeout(p.timer)
      if ('error' in msg) {
        const e = msg.error as { message?: string } | undefined
        p.reject(new Error(e?.message ?? JSON.stringify(msg.error)))
      } else {
        p.resolve(msg.result)
      }
    } else if (typeof msg.method === 'string') {
      fanout(msg.method, (msg.params ?? {}) as Json)
    }
  }

  async function readLoop() {
    const reader = (proc.stdout as ReadableStream<Uint8Array>).getReader()
    const decoder = new TextDecoder()
    let buf = ''
    try {
      while (true) {
        const { done, value } = await reader.read()
        if (done) break
        buf += decoder.decode(value, { stream: true })
        let nl: number
        while ((nl = buf.indexOf('\n')) >= 0) {
          handleLine(buf.slice(0, nl))
          buf = buf.slice(nl + 1)
        }
      }
    } finally {
      alive = false
      if (clients.get(workspacePath) === recordPromise) clients.delete(workspacePath)
      if (liveProcesses.get(workspacePath)?.proc === proc) liveProcesses.delete(workspacePath)
      for (const [, p] of pending) {
        clearTimeout(p.timer)
        p.reject(new Error('codex app-server exited'))
      }
      pending.clear()
      fanout('__exit', {})
      debug(`codex app-server exited ws=${workspacePath}`)
    }
  }

  async function drainStderr() {
    const reader = (proc.stderr as ReadableStream<Uint8Array>).getReader()
    const decoder = new TextDecoder()
    while (true) {
      const { done, value } = await reader.read()
      if (done) break
      const text = decoder.decode(value).trim()
      if (text) console.error('[codex stderr]', text)
    }
  }

  // recordPromise is referenced by readLoop's cleanup to make sure we only
  // delete our own registry entry (not a replacement spawned after a crash).
  const client: CodexClient = {
    rpc,
    onNotification(l) {
      listeners.add(l)
      return () => listeners.delete(l)
    },
    isAlive: () => alive,
    workspacePath,
    supportsAdditionalContext: false
  }
  const record: ClientRecord = { client, proc }
  const recordPromise = Promise.resolve(record)
  liveProcesses.set(workspacePath, { proc, startedAt, isAlive: client.isAlive })

  void readLoop()
  void drainStderr()

  // experimentalApi opts this connection into experimental fields — required
  // for `turn/start.additionalContext`; servers new enough to gate it reject
  // the field otherwise. The response's userAgent carries the CLI version.
  const init = await rpc<{ userAgent?: string }>('initialize', {
    clientInfo: { name: 'moi', title: 'moi', version: '0.1' },
    capabilities: { experimentalApi: true, requestAttestation: false }
  })
  client.supportsAdditionalContext = codexSupportsAdditionalContext(init?.userAgent)
  send({ jsonrpc: '2.0', method: 'initialized', params: {} })
  debug(
    `codex app-server started ws=${workspacePath} bin=${bin} ua=${init?.userAgent ?? 'unknown'} additionalContext=${client.supportsAdditionalContext}`
  )
  return record
}

export async function getCodexClient(workspacePath: string): Promise<CodexClient> {
  const existing = clients.get(workspacePath)
  if (existing) {
    const rec = await existing
    if (rec.client.isAlive()) return rec.client
    clients.delete(workspacePath)
  }
  const started = startClient(workspacePath)
  clients.set(workspacePath, started)
  try {
    return (await started).client
  } catch (err) {
    clients.delete(workspacePath)
    throw err
  }
}

// Preview reads must not spawn an app-server — home-page cards render for
// every workspace, and starting a codex process per card is disproportionate.
// Returns the workspace's client only if one is already running.
async function peekCodexClient(workspacePath: string): Promise<CodexClient | null> {
  const existing = clients.get(workspacePath)
  if (!existing) return null
  try {
    const rec = await existing
    return rec.client.isAlive() ? rec.client : null
  } catch {
    return null
  }
}

// Kill a workspace's app-server so the next message respawns it with fresh
// env (env is process-level and frozen at spawn). Mirrors the semantics of
// cc-session's restartWorkspaceSessions; in-flight turns are lost.
export function killCodexWorkspace(workspacePath: string): void {
  const rec = clients.get(workspacePath)
  if (!rec) return
  clients.delete(workspacePath)
  void rec.then(r => r.proc.kill()).catch(() => {})
}

// Server shutdown: kill every app-server so no codex process is orphaned.
export function killAllCodexClients(): void {
  for (const [path, rec] of clients) {
    clients.delete(path)
    void rec.then(r => r.proc.kill()).catch(() => {})
  }
}

// ---- discovery (sessions list / models / cold history) -----------------------

// Threads whose recorded cwd is exactly the workspace path, newest first.
export async function getCodexSessions(workspacePath: string): Promise<SessionInfo[]> {
  try {
    const client = await getCodexClient(workspacePath)
    const res = await client.rpc<{ data?: CodexThread[] }>('thread/list', {
      cwd: workspacePath,
      limit: 50
    })
    return (res.data ?? []).map(codexThreadToSessionInfo)
  } catch (err) {
    console.error('[codex] thread/list failed', err)
    return []
  }
}

export async function archiveCodexThread(
  client: Pick<CodexClient, 'rpc'>,
  threadId: string
): Promise<void> {
  await client.rpc('thread/archive', { threadId })
}

export async function interruptCodexTurn(
  client: Pick<CodexClient, 'rpc'>,
  threadId: string,
  turnId: string
): Promise<void> {
  await client.rpc('turn/interrupt', { threadId, turnId })
}

export async function archiveCodexSession(workspacePath: string, threadId: string): Promise<void> {
  await archiveCodexThread(await getCodexClient(workspacePath), threadId)
}

// Home-page card preview. Peek-only: with no live app-server the card simply
// omits the activity fields until the workspace is opened once.
export async function getCodexWorkspacePreview(
  workspacePath: string,
  includeFirstUserMessage: boolean
): Promise<WorkspaceActivityPreview> {
  const client = await peekCodexClient(workspacePath)
  if (!client) return {}
  try {
    const res = await client.rpc<{ data?: CodexThread[] }>('thread/list', {
      cwd: workspacePath,
      limit: 50
    })
    return selectCodexWorkspacePreview(res.data ?? [], includeFirstUserMessage)
  } catch {
    return {}
  }
}

// Account-wide model catalog (identical for every workspace); cached like the
// Claude list in agent.ts, cleared on failure so a later call can retry. Keep
// the raw rows so internal helpers can read isDefault and effort ordering.
// The effective service tier is cwd-scoped and applied after reading the cache.
let codexModelCatalogPromise: Promise<CodexModel[]> | null = null

export function getCodexModelCatalog(workspacePath: string): Promise<CodexModel[]> {
  if (!codexModelCatalogPromise) {
    codexModelCatalogPromise = (async () => {
      const client = await getCodexClient(workspacePath)
      const res = await client.rpc<{ data?: CodexModel[] }>('model/list', {})
      return (res.data ?? []).filter(model => !model.hidden)
    })().catch(err => {
      codexModelCatalogPromise = null
      throw err
    })
  }
  return codexModelCatalogPromise
}

async function getCodexConfiguredServiceTier(
  workspacePath: string
): Promise<string | null | undefined> {
  try {
    const client = await getCodexClient(workspacePath)
    const res = await client.rpc<{ config?: { service_tier?: string | null } }>('config/read', {
      cwd: workspacePath,
      includeLayers: false
    })
    return res.config?.service_tier
  } catch (err) {
    debug(
      `codex config/read failed cwd=${workspacePath}: ${err instanceof Error ? err.message : String(err)}`
    )
    return undefined
  }
}

export async function getCodexModels(workspacePath: string): Promise<Model[]> {
  const [models, configuredServiceTier] = await Promise.all([
    getCodexModelCatalog(workspacePath),
    getCodexConfiguredServiceTier(workspacePath)
  ])
  return models.map(model => codexModelToModel(model, configuredServiceTier))
}

// MCP servers as Codex sees them (from ~/.codex/config.toml), for the
// connectors UI. `authStatus` "notLoggedIn" means an OAuth server awaiting
// login; anything else that returned at all is reachable.
export async function getCodexMcpStatus(workspacePath: string): Promise<McpServer[]> {
  try {
    const client = await getCodexClient(workspacePath)
    const res = await client.rpc<{ data?: { name?: string; authStatus?: string }[] }>(
      'mcpServerStatus/list',
      {}
    )
    return (res.data ?? [])
      .filter((s): s is { name: string; authStatus?: string } => typeof s.name === 'string')
      .map(s => ({
        name: s.name,
        status: s.authStatus === 'notLoggedIn' ? ('needs-auth' as const) : ('connected' as const)
      }))
  } catch (err) {
    console.error('[codex] mcpServerStatus/list failed', err)
    return []
  }
}

// Rebuild the SubagentRecords for a replayed parent thread: read each child
// thread announced by a `subAgentActivity` item and fold its transcript into
// a replay record. A child that can't be read just leaves its bare card.
export async function readSubagentRecords(
  client: CodexClient,
  thread: CodexThread
): Promise<Map<string, SubagentReplay>> {
  const map = new Map<string, SubagentReplay>()
  const parentTurnIds = new Set((thread.turns ?? []).map(t => t.id))
  for (const activity of collectSubagentActivities(thread)) {
    const childId = activity.agentThreadId
    if (!childId) continue
    try {
      const res = await client.rpc<{ thread?: CodexThread }>('thread/read', {
        threadId: childId,
        includeTurns: true
      })
      if (!res.thread) continue
      map.set(childId, {
        toolCallId: activity.id,
        record: childThreadToSubagentRecord(res.thread, activity, parentTurnIds)
      })
    } catch {
      // child thread unreadable (deleted, other cwd) — keep the plain card
    }
  }
  return map
}

// Static history replay for the REST events endpoint (no live subscription).
export async function getCodexThreadEvents(
  workspacePath: string,
  threadId: string
): Promise<StreamEvent[]> {
  try {
    const client = await getCodexClient(workspacePath)
    const res = await client.rpc<{ thread?: CodexThread }>('thread/read', {
      threadId,
      includeTurns: true
    })
    if (!res.thread) return []
    return codexThreadToEvents(res.thread, await readSubagentRecords(client, res.thread))
  } catch {
    return []
  }
}