orca/src/main/codex/codex-app-server-connection.ts

import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'
import { buildCodexAppServerExitError } from './codex-app-server-exit-error'
import { isAppServerRecord, parseCodexAppServerJsonLine } from './codex-app-server-jsonl'
import { waitForProcessExitUntil } from './codex-process-exit-deadline'
import {
  CodexAppServerTimeoutError,
  CodexAppServerUnsupportedError,
  isCodexMethodNotFoundError,
  killCodexAppServerProcessTree
} from './codex-app-server-session'
import type {
  CodexAppServerConnection,
  CodexAppServerConnectionHandlers
} from './codex-app-server-connection-types'

export type {
  CodexAppServerConnection,
  CodexAppServerConnectionHandlers,
  CodexAppServerServerRequest
} from './codex-app-server-connection-types'

// Structured chat needs a persistent bidirectional child and per-request deadlines;
// the request-scoped app-server runner cannot carry approvals or streamed turns.

/** Codex answered the call and refused it. Distinct from a timeout or a dead
 *  child, which leave the call unsettled rather than declined. */
export class CodexAppServerRequestError extends Error {
  constructor(
    readonly method: string,
    readonly code: number | null,
    message: string
  ) {
    super(message)
    this.name = 'CodexAppServerRequestError'
  }
}

export function isCodexAppServerRequestError(error: unknown): error is CodexAppServerRequestError {
  return error instanceof Error && error.name === 'CodexAppServerRequestError'
}

export type CodexAppServerLaunch = {
  command: string
  args: string[]
  /** Overlay on the inherited environment — the pinned CODEX_HOME lives here. */
  env?: Record<string, string>
  /** Keys stripped after the overlay, matching `CodexAppServerInvocation`. */
  envToDelete?: readonly string[]
}

const DEFAULT_REQUEST_TIMEOUT_MS = 30_000
const HANDSHAKE_TIMEOUT_MS = 15_000
const GRACEFUL_EXIT_MS = 1_500
const FORCED_EXIT_MS = 1_000
const STDERR_TAIL_MAX_BYTES = 8192
const STDOUT_LINE_MAX_BYTES = 1024 * 1024

type PendingRequest = {
  method: string
  resolve: (result: unknown) => void
  reject: (error: Error) => void
  timer: ReturnType<typeof setTimeout>
}

/**
 * Spawns `codex app-server`, completes the initialize handshake, and returns a
 * connection that stays open until `close()`. Rejects — after reaping the child
 * — when the handshake cannot complete.
 */
export async function openCodexAppServerConnection(
  launch: CodexAppServerLaunch,
  handlers: CodexAppServerConnectionHandlers = {},
  spawnImpl: typeof spawn = spawn
): Promise<CodexAppServerConnection> {
  const childEnv: NodeJS.ProcessEnv = { ...process.env, ...launch.env }
  for (const key of launch.envToDelete ?? []) {
    delete childEnv[key]
  }
  const child = spawnImpl(launch.command, launch.args, {
    env: childEnv,
    stdio: ['pipe', 'pipe', 'pipe'],
    windowsHide: true
  }) as ChildProcessWithoutNullStreams

  const pending = new Map<number, PendingRequest>()
  let stderrTail = ''
  let nextRequestId = 1
  let exited = false
  let exitObserved = false
  let closing = false
  /** First terminal cause, or null while the transport is still usable. Set once:
   *  a child that dies reaches us through several listeners, and the specific
   *  first cause is the one worth reporting. */
  let terminalError: Error | null = null

  const exitPromise = new Promise<void>((resolve) => {
    child.on('exit', () => {
      exited = true
      exitObserved = true
      resolve()
    })
  })

  function buildExitError(cause?: Error): Error {
    return buildCodexAppServerExitError(stderrTail, cause)
  }

  function failPending(error: Error): void {
    for (const waiter of pending.values()) {
      clearTimeout(waiter.timer)
      waiter.reject(error)
    }
    pending.clear()
  }

  /** A death nobody asked for kills every in-flight call AND tells the owner,
   *  which is the only signal the session has that its lease is now worthless.
   *  Once only: an oversized line kills the child and its `close` arrives after,
   *  and a spawn failure arrives as both `error` and `close`. */
  function handleUnexpectedEnd(cause?: Error): void {
    if (terminalError) {
      return
    }
    terminalError = buildExitError(cause)
    failPending(terminalError)
    if (!closing) {
      handlers.onExit?.(terminalError)
    }
  }

  child.on('error', (error) => {
    handleUnexpectedEnd(error)
  })
  // Why: 'close' rather than 'exit' guarantees the stderr tail is complete, so
  // an early death classifies as missing-subcommand instead of transient.
  child.on('close', () => {
    handleUnexpectedEnd()
  })
  child.stderr.setEncoding('utf8').on('data', (chunk: string) => {
    stderrTail = (stderrTail + chunk).slice(-STDERR_TAIL_MAX_BYTES)
  })
  child.stdin.on('error', (error) => {
    // A broken pipe is terminal, not one failed write: every later request can
    // only error or time out, so the session must learn its lease is worthless
    // instead of staying live in front of a child nobody can reach. During a
    // close the reap is already under way and `exited` must stay honest, or
    // `close` would skip the kill it still owes.
    if (closing) {
      failPending(error)
      return
    }
    killCodexAppServerProcessTree(child)
    handleUnexpectedEnd(error)
  })

  function dispatchMessage(message: Record<string, unknown>): void {
    const hasMethod = typeof message.method === 'string'
    const hasId = typeof message.id === 'number' || typeof message.id === 'string'
    if (hasMethod && hasId) {
      handlers.onServerRequest?.({
        id: message.id as number | string,
        method: message.method as string,
        params: message.params
      })
      return
    }
    if (hasMethod) {
      handlers.onNotification?.(message.method as string, message.params)
      return
    }
    if (typeof message.id !== 'number') {
      handlers.onUnhandledFrame?.('frame:unclassified', message)
      return
    }
    const waiter = pending.get(message.id)
    if (!waiter) {
      handlers.onUnhandledFrame?.('response:unmatched', message)
      return
    }
    pending.delete(message.id)
    clearTimeout(waiter.timer)
    const error = message.error
    if (isAppServerRecord(error)) {
      const detail = typeof error.message === 'string' ? error.message : 'unknown error'
      waiter.reject(
        isCodexMethodNotFoundError(error)
          ? new CodexAppServerUnsupportedError(
              `codex app-server does not support ${waiter.method}: ${detail}`
            )
          : new CodexAppServerRequestError(
              waiter.method,
              typeof error.code === 'number' ? error.code : null,
              `codex app-server ${waiter.method} failed: ${detail}`
            )
      )
      return
    }
    waiter.resolve(message.result)
  }

  let stdoutBuffer = ''
  // Why: stream decoding must retain a multibyte character split across pipe
  // chunks, or a non-ASCII turn becomes invalid JSON.
  child.stdout.setEncoding('utf8').on('data', (chunk: string) => {
    stdoutBuffer += chunk
    if (Buffer.byteLength(stdoutBuffer) > STDOUT_LINE_MAX_BYTES) {
      child.stdout.destroy()
      killCodexAppServerProcessTree(child)
      handleUnexpectedEnd(new Error('codex app-server emitted an oversized JSONL line'))
      return
    }
    let newlineIndex: number
    while ((newlineIndex = stdoutBuffer.indexOf('\n')) !== -1) {
      const line = stdoutBuffer.slice(0, newlineIndex).trim()
      stdoutBuffer = stdoutBuffer.slice(newlineIndex + 1)
      if (!line) {
        continue
      }
      const parsed = parseCodexAppServerJsonLine(line)
      if (!parsed) {
        handlers.onUnhandledFrame?.('frame:invalid-json', line)
        continue
      }
      try {
        dispatchMessage(parsed)
      } catch (error) {
        child.stdout.destroy()
        killCodexAppServerProcessTree(child)
        handleUnexpectedEnd(error instanceof Error ? error : new Error(String(error)))
        return
      }
    }
  })

  function sendLine(payload: Record<string, unknown>): void {
    child.stdin.write(`${JSON.stringify(payload)}\n`)
  }

  function notify(method: string, params?: Record<string, unknown>): void {
    if (exited || terminalError) {
      return
    }
    try {
      sendLine(params === undefined ? { method } : { method, params })
    } catch {
      // Fire-and-forget; the next request surfaces a dead child.
    }
  }

  function request(
    method: string,
    params?: Record<string, unknown>,
    options: { timeoutMs?: number } = {}
  ): Promise<unknown> {
    if (closing) {
      return Promise.reject(new Error(`codex app-server connection is closed (${method})`))
    }
    if (terminalError) {
      return Promise.reject(terminalError)
    }
    if (exited) {
      return Promise.reject(buildExitError())
    }
    const id = nextRequestId++
    const timeoutMs = options.timeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS
    return new Promise<unknown>((resolve, reject) => {
      // Why: per request, not per session — a chat session outlives every call,
      // so only the individual call can carry a deadline.
      const timer = setTimeout(() => {
        pending.delete(id)
        reject(new CodexAppServerTimeoutError(`codex app-server ${method} exceeded ${timeoutMs}ms`))
      }, timeoutMs)
      pending.set(id, { method, resolve, reject, timer })
      try {
        sendLine(params === undefined ? { method, id } : { method, id, params })
      } catch (error) {
        pending.delete(id)
        clearTimeout(timer)
        reject(error instanceof Error ? error : new Error(String(error)))
      }
    })
  }

  function writeResponse(payload: Record<string, unknown>): void {
    if (exited || terminalError || child.stdin.destroyed || !child.stdin.writable) {
      return
    }
    try {
      sendLine(payload)
    } catch {
      // The turn that asked is already gone with the child.
    }
  }

  async function close(): Promise<boolean> {
    if (closing) {
      return exitObserved
    }
    closing = true
    try {
      child.stdin.end()
    } catch {
      // Already destroyed; the reap below still runs.
    }
    if (!exited) {
      await waitForProcessExitUntil(exitPromise, GRACEFUL_EXIT_MS)
      if (!exited) {
        killCodexAppServerProcessTree(child)
        await waitForProcessExitUntil(exitPromise, FORCED_EXIT_MS)
      }
    }
    failPending(new Error('codex app-server connection closed'))
    return exitObserved
  }

  const connection: CodexAppServerConnection = {
    get pid() {
      return child.pid
    },
    get closed() {
      return closing || exited || terminalError !== null
    },
    request,
    notify,
    respond: (id, result) => writeResponse({ id, result }),
    respondWithError: (id, code, message) => writeResponse({ id, error: { code, message } }),
    close
  }

  try {
    await request(
      'initialize',
      {
        clientInfo: { name: 'orca_desktop', title: 'Orca', version: '0.0.0' },
        capabilities: {
          experimentalApi: false,
          requestAttestation: false,
          mcpServerOpenaiFormElicitation: false,
          extensions: {}
        }
      },
      { timeoutMs: HANDSHAKE_TIMEOUT_MS }
    )
    notify('initialized')
  } catch (error) {
    await close()
    throw error instanceof CodexAppServerUnsupportedError ||
      error instanceof CodexAppServerTimeoutError
      ? error
      : buildExitError(error instanceof Error ? error : new Error(String(error)))
  }
  return connection
}