/**
 * Codex (OpenAI app-server) harness — Phase 2.
 *
 * One long-lived `codex app-server` subprocess per conversation. We talk to
 * it via JSON-RPC 2.0 over stdio — bidirectional: client → server requests,
 * server → client notifications. Tokens come from `~/.codex/auth.json`
 * (codex reads them itself; we don't pass anything).
 *
 * Lifecycle per live conversation:
 *   spawn → initialize → initialized → thread/start → turn/start (per
 *   user message; messages arriving mid-turn are QUEUED and each gets its
 *   own turn — see queueOrStart) → turn/completed → idle
 *   → endConversation → turn/interrupt (if needed) → kill subprocess
 *
 * Lifecycle per one-shot query: same as above, but the subprocess is killed
 * as soon as `turn/completed` arrives.
 *
 * Model strings accept either bare ids (`gpt-5.5`) or `<id>:<effort>` where
 * effort is one of low|medium|high|xhigh — the suffix is split off and
 * passed as `effort` on `turn/start`.
 *
 * Notes on parity with Claude harness:
 *   - System prompt → `developerInstructions` on `thread/start` (ADDITIVE —
 *     `baseInstructions` would override codex's native apply_patch/shell spec)
 *   - Mid-turn input is queued, one turn per message — same one-bot:response-
 *     per-push contract the channel routing FIFO depends on. (`turn/steer`
 *     merges inputs into the in-flight turn and breaks that contract.)
 *   - Sub-agents → not implemented (codex collab tools are still experimental;
 *     the collabAgentToolCall handlers below light up if that ever ships)
 *   - MCP servers → MCP.json translated to `-c mcp_servers.*` spawn flags
 *   - Pre-warm → spawn+initialize cached ahead of time (thread/start is issued
 *     at claim time, so the warm process is reusable for any model/instructions)
 */

import { spawn, type ChildProcessWithoutNullStreams } from 'child_process';
import { createRequire } from 'module';
import readline from 'readline';
import crypto from 'crypto';
import fs from 'fs';
import path from 'path';
import { log } from '../../shared/logger.js';
import { WORKSPACE_DIR } from '../../shared/paths.js';
import type { SavedFile } from '../file-saver.js';
import { getCodexAccessToken } from '../../worker/codex-auth.js';
import { assembleSystemPrompt } from '../../worker/prompts/prompt-assembler.js';
import { mirrorSkillsInto } from './skills.js';
import {
  routeAttachment,
  normalizeImageMediaType,
  approxBase64Bytes,
  buildSavedFilesNote,
  INLINE_TEXT_PER_FILE_CHARS,
  INLINE_TEXT_TOTAL_CHARS,
  MAX_INLINE_IMAGE_BYTES,
} from './attachment-policy.js';
import type { OnAgentMessage, RecentMessage, AgentAttachment, AgentQueryRequest, AgentQueryResult } from './types.js';
export type { RecentMessage, AgentAttachment };

/* ── Constants ─────────────────────────────────────────────────────────── */

const CLIENT_INFO = { name: 'bloby', title: 'Bloby', version: '1' };
const REQUEST_TIMEOUT_MS = 60_000;
const VALID_EFFORTS = new Set(['none', 'minimal', 'low', 'medium', 'high', 'xhigh']);
/**
 * Per-turn IDLE watchdog. `turn/completed` is a NON-guaranteed notification — if
 * the app-server stalls mid-turn without exiting, the RPC `exit` handler never
 * fires and `busy` stays true forever (live: wedges the dashboard + defers
 * backend restarts; one-shot: pins the WhatsApp/scheduler slot since bot:done
 * never arrives). This is an IDLE timeout, reset on every notification for the
 * conversation — a legitimately long turn (deep reasoning, many tool calls)
 * keeps emitting events and is never killed; only true silence trips recovery.
 */
const TURN_WATCHDOG_MS = 5 * 60_000;
/**
 * Hard WALL-CLOCK cap for one-shot turns (pulse/cron, customer WhatsApp) —
 * codex has no maxTurns equivalent, and the idle watchdog above never trips on
 * an actively-emitting runaway turn (tool ping-pong). Claude one-shots abort at
 * 5 minutes; mirror that so a runaway can't pin a customer slot forever.
 */
const ONE_SHOT_MAX_TURN_MS = 5 * 60_000;
/**
 * Micro-batch window for bot:token. Codex streams true per-word deltas — an
 * order of magnitude more WS frames per reply than claude's per-block stream.
 * Coalescing at ~60ms keeps the stream visually smooth while cutting frame
 * volume. Every non-token event flushes first, so ordering (and the
 * bot:response === streamed-text invariant) is preserved.
 */
const TOKEN_FLUSH_MS = 60;

/**
 * Resolve the `codex` binary. We don't trust $PATH because Bloby may be
 * installed globally without `@openai/codex` also globally available. Order:
 *   1. BLOBY_CODEX_BIN env override (advanced users / dev)
 *   2. The `bin` entry from the bundled `@openai/codex` npm package
 *      (this is what `npm install bloby-bot` will pull in)
 *   3. Fall back to `codex` on PATH (works when the user installed codex CLI
 *      separately, e.g. via apt/brew).
 */
let cachedCodexBin: string | null = null;
function resolveCodexBin(): string {
  if (cachedCodexBin) return cachedCodexBin;

  if (process.env.BLOBY_CODEX_BIN) {
    cachedCodexBin = process.env.BLOBY_CODEX_BIN;
    return cachedCodexBin;
  }

  try {
    const requireFromHere = createRequire(import.meta.url);
    const pkgPath = requireFromHere.resolve('@openai/codex/package.json');
    const pkg = JSON.parse(fs.readFileSync(pkgPath, 'utf-8'));
    const binEntry = typeof pkg.bin === 'string' ? pkg.bin : pkg.bin?.codex;
    if (binEntry) {
      const resolved = path.join(path.dirname(pkgPath), binEntry);
      if (fs.existsSync(resolved)) {
        cachedCodexBin = resolved;
        log.info(`[codex] using bundled binary: ${resolved}`);
        return cachedCodexBin;
      }
    }
  } catch {
    // @openai/codex not installed as a dep — fall through.
  }

  cachedCodexBin = 'codex';
  return cachedCodexBin;
}

/* ── Prompt-assembly helpers (duplicated from claude.ts to keep that file
 *    untouched per the project rule) ───────────────────────────────────── */

function readMemoryFile(filename: string): string {
  try {
    const content = fs.readFileSync(path.join(WORKSPACE_DIR, filename), 'utf-8').trim();
    return content || '(empty)';
  } catch {
    return '(empty)';
  }
}

function readMemoryFiles() {
  return {
    myself: readMemoryFile('MYSELF.md'),
    myhuman: readMemoryFile('MYHUMAN.md'),
    memory: readMemoryFile('MEMORY.md'),
    pulse: readMemoryFile('PULSE.json'),
    crons: readMemoryFile('CRONS.json'),
  };
}

function formatConversationHistory(messages: RecentMessage[]): string {
  if (!messages.length) return '';
  return messages.map((m) => `${m.role}: ${m.content}`).join('\n\n');
}

async function assembleBaseInstructions(
  names?: { botName: string; humanName: string },
  recentMessages?: RecentMessage[],
): Promise<string> {
  const memoryFiles = readMemoryFiles();
  const basePrompt = await assembleSystemPrompt(names?.botName, names?.humanName, 'codex');
  let prompt = basePrompt;
  prompt += `\n\n---\n# Your Memory Files\n\n## MYSELF.md\n${memoryFiles.myself}\n\n## MYHUMAN.md\n${memoryFiles.myhuman}\n\n## MEMORY.md\n${memoryFiles.memory}\n\n---\n# Your Config Files\n\n## PULSE.json\n${memoryFiles.pulse}\n\n## CRONS.json\n${memoryFiles.crons}`;

  try {
    const { loadConfig: loadCfg } = await import('../../shared/config.js');
    const cfg = loadCfg();
    const channels = (cfg as any).channels;
    if (channels) {
      prompt += `\n\n---\n# Channel Config\n\`\`\`json\n${JSON.stringify(channels, null, 2)}\n\`\`\``;
    }
  } catch {}

  if (recentMessages?.length) {
    prompt += `\n\n---\n# Recent Conversation\n${formatConversationHistory(recentMessages)}`;
  }

  return prompt;
}

/** Split `gpt-5.5:high` into `{ id: 'gpt-5.5', effort: 'high' }`. */
function parseModelString(model: string): { id: string; effort?: string } {
  const idx = model.lastIndexOf(':');
  if (idx <= 0) return { id: model };
  const candidate = model.slice(idx + 1);
  if (!VALID_EFFORTS.has(candidate)) return { id: model };
  return { id: model.slice(0, idx), effort: candidate };
}

/* ── JSON-RPC client over stdio ────────────────────────────────────────── */

type RpcResult<T = any> = { id: number; result?: T; error?: { code?: number; message: string } };
type RpcNotification = { method: string; params?: any };

/** Distinguishes a JSON-RPC error RESPONSE (server alive, request rejected)
 *  from transport failures (timeout / closed / process exit) — turn/start
 *  error containment keeps the conversation alive only for the former. */
type RpcErrorKind = 'rpc-error' | 'timeout' | 'closed';
function rpcError(message: string, kind: RpcErrorKind): Error {
  const err = new Error(message);
  (err as any).rpcKind = kind;
  return err;
}
function isServerRejection(err: any): boolean {
  return err?.rpcKind === 'rpc-error';
}

interface PendingRequest {
  resolve: (value: any) => void;
  reject: (err: Error) => void;
  timer: NodeJS.Timeout;
}

class CodexRpc {
  private proc: ChildProcessWithoutNullStreams | null = null;
  private pending = new Map<number, PendingRequest>();
  private nextId = 1;
  private notificationHandler: (n: RpcNotification) => void = () => {};
  private closeHandler: (code: number | null, errMsg?: string) => void = () => {};
  private closed = false;
  private stderrBuf = '';

  start(extraArgs: string[] = []): void {
    this.proc = spawn(resolveCodexBin(), ['app-server', ...extraArgs], { stdio: ['pipe', 'pipe', 'pipe'] });
    const rl = readline.createInterface({ input: this.proc.stdout });
    rl.on('line', (line) => this.onLine(line));

    this.proc.stderr.on('data', (chunk) => {
      this.stderrBuf += chunk.toString();
      // Trim if growing unbounded.
      if (this.stderrBuf.length > 16_000) this.stderrBuf = this.stderrBuf.slice(-8_000);
    });

    this.proc.on('exit', (code) => {
      if (this.closed) return;
      this.closed = true;
      const errMsg = `codex app-server exited (code=${code}). Stderr tail:\n${this.stderrBuf.trim().slice(-1000)}`;
      const err = rpcError(errMsg, 'closed');
      for (const p of this.pending.values()) {
        clearTimeout(p.timer);
        p.reject(err);
      }
      this.pending.clear();
      // Pass the composed exit error along — mid-turn there are usually no
      // pending requests, so this is the ONLY way crash detail reaches the
      // conversation (the watchdog would otherwise ghost-fire 5 min later
      // with a misleading "timed out" message).
      this.closeHandler(code, errMsg);
    });

    this.proc.on('error', (err) => {
      if (this.closed) return;
      this.closed = true;
      log.warn(`[codex-rpc] spawn error: ${err.message}`);
      for (const p of this.pending.values()) {
        clearTimeout(p.timer);
        p.reject(rpcError(err.message, 'closed'));
      }
      this.pending.clear();
      this.closeHandler(null, `codex app-server failed to spawn: ${err.message}`);
    });
  }

  onNotification(handler: (n: RpcNotification) => void): void { this.notificationHandler = handler; }
  onClose(handler: (code: number | null, errMsg?: string) => void): void { this.closeHandler = handler; }

  private onLine(line: string): void {
    // close() already rejected everything pending; late stdout (buffered deltas,
    // a turn/completed that landed the same instant as teardown) must not
    // re-enter the event pipeline — a post-teardown turn/completed could drain
    // pendingInputs into a dead rpc and wedge agentQueryActive forever.
    if (this.closed) return;
    if (!line.trim()) return;
    let msg: any;
    try { msg = JSON.parse(line); } catch {
      log.warn(`[codex-rpc] malformed JSON from server: ${line.slice(0, 200)}`);
      return;
    }

    // Server-initiated REQUEST (has both id AND method) — must reply or the
    // server hangs forever. Approval requests get auto-accepted to match our
    // bypass-permissions posture; anything else gets a method-not-found
    // error reply so we never silently stall.
    if (typeof msg.id === 'number' && typeof msg.method === 'string') {
      this.handleServerRequest(msg);
      return;
    }

    // RESPONSE to a request we sent.
    if (typeof msg.id === 'number') {
      const pending = this.pending.get(msg.id);
      if (!pending) return;
      this.pending.delete(msg.id);
      clearTimeout(pending.timer);
      if (msg.error) pending.reject(rpcError(msg.error.message || 'RPC error', 'rpc-error'));
      else pending.resolve(msg.result);
      return;
    }

    // NOTIFICATION (no id).
    if (typeof msg.method === 'string') {
      this.notificationHandler({ method: msg.method, params: msg.params });
    }
  }

  private handleServerRequest(msg: { id: number; method: string; params?: any }): void {
    // Responses are OBJECTS, not bare strings: CommandExecution/FileChange approval
    // responses are `{ decision }` (CommandExecutionApprovalDecision / FileChangeApprovalDecision),
    // and the legacy v1 aliases take `{ decision }` with the ReviewDecision enum.
    // (None of these fire under approvalPolicy:'never' + danger-full-access, but reply
    // correctly so an edge-case request can't stall the turn with a malformed reply.)
    switch (msg.method) {
      case 'item/commandExecution/requestApproval':
      case 'item/fileChange/requestApproval':
        log.info(`[codex-rpc] auto-accepting ${msg.method}`);
        this.respond(msg.id, { decision: 'acceptForSession' });
        return;
      case 'execCommandApproval':
      case 'applyPatchApproval':
        log.info(`[codex-rpc] auto-accepting (legacy) ${msg.method}`);
        this.respond(msg.id, { decision: 'approved_for_session' });
        return;
      // account/chatgptAuthTokens/refresh is only used by client-managed-token
      // clients; Bloby authenticates via chatgpt OAuth and the app-server refreshes
      // ~/.codex/auth.json itself, so this server-request never fires for us. Decline
      // cleanly (a stale-credential edge would surface as a normal turn error instead).
      default:
        log.warn(`[codex-rpc] unhandled server request ${msg.method} — replying -32601`);
        this.respondError(msg.id, -32601, `Method ${msg.method} not implemented by Bloby client`);
    }
  }

  private respond(id: number, result: any): void {
    if (this.closed || !this.proc) return;
    try { this.proc.stdin.write(JSON.stringify({ id, result }) + '\n'); }
    catch (err: any) { log.warn(`[codex-rpc] respond failed: ${err.message}`); }
  }

  private respondError(id: number, code: number, message: string): void {
    if (this.closed || !this.proc) return;
    try { this.proc.stdin.write(JSON.stringify({ id, error: { code, message } }) + '\n'); }
    catch (err: any) { log.warn(`[codex-rpc] respondError failed: ${err.message}`); }
  }

  request<T = any>(method: string, params?: any, timeoutMs = REQUEST_TIMEOUT_MS): Promise<T> {
    if (this.closed || !this.proc) return Promise.reject(rpcError('RPC connection closed', 'closed'));
    const id = this.nextId++;
    return new Promise<T>((resolve, reject) => {
      const timer = setTimeout(() => {
        this.pending.delete(id);
        reject(rpcError(`codex app-server: ${method} timed out after ${timeoutMs}ms`, 'timeout'));
      }, timeoutMs);
      this.pending.set(id, { resolve, reject, timer });
      try {
        this.proc!.stdin.write(JSON.stringify({ method, id, params }) + '\n');
      } catch (err: any) {
        this.pending.delete(id);
        clearTimeout(timer);
        reject(rpcError(err.message, 'closed'));
      }
    });
  }

  notify(method: string, params?: any): void {
    if (this.closed || !this.proc) return;
    try {
      this.proc.stdin.write(JSON.stringify({ method, params }) + '\n');
    } catch (err: any) {
      log.warn(`[codex-rpc] notify ${method} failed: ${err.message}`);
    }
  }

  close(): void {
    if (this.closed) return;
    this.closed = true;
    for (const p of this.pending.values()) {
      clearTimeout(p.timer);
      p.reject(rpcError('RPC connection closed', 'closed'));
    }
    this.pending.clear();
    const proc = this.proc;
    try { proc?.stdin.end(); } catch {}
    try { proc?.kill('SIGTERM'); } catch {}
    // Escalate to SIGKILL if the app-server ignores SIGTERM (no true leak today
    // since SIGTERM reaps it, but a SIGTERM-ignoring build would otherwise survive).
    if (proc) {
      const t = setTimeout(() => { try { proc.kill('SIGKILL'); } catch {} }, 2000);
      if (typeof t.unref === 'function') t.unref();
      proc.once('exit', () => clearTimeout(t));
    }
    this.proc = null;
  }
}

/* ── Per-conversation state ────────────────────────────────────────────── */

interface QueuedInput {
  content: string;
  savedFiles?: SavedFile[];
  attachments?: AgentAttachment[];
}

interface CodexConversation {
  id: string;
  rpc: CodexRpc;
  threadId: string;
  effort?: string;
  /** Original model string (with effort suffix) + names — kept for the post-teardown re-warm. */
  model: string;
  names?: { botName: string; humanName: string };
  onMessage: OnAgentMessage;
  /** Currently in-flight turn id (set on `turn/started`, cleared on `turn/completed`). */
  currentTurnId: string | null;
  /** itemId of the agentMessage currently streaming — used to insert a paragraph
   *  break when a turn emits multiple separate agentMessage items. */
  currentMsgItemId: string | null;
  /** Streaming text accumulator for the current turn's agentMessage items. */
  fullText: string;
  /** Per-itemId delta accumulation for the current turn. item/completed is the
   *  authoritative final text per the docs — delta concatenation is not guaranteed
   *  to match it, and a later agentMessage item may complete with NO deltas at all.
   *  This map lets item/completed emit exactly the missing remainder per item. */
  itemTexts: Map<string, string>;
  /** Tools/items used during the current turn, for the bot:turn-complete payload. */
  usedFileTools: boolean;
  /** Messages pushed while a turn is active. Each is drained into its OWN turn on
   *  turn/completed — one bot:response per push, mirroring claude's input queue.
   *  (turn/steer would merge them into the in-flight turn and desync the channel
   *  routing FIFO, which enqueues one target per push.) */
  pendingInputs: QueuedInput[];
  /** True while a turn is being processed. */
  busy: boolean;
  /** True for one-shot queries — the conversation ends after the first turn completes. */
  oneShot: boolean;
  /** True once bot:error fired for the current turn — dedups the double-emit the
   *  app-server produces on hard failures (a non-retry `error` notification followed
   *  by `turn/completed {status:'failed'}` with the same message; live-verified). */
  errorEmitted: boolean;
  /** True once bot:done fired for this one-shot — teardown uses it to guarantee
   *  exactly one bot:done on EVERY terminal path (slot-freeing consumers rely on it). */
  doneEmitted: boolean;
  /** Non-retry `error` notification stashed while a turn is active; surfaced by
   *  turn/completed {failed} (its TurnError is authoritative, the stash is fallback). */
  stashedError: { message: string; info: any } | null;
  /**
   * Latest context occupancy from `thread/tokenUsage/updated` (codex does NOT
   * report usage on `turn/completed` — Turn has no usage field). Emitted on
   * `bot:turn-complete` so the orchestrator's proactive recycling can fire.
   */
  lastContextTokens: number;
  lastContextWindow: number;
  /** Active per-turn watchdog timer (see TURN_WATCHDOG_MS). */
  turnWatchdog: NodeJS.Timeout | null;
  /** Hard wall-clock cap for one-shot turns (see ONE_SHOT_MAX_TURN_MS). Non-resetting. */
  hardTurnTimer: NodeJS.Timeout | null;
  /** bot:token micro-batch buffer (see TOKEN_FLUSH_MS). */
  tokenBuf: string;
  tokenFlushTimer: NodeJS.Timeout | null;
}

const conversations = new Map<string, CodexConversation>();

/** One-shot queries in their init window (token refresh + spawn + initialize +
 *  thread/start) — not yet in `conversations`, so anyConversationBusy() can't see
 *  them. Without this, a queued self-update / backend restart can fire mid-launch
 *  and kill the just-spawning query. Mirrors claude's activeQueries registration. */
const inFlightOneShots = new Set<string>();

/* ── Event emission helpers ────────────────────────────────────────────── */

function flushTokens(conv: CodexConversation): void {
  if (conv.tokenFlushTimer) {
    clearTimeout(conv.tokenFlushTimer);
    conv.tokenFlushTimer = null;
  }
  if (conv.tokenBuf) {
    const token = conv.tokenBuf;
    conv.tokenBuf = '';
    conv.onMessage('bot:token', { conversationId: conv.id, token });
  }
}

/** Drop buffered tokens WITHOUT emitting. Teardown must discard, not flush:
 *  on user-stop the frontend has already cleared its stream state, and a stray
 *  trailing bot:token would re-open streaming with no bot:idle ever coming —
 *  then corrupt the next turn's committedTextLength accounting. */
function discardTokens(conv: CodexConversation): void {
  if (conv.tokenFlushTimer) {
    clearTimeout(conv.tokenFlushTimer);
    conv.tokenFlushTimer = null;
  }
  conv.tokenBuf = '';
}

function emitToken(conv: CodexConversation, token: string): void {
  conv.tokenBuf += token;
  if (!conv.tokenFlushTimer) {
    conv.tokenFlushTimer = setTimeout(() => {
      conv.tokenFlushTimer = null;
      flushTokens(conv);
    }, TOKEN_FLUSH_MS);
  }
}

/** Emit any non-token event, flushing buffered tokens first so ordering is
 *  preserved (bot:tool commits the stream bubble; bot:response must equal the
 *  streamed text the frontend already saw). */
function emitEvent(conv: CodexConversation, type: string, data: any): void {
  flushTokens(conv);
  conv.onMessage(type, data);
}

/** Kinds we can branch on from CodexErrorInfo (string variant or `{ kind: {...} }`). */
function errorInfoKind(info: any): string | undefined {
  if (typeof info === 'string') return info;
  if (info && typeof info === 'object') return Object.keys(info)[0];
  return undefined;
}

/**
 * Map codex's structured error onto a message that gives BLOBY's remedy.
 * The raw upstream text actively misleads ("Please log out and sign in again"
 * points at the codex CLI, not the dashboard re-auth flow).
 */
function humanizeCodexError(message?: string, info?: any, additionalDetails?: string | null): string {
  const raw = message || 'Codex error';
  switch (errorInfoKind(info)) {
    case 'unauthorized':
      return 'Codex session expired or unauthorized. Re-authenticate from the dashboard.';
    case 'usageLimitExceeded':
      return `Codex usage limit reached — ${raw}`;
    case 'contextWindowExceeded':
      return 'The conversation outgrew the model context window. Send your message again to continue in a fresh session.';
    case 'serverOverloaded':
      return 'OpenAI servers are overloaded right now — try again in a moment.';
    case 'httpConnectionFailed':
    case 'responseStreamConnectionFailed':
    case 'responseStreamDisconnected': {
      const status = (Object.values(info || {})[0] as any)?.httpStatusCode;
      return `Connection to OpenAI failed${status ? ` (HTTP ${status})` : ''} — try again in a moment.`;
    }
    default:
      return additionalDetails ? `${raw}\n${additionalDetails}` : raw;
  }
}

/** Emit bot:error exactly once per turn (see errorEmitted). */
function emitError(conv: CodexConversation, message?: string, info?: any, additionalDetails?: string | null): void {
  const friendly = humanizeCodexError(message, info, additionalDetails);
  if (conv.errorEmitted) {
    log.info(`[codex] suppressing duplicate bot:error for conv=${conv.id}: ${friendly.slice(0, 120)}`);
    return;
  }
  conv.errorEmitted = true;
  emitEvent(conv, 'bot:error', { conversationId: conv.id, error: friendly });
}

/** Emit bot:done exactly once per one-shot (see doneEmitted). */
function emitDone(conv: CodexConversation): void {
  if (conv.doneEmitted) return;
  conv.doneEmitted = true;
  emitEvent(conv, 'bot:done', { conversationId: conv.id, usedFileTools: conv.usedFileTools });
}

/* ── Input building ────────────────────────────────────────────────────── */

/**
 * Build codex `UserInput` blocks from the user text + saved files + raw
 * attachments. Routing is delegated to the shared attachment-policy so codex
 * stays byte-for-byte consistent with the Claude/PI harnesses. Codex's UserInput
 * has NO native document type (verified against 0.138), so canNativeDocument is
 * FALSE: PDFs/binaries become a disk-pointer note and the agent opens them with
 * its file tools.
 *
 * Block order is MEDIA-first then TEXT (matching Claude/PI): the inline-text
 * notes and the saved-files pointer are folded into the trailing text block.
 */
function buildUserInput(text: string, savedFiles?: SavedFile[], attachments?: AgentAttachment[]): Array<Record<string, any>> {
  const input: Array<Record<string, any>> = [];

  // Codex understands `localImage` (path on disk) — Bloby's file-saver already
  // wrote image attachments to disk, so we point at the absolute path. Track a
  // per-name COUNT (not presence): WhatsApp multi-image pushes share one
  // attachment name and each saved file covers exactly one of them.
  const savedImageCounts = new Map<string, number>();
  if (savedFiles?.length) {
    for (const f of savedFiles) {
      if (f.type === 'image') {
        input.push({ type: 'localImage', path: f.absPath });
        savedImageCounts.set(f.name, (savedImageCounts.get(f.name) || 0) + 1);
      }
    }
  }

  // Route every attachment through the shared policy. Inline-text notes are
  // accumulated into `inlineNotes` (appended to the trailing text block);
  // images become data-URL blocks (with the localImage path already covering
  // disk-saved copies); everything else falls back to the saved-files pointer.
  let promptText = text || '(attached files)';
  const inlineNotes: string[] = [];
  let inlineBudget = INLINE_TEXT_TOTAL_CHARS;
  if (attachments?.length) {
    for (const att of attachments) {
      switch (routeAttachment(att, { canNativeDocument: false })) {
        case 'image': {
          if (!att.data) break;
          // Skip data-URL inlining when a disk copy exists (localImage already
          // points codex at it) or when the payload is too big to resend on
          // every stateless turn — the saved-files pointer covers it instead.
          const remaining = savedImageCounts.get(att.name) || 0;
          if (remaining > 0) {
            savedImageCounts.set(att.name, remaining - 1);
            break;
          }
          if (approxBase64Bytes(att.data) > MAX_INLINE_IMAGE_BYTES) break;
          const mediaType = normalizeImageMediaType(att.mediaType);
          input.push({ type: 'image', url: `data:${mediaType};base64,${att.data}` });
          break;
        }
        case 'inline-text': {
          if (!att.data) break;
          try {
            let decoded = Buffer.from(att.data, 'base64').toString('utf-8');
            if (decoded.length > INLINE_TEXT_PER_FILE_CHARS) decoded = decoded.slice(0, INLINE_TEXT_PER_FILE_CHARS);
            if (decoded.length > inlineBudget) decoded = decoded.slice(0, inlineBudget);
            if (!decoded.length) break;
            inlineBudget -= decoded.length;
            inlineNotes.push(`\n\n[Attached file content: ${att.name}]\n\`\`\`\n${decoded}\n\`\`\``);
          } catch {}
          break;
        }
        // 'native-document' cannot occur (canNativeDocument:false); it and
        // 'reference-only' both rely on the saved-files disk pointer below.
        default:
          break;
      }
    }
  }

  for (const note of inlineNotes) promptText += note;

  const savedNote = buildSavedFilesNote(savedFiles || []);
  if (savedNote) promptText += `\n\n${savedNote}`;

  input.push({ type: 'text', text: promptText });

  return input;
}

/* ── Turn lifecycle ────────────────────────────────────────────────────── */

function clearTurnWatchdog(conv: CodexConversation): void {
  if (conv.turnWatchdog) {
    clearTimeout(conv.turnWatchdog);
    conv.turnWatchdog = null;
  }
}

function clearHardTurnTimer(conv: CodexConversation): void {
  if (conv.hardTurnTimer) {
    clearTimeout(conv.hardTurnTimer);
    conv.hardTurnTimer = null;
  }
}

/**
 * Arm the per-turn watchdog. On fire, unstick the conversation the same way a
 * real `turn/completed` would (so the dashboard, `anyConversationBusy`, and the
 * channel slot all release), then tear the conversation down — the next message
 * cold-starts a fresh thread.
 */
function armTurnWatchdog(conv: CodexConversation): void {
  clearTurnWatchdog(conv);
  conv.turnWatchdog = setTimeout(() => {
    conv.turnWatchdog = null;
    log.warn(`[codex] turn watchdog fired (${TURN_WATCHDOG_MS}ms) — conv=${conv.id}; unsticking + tearing down`);
    conv.busy = false;
    conv.currentTurnId = null;
    // Prefer a stashed non-retry error: if the server wedged right after
    // emitting it (the turn/completed that normally surfaces it never came),
    // the stash carries the real cause + the M4 remedy mapping.
    const stash = conv.stashedError;
    emitError(conv, stash?.message || 'Codex turn timed out — no response from app-server.', stash?.info);
    if (conv.oneShot) {
      emitDone(conv);
    } else {
      emitEvent(conv, 'bot:turn-complete', {
        conversationId: conv.id,
        usedFileTools: conv.usedFileTools,
        contextTokens: conv.lastContextTokens || 0,
        contextWindow: conv.lastContextWindow || 0,
        idle: true,
      });
    }
    teardownConversation(conv.id);
  }, TURN_WATCHDOG_MS);
}

/** Hard wall-clock cap for one-shot turns — see ONE_SHOT_MAX_TURN_MS. */
function armHardTurnTimer(conv: CodexConversation): void {
  if (!conv.oneShot || conv.hardTurnTimer) return;
  conv.hardTurnTimer = setTimeout(() => {
    conv.hardTurnTimer = null;
    log.warn(`[codex] one-shot hard cap fired (${ONE_SHOT_MAX_TURN_MS}ms) — conv=${conv.id}; interrupting`);
    if (conv.currentTurnId) {
      void conv.rpc.request('turn/interrupt', { threadId: conv.threadId, turnId: conv.currentTurnId }).catch(() => {});
    }
    conv.busy = false;
    conv.currentTurnId = null;
    emitError(conv, `Codex query hit the ${Math.round(ONE_SHOT_MAX_TURN_MS / 60_000)}-minute limit and was stopped.`);
    teardownConversation(conv.id); // emits the guaranteed bot:done
  }, ONE_SHOT_MAX_TURN_MS);
}

function resetTurnState(conv: CodexConversation): void {
  conv.fullText = '';
  conv.currentMsgItemId = null;
  conv.itemTexts = new Map();
  conv.usedFileTools = false;
  conv.errorEmitted = false;
  conv.stashedError = null;
}

async function startTurn(conv: CodexConversation, content: string, savedFiles?: SavedFile[], attachments?: AgentAttachment[]): Promise<void> {
  const input = buildUserInput(content, savedFiles, attachments);
  conv.busy = true;
  resetTurnState(conv);
  emitEvent(conv, 'bot:typing', { conversationId: conv.id });
  armTurnWatchdog(conv);
  armHardTurnTimer(conv);
  try {
    const params: Record<string, any> = { threadId: conv.threadId, input };
    if (conv.effort) params.effort = conv.effort;
    // turn/start resolves immediately with { turn }; seize the id now so an
    // interrupt arriving before the turn/started notification still has a target.
    const res = await conv.rpc.request<{ turn?: { id?: string } }>('turn/start', params);
    if (res?.turn?.id) conv.currentTurnId = res.turn.id;
  } catch (err: any) {
    clearTurnWatchdog(conv);
    conv.busy = false;
    conv.currentTurnId = null;
    emitError(conv, `turn/start failed: ${err.message}`);
    // turn/start produced no turn, so no turn/completed will arrive to clear the
    // supervisor's agentQueryActive (set on bot:typing above).
    //
    // - One-shots and transport failures (timeout / process exit): tear down. A
    //   60s timeout most plausibly means a hung app-server — keeping the conv
    //   would just hang the next message too. teardown emits bot:conversation-
    //   ended (clears agentQueryActive) and, for one-shots, the guaranteed bot:done.
    // - A fast JSON-RPC REJECTION on a live conv means the server is alive (bad
    //   effort value, transient thread error): keep the thread — its server-side
    //   context (files read, compacted history) survives, matching claude's
    //   per-turn error containment. bot:turn-complete unsticks the supervisor.
    if (conv.oneShot || !isServerRejection(err)) {
      teardownConversation(conv.id);
    } else {
      emitEvent(conv, 'bot:turn-complete', {
        conversationId: conv.id,
        usedFileTools: false,
        contextTokens: conv.lastContextTokens || 0,
        contextWindow: conv.lastContextWindow || 0,
        idle: conv.pendingInputs.length === 0,
      });
      // Keep draining queued messages — each failure surfaces its own error,
      // and the finite queue guarantees termination.
      const next = conv.pendingInputs.shift();
      if (next !== undefined) void startTurn(conv, next.content, next.savedFiles, next.attachments);
    }
  }
}

/**
 * Queue-or-start: the codex side of claude's async input queue. While a turn
 * is active, pushes are queued and drained ONE TURN PER MESSAGE on
 * turn/completed — preserving the one-bot:response-per-push contract that
 * channels/manager.ts's routing FIFO depends on (it enqueues exactly one
 * routing target per push and consumes one per bot:response).
 */
function queueOrStart(conv: CodexConversation, content: string, savedFiles?: SavedFile[], attachments?: AgentAttachment[]): void {
  if (conv.busy || conv.currentTurnId) {
    conv.pendingInputs.push({ content, savedFiles, attachments });
    // Mirror claude's pushMessage: every accepted push shows typing immediately.
    emitEvent(conv, 'bot:typing', { conversationId: conv.id });
    return;
  }
  void startTurn(conv, content, savedFiles, attachments);
}

/* ── Notification handling ─────────────────────────────────────────────── */

function handleNotification(conv: CodexConversation, n: { method: string; params?: any }): void {
  const p = n.params || {};
  // Any notification for this conv proves the app-server is alive and working —
  // reset the idle watchdog so a long-but-active turn isn't torn down.
  if (conv.turnWatchdog) armTurnWatchdog(conv);
  switch (n.method) {
    case 'turn/started': {
      conv.currentTurnId = p.turn?.id || null;
      resetTurnState(conv);
      break;
    }

    case 'item/agentMessage/delta': {
      const delta: string = p.delta || '';
      if (!delta) break;
      // A turn can emit multiple agentMessage items (commentary then final_answer).
      // On a new itemId, insert a paragraph break so they don't run together (mirrors claude.ts).
      if (p.itemId && conv.currentMsgItemId && p.itemId !== conv.currentMsgItemId && conv.fullText && !conv.fullText.endsWith('\n')) {
        conv.fullText += '\n\n';
        emitToken(conv, '\n\n');
      }
      if (p.itemId) {
        conv.currentMsgItemId = p.itemId;
        conv.itemTexts.set(p.itemId, (conv.itemTexts.get(p.itemId) || '') + delta);
      }
      conv.fullText += delta;
      emitToken(conv, delta);
      break;
    }

    case 'item/started': {
      const item = p.item || {};
      // Surface tool-like items so the dashboard can show activity. Names use
      // claude's tool vocabulary (Bash/Edit/WebSearch/mcp__server__tool) so one
      // UI label map serves both harnesses.
      switch (item.type) {
        case 'commandExecution':
          emitEvent(conv, 'bot:tool', {
            conversationId: conv.id,
            name: 'Bash',
            input: { command: item.command || item.commandLine || '' },
          });
          break;
        case 'mcpToolCall':
          // ThreadItem.mcpToolCall fields are `server` + `tool` (no toolName/name/input).
          emitEvent(conv, 'bot:tool', {
            conversationId: conv.id,
            name: item.tool ? (item.server ? `mcp__${item.server}__${item.tool}` : item.tool) : 'mcp_tool',
            input: item.arguments || {},
          });
          break;
        case 'fileChange':
          conv.usedFileTools = true;
          emitEvent(conv, 'bot:tool', {
            conversationId: conv.id,
            name: 'Edit',
            input: { changes: (item.changes || []).map((c: any) => c.path).filter(Boolean) },
          });
          break;
        case 'webSearch':
          emitEvent(conv, 'bot:tool', {
            conversationId: conv.id,
            name: 'WebSearch',
            input: { query: item.query || '' },
          });
          break;
        case 'reasoning':
          // Activity signal during the (often long) thinking phase — without it
          // the chat shows bare typing dots with zero events for the entire
          // stretch and turns feel hung. The chat hooks dedup repeated
          // same-name running tools, and channel chunk-flushes no-op on an
          // empty buffer, so this is purely additive. Reasoning TEXT is never
          // emitted as bot:token — that would corrupt the bot:response ===
          // streamed-text contract and leak reasoning to channels.
          emitEvent(conv, 'bot:tool', {
            conversationId: conv.id,
            name: 'thinking',
            status: 'running',
          });
          break;
        case 'collabAgentToolCall':
          // Codex's collaborating sub-agents (experimental, not enabled by our
          // initialize) → Bloby's sub-agent UX, if it ever lights up.
          if (item.tool === 'spawnAgent') {
            emitEvent(conv, 'bot:task-created', {
              conversationId: conv.id,
              taskId: item.id,
              description: item.prompt || 'sub-agent',
              type: 'collab',
            });
          }
          break;
        // userMessage / agentMessage — no tool-style event.
      }
      break;
    }

    case 'turn/plan/updated': {
      // Codex's native planning (its TodoWrite equivalent). Emit it in claude's
      // TodoWrite shape so plans drive the same activity label + bubble/chunk
      // boundaries — without this codex feels opaque exactly on the big agentic
      // asks where claude feels alive. Statuses are camelCase in 0.138
      // (pending | inProgress | completed) — normalize to claude's snake_case.
      const steps = Array.isArray(p.plan) ? p.plan : [];
      emitEvent(conv, 'bot:tool', {
        conversationId: conv.id,
        name: 'TodoWrite',
        input: {
          todos: steps.map((s: any) => ({
            content: s.step || '',
            status: s.status === 'inProgress' ? 'in_progress' : (s.status || 'pending'),
          })),
          ...(p.explanation ? { explanation: p.explanation } : {}),
        },
      });
      break;
    }

    case 'item/completed': {
      const item = p.item || {};
      if (item.type === 'fileChange') conv.usedFileTools = true;
      // item/completed carries the AUTHORITATIVE final text for an agentMessage.
      // Reconcile per-itemId: an item that never streamed deltas (or whose final
      // text extends beyond the concatenated deltas) gets the remainder emitted
      // here — previously a turn-level `!conv.fullText` gate silently dropped any
      // non-delta'd message once ANYTHING had streamed.
      if (item.type === 'agentMessage') {
        const finalText: string = (typeof item.text === 'string' && item.text)
          ? item.text
          : ((item.content || []).map((c: any) => c.text || '').join(''));
        if (finalText) {
          const itemId: string | undefined = item.id;
          const streamed = itemId ? (conv.itemTexts.get(itemId) || '') : conv.fullText;
          let remainder = '';
          if (!streamed) {
            remainder = finalText;
          } else if (finalText.length > streamed.length && finalText.startsWith(streamed)) {
            remainder = finalText.slice(streamed.length);
          }
          if (remainder) {
            // Same paragraph-break rule as the delta path for a fresh item.
            if (!streamed && conv.fullText && !conv.fullText.endsWith('\n')) {
              conv.fullText += '\n\n';
              emitToken(conv, '\n\n');
            }
            conv.fullText += remainder;
            emitToken(conv, remainder);
            if (itemId) {
              conv.itemTexts.set(itemId, finalText);
              conv.currentMsgItemId = itemId;
            }
          }
        }
      }
      if (item.type === 'collabAgentToolCall' && item.tool === 'spawnAgent') {
        emitEvent(conv, 'bot:task-done', {
          conversationId: conv.id,
          taskId: item.id,
          status: item.status,
          summary: item.prompt || '',
        });
      }
      break;
    }

    case 'thread/tokenUsage/updated': {
      // Codex's only token-usage signal. ThreadTokenUsage = { total, last, modelContextWindow };
      // `last.inputTokens` is the current prompt occupancy — the right basis for
      // the recycle compare in supervisor/index.ts (fraction*window, not lifetime).
      // NB: codex's cachedInputTokens is a SUBSET of inputTokens (unlike
      // Anthropic's disjoint cache counters) — summing them double-counts the
      // cache and recycles sessions at ~35% real occupancy.
      const tu = p.tokenUsage || {};
      const last = tu.last || {};
      conv.lastContextTokens = last.inputTokens || 0;
      if (typeof tu.modelContextWindow === 'number' && tu.modelContextWindow > 0) {
        conv.lastContextWindow = tu.modelContextWindow;
      }
      break;
    }

    case 'turn/completed': {
      const status: string = p.turn?.status || 'completed';
      const turnError = p.turn?.error;

      clearTurnWatchdog(conv);
      clearHardTurnTimer(conv);
      conv.currentTurnId = null;
      conv.busy = false;

      let failureKind: string | undefined;
      if (status === 'failed') {
        // Preserve streamed partials exactly like claude: if text already reached
        // the user, commit it as the reply (the frontend's bot:error handler
        // ERASES the uncommitted stream buffer); surface bot:error only when the
        // turn died before producing anything.
        const message = turnError?.message || conv.stashedError?.message || 'Codex turn failed.';
        const info = turnError?.codexErrorInfo ?? conv.stashedError?.info;
        failureKind = errorInfoKind(info);
        if (conv.fullText) {
          log.warn(`[codex] turn failed after streaming ${conv.fullText.length} chars (${message.slice(0, 160)}) — preserving partial as the reply`);
          emitEvent(conv, 'bot:response', { conversationId: conv.id, content: conv.fullText });
        } else {
          emitError(conv, message, info, turnError?.additionalDetails);
        }
      } else if (status === 'interrupted') {
        // Interrupted turns carry no final answer — stay silent.
      } else if (conv.fullText) {
        emitEvent(conv, 'bot:response', { conversationId: conv.id, content: conv.fullText });
      }
      conv.stashedError = null;

      if (conv.oneShot) {
        emitDone(conv);
        teardownConversation(conv.id);
      } else {
        // Context-size signal for the orchestrator's proactive session recycling,
        // sourced from the cached `thread/tokenUsage/updated` values above. 0 if codex
        // never sent one this turn → falls back to codex's own in-thread auto-compaction.
        const idle = conv.pendingInputs.length === 0;
        emitEvent(conv, 'bot:turn-complete', {
          conversationId: conv.id,
          usedFileTools: conv.usedFileTools,
          contextTokens: conv.lastContextTokens || 0,
          contextWindow: conv.lastContextWindow || 0,
          idle,
        });

        // An unauthorized/context-wall thread can't run further turns — tear it
        // down so the NEXT message pre-flights cleanly (friendly dashboard
        // message / fresh thread) instead of repeating the same failure forever.
        if (failureKind === 'unauthorized' || failureKind === 'contextWindowExceeded') {
          // Every queued message got bot:typing on push — give each a terminal
          // event too, or it vanishes without feedback. (Direct emit: the
          // per-turn errorEmitted dedup must not swallow these.)
          const reason = failureKind === 'unauthorized' ? 'authentication expired' : 'context window full';
          for (const dropped of conv.pendingInputs.splice(0)) {
            const snippet = dropped.content.slice(0, 60) + (dropped.content.length > 60 ? '…' : '');
            emitEvent(conv, 'bot:error', {
              conversationId: conv.id,
              error: `Codex session ended (${reason}) before your message "${snippet}" could run — please resend it.`,
            });
          }
          log.warn(`[codex] tearing down conv=${conv.id} after ${failureKind} so the next message starts clean`);
          teardownConversation(conv.id);
          break;
        }

        // Drain any messages that were queued while we were busy — each gets
        // its own turn (and thus its own bot:response).
        const next = conv.pendingInputs.shift();
        if (next !== undefined) void startTurn(conv, next.content, next.savedFiles, next.attachments);
      }
      break;
    }

    case 'error': {
      // ErrorNotification carries willRetry — codex will retry transient errors
      // itself; don't surface those as a hard bot:error before the retry lands.
      if (p.willRetry) {
        log.info(`[codex] transient error (will retry): ${p.error?.message || 'unknown'}`);
        break;
      }
      // Hard failures emit BOTH this notification AND turn/completed{failed}
      // with the same TurnError (live-verified on 0.138.0). While a turn is
      // active, stash it and let turn/completed be the single surface — its
      // TurnError is authoritative per the protocol. Outside a turn there is
      // no turn/completed coming, so emit directly.
      const message = p.error?.message || 'Codex error notification';
      const info = p.error?.codexErrorInfo;
      if (conv.busy) {
        log.info(`[codex] stashing mid-turn error for turn/completed: ${message.slice(0, 160)}`);
        conv.stashedError = { message, info };
      } else {
        emitError(conv, message, info, p.error?.additionalDetails);
      }
      break;
    }

    case 'mcpServer/startupStatus/updated': {
      // Surface MCP servers (from MCP.json → -c overrides) that fail to start,
      // so a misconfigured server is visible instead of silently absent.
      if (p.status === 'failed' || p.status === 'cancelled') {
        log.warn(`[codex] MCP server "${p.name}" ${p.status}${p.error ? `: ${p.error}` : ''}`);
      }
      break;
    }

    // thread/started, thread/status/changed, remoteControl/status/changed —
    // informational, no-op for the dashboard.
  }
}

/* ── Conversation teardown ─────────────────────────────────────────────── */

function teardownConversation(conversationId: string): void {
  const conv = conversations.get(conversationId);
  if (!conv) return;
  clearTurnWatchdog(conv);
  clearHardTurnTimer(conv);
  conversations.delete(conversationId);
  discardTokens(conv);
  try { conv.rpc.close(); } catch {}
  // bot:done guarantee: slot-freeing consumers (WhatsApp activeAgents, the
  // scheduler) resolve ONLY on bot:done — claude guarantees it in a finally,
  // so every codex terminal path (stop, settings:save teardown, init failure,
  // crash) must too, or each failed customer message pins a slot until restart.
  if (conv.oneShot) emitDone(conv);
  conv.onMessage('bot:conversation-ended', { conversationId });
  // Re-warm for the next live conversation (mirrors claude's finally).
  if (!conv.oneShot) void warmUpForLiveConversation(conv.model, conv.names);
}

/* ── Spawn + initialize (with pre-warm pool) ───────────────────────────── */

/**
 * Pre-warmed app-server: spawned + initialize handshake done. `thread/start`
 * is deliberately NOT issued at warm time — it's a single fast local RPC and
 * deferring it means the warm process is claimable for ANY model/instructions
 * (claude's warmer must bake the full options in and misses whenever
 * recentMessages differ; this design sidesteps that). Keyed on the MCP spawn
 * flags, the only thing fixed at spawn time.
 */
interface CodexWarmEntry {
  key: string;
  rpc: CodexRpc;
}

let codexWarm: CodexWarmEntry | null = null;
let codexWarmInflight: Promise<void> | null = null;

function warmKeyFor(mcpArgs: string[]): string {
  return crypto.createHash('sha256').update(JSON.stringify(mcpArgs)).digest('hex');
}

function claimWarmRpc(key: string): CodexRpc | null {
  if (!codexWarm) return null;
  if (codexWarm.key !== key) {
    // MCP config changed since warm time — the flags are baked into the spawn.
    try { codexWarm.rpc.close(); } catch {}
    codexWarm = null;
    return null;
  }
  const rpc = codexWarm.rpc;
  codexWarm = null;
  log.info('[codex] claimed pre-warmed app-server');
  return rpc;
}

function discardCodexWarmup(): void {
  if (codexWarm) {
    try { codexWarm.rpc.close(); } catch {}
    codexWarm = null;
  }
}

export async function warmUpForLiveConversation(
  _model: string,
  _names?: { botName: string; humanName: string },
): Promise<void> {
  if (codexWarmInflight) return codexWarmInflight;
  const mcpArgs = buildMcpConfigArgs();
  const key = warmKeyFor(mcpArgs);
  if (codexWarm?.key === key) return;

  codexWarmInflight = (async () => {
    try {
      const token = await getCodexAccessToken();
      if (!token) return; // not authed — nothing to warm
      if (codexWarm && codexWarm.key !== key) discardCodexWarmup();
      const rpc = new CodexRpc();
      rpc.start(mcpArgs);
      rpc.onClose(() => {
        // Warm process died on its own — drop it from the cache so a claim
        // never hands out a dead rpc.
        if (codexWarm?.rpc === rpc) codexWarm = null;
      });
      await rpc.request('initialize', { clientInfo: CLIENT_INFO });
      rpc.notify('initialized', {});
      codexWarm = { key, rpc };
      log.ok('[codex] app-server pre-warmed');
    } catch (err: any) {
      log.warn(`[codex] pre-warm skipped: ${err?.message || err}`);
    } finally {
      codexWarmInflight = null;
    }
  })();
  return codexWarmInflight;
}

interface SpawnOpts {
  oneShot: boolean;
  /** Customer-facing runs (supportPrompt personas) get NO workspace skills —
   *  mirrors claude's `skills: supportPrompt ? [] : ...` gate so internal ops
   *  skills can't leak into the customer context. */
  wantSkills: boolean;
  names?: { botName: string; humanName: string };
}

async function spawnAndInitialize(
  conversationId: string,
  model: string,
  onMessage: OnAgentMessage,
  instructions: string,
  opts: SpawnOpts,
): Promise<CodexConversation | null> {
  // Pre-flight: confirm we have valid OAuth tokens before spending time spawning.
  const token = await getCodexAccessToken();
  if (!token) {
    onMessage('bot:error', {
      conversationId,
      error: 'Codex credentials not found or expired. Re-authenticate from the dashboard.',
    });
    return null;
  }

  const { id: parsedModelId, effort: parsedEffort } = parseModelString(model);
  const mcpArgs = buildMcpConfigArgs();

  // One attempt with the pre-warmed process (live convs only — one-shots churn
  // too fast to be worth re-warming for), falling back to a cold spawn if the
  // warm one fails its thread/start.
  const attempts: Array<'warm' | 'cold'> = [];
  if (!opts.oneShot && codexWarm) attempts.push('warm');
  attempts.push('cold');

  for (const attempt of attempts) {
    let rpc: CodexRpc;
    if (attempt === 'warm') {
      const claimed = claimWarmRpc(warmKeyFor(mcpArgs));
      if (!claimed) continue;
      rpc = claimed;
    } else {
      rpc = new CodexRpc();
      rpc.start(mcpArgs);
    }

    const conv: CodexConversation = {
      id: conversationId,
      rpc,
      threadId: '',
      effort: parsedEffort,
      model,
      names: opts.names,
      onMessage,
      currentTurnId: null,
      currentMsgItemId: null,
      fullText: '',
      itemTexts: new Map(),
      usedFileTools: false,
      pendingInputs: [],
      busy: false,
      oneShot: opts.oneShot,
      errorEmitted: false,
      doneEmitted: false,
      stashedError: null,
      lastContextTokens: 0,
      lastContextWindow: 0,
      turnWatchdog: null,
      hardTurnTimer: null,
      tokenBuf: '',
      tokenFlushTimer: null,
    };

    rpc.onNotification((n) => handleNotification(conv, n));
    rpc.onClose((code, errMsg) => {
      if (conversations.get(conversationId) !== conv) return;
      // App-server died out from under a live conversation. Surface it NOW —
      // without this the user stares at frozen typing dots until the leaked
      // watchdog ghost-fires 5 minutes later with a misleading "timed out"
      // (and that orphaned watchdog could tear down a successor conversation
      // started under the same convId in the meantime).
      clearTurnWatchdog(conv);
      clearHardTurnTimer(conv);
      conversations.delete(conversationId);
      flushTokens(conv);
      if (conv.busy) {
        conv.busy = false;
        conv.currentTurnId = null;
        emitError(conv, errMsg || `codex app-server exited unexpectedly (code=${code}).`);
        if (!conv.oneShot) {
          emitEvent(conv, 'bot:turn-complete', {
            conversationId: conv.id,
            usedFileTools: conv.usedFileTools,
            contextTokens: conv.lastContextTokens || 0,
            contextWindow: conv.lastContextWindow || 0,
            idle: true,
          });
        }
      }
      if (conv.oneShot) emitDone(conv);
      onMessage('bot:conversation-ended', { conversationId });
      if (!conv.oneShot) void warmUpForLiveConversation(conv.model, conv.names);
    });

    try {
      log.info(`[codex] init conversation ${conversationId} (model=${parsedModelId}${parsedEffort ? `, effort=${parsedEffort}` : ''}, ${attempt})`);
      if (attempt === 'cold') {
        await rpc.request('initialize', { clientInfo: CLIENT_INFO });
        rpc.notify('initialized', {});
      }

      // Validate/repair the configured model + effort against the live catalog —
      // a stale model id or retired effort tier otherwise fails the first message
      // with a cryptic teardown (codex has no config-level model migrations).
      const { modelId, effort } = await validateModelSelection(rpc, parsedModelId, parsedEffort);
      conv.effort = effort;

      // Context auto-compaction is ON by default in the codex app-server: when the
      // thread's token count crosses the model's threshold it compacts history in
      // place (emitting a `contextCompaction` item) and continues — no flag needed
      // here. A manual trigger also exists (`thread/compact/start`) if we ever want
      // to force it from the UI.
      const startResult = await rpc.request<{ thread: { id: string } }>('thread/start', {
        cwd: WORKSPACE_DIR,
        model: modelId,
        // Bloby's persona/workflow prompt rides developerInstructions (ADDITIVE),
        // NOT baseInstructions. baseInstructions fully OVERRIDES codex's native base
        // prompt — which carries the apply_patch FREEFORM spec + shell protocol the
        // model needs to edit files. Leaving baseInstructions unset keeps that native
        // scaffolding; developerInstructions layers Bloby's persona on top of it.
        developerInstructions: instructions,
        personality: 'pragmatic',
        // Bloby's posture matches Claude's bypassPermissions — the bot is
        // running on the user's own machine with their full consent. Skip the
        // approval prompts and give it write access to the workspace + beyond.
        approvalPolicy: 'never',
        sandbox: 'danger-full-access',
      });
      conv.threadId = startResult.thread.id;
      conversations.set(conversationId, conv);
      log.ok(`[codex] thread started ${conv.threadId}`);

      // Prime codex's per-thread skill cache with the workspace skills
      // directory. Without this, codex only sees its system-scope skills and
      // never discovers anything Bloby ships in `workspace/skills/*`. Fire and
      // forget — failure here just means workspace skills won't be auto-routable
      // for this thread, but the agent can still read SKILL.md files directly.
      // Customer-facing personas skip it entirely (see SpawnOpts.wantSkills).
      if (opts.wantSkills) primeWorkspaceSkills(rpc);

      return conv;
    } catch (err: any) {
      rpc.close();
      if (attempt === 'warm') {
        log.warn(`[codex] warm claim failed (${err.message}) — retrying with a cold spawn`);
        continue;
      }
      onMessage('bot:error', { conversationId, error: `Failed to initialize Codex: ${err.message}` });
      return null;
    }
  }
  return null; // unreachable — 'cold' attempt always returns/errors above
}

/* ── Model catalog validation ──────────────────────────────────────────── */

interface ValidatedModel { modelId: string; effort?: string }
const modelValidationCache = new Map<string, ValidatedModel>();

/**
 * Validate the configured model id + effort against `model/list` (local
 * catalog, one fast RPC — cached per model string for the process lifetime).
 * Unknown model → repair to the catalog default with a logged warning instead
 * of letting thread/start or turn/start fail with a cryptic teardown. Unknown
 * effort → the model's defaultReasoningEffort. If model/list itself fails,
 * proceed unvalidated (today's behavior).
 */
async function validateModelSelection(rpc: CodexRpc, modelId: string, effort?: string): Promise<ValidatedModel> {
  const cacheKey = `${modelId}:${effort || ''}`;
  const cached = modelValidationCache.get(cacheKey);
  if (cached) return cached;

  let result: ValidatedModel = { modelId, effort };
  try {
    const res = await rpc.request<{ data?: any[] }>('model/list', { includeHidden: true });
    const catalog = Array.isArray(res?.data) ? res.data : [];
    if (catalog.length) {
      let entry = catalog.find((m: any) => m.id === modelId || m.model === modelId);
      if (!entry) {
        const fallback = catalog.find((m: any) => m.isDefault) || catalog[0];
        log.warn(`[codex] model "${modelId}" not in catalog — falling back to "${fallback.id}"`);
        entry = fallback;
        result.modelId = entry.id;
        result.effort = undefined; // stale effort may not apply to the fallback
      }
      const supported: string[] = (entry.supportedReasoningEfforts || [])
        .map((o: any) => (typeof o === 'string' ? o : o?.reasoningEffort))
        .filter(Boolean);
      if (result.effort && supported.length && !supported.includes(result.effort)) {
        const repaired = entry.defaultReasoningEffort || undefined;
        log.warn(`[codex] effort "${result.effort}" not supported by ${result.modelId} (supported: ${supported.join(', ')}) — using ${repaired || 'server default'}`);
        result.effort = repaired;
      }
      // Only cache results actually validated against a catalog — caching the
      // passthrough on a failed/empty model/list would disable the auto-repair
      // for this model string for the whole process lifetime.
      modelValidationCache.set(cacheKey, result);
    }
  } catch (err: any) {
    log.warn(`[codex] model/list validation skipped: ${err.message}`);
  }
  return result;
}

// Codex discovers "repo"-scope skills under `<cwd>/.codex/skills` (verified
// against 0.135.0 — a bare `<cwd>/skills` is NOT scanned). Bloby keeps the
// canonical skills in `workspace/skills/<name>`, so we mirror each one into
// `.codex/skills/<name>` as a symlink — single source of truth, discoverable
// by codex's native router. 0.138 added `skills/extraRoots/set` which could
// replace this mirror — deliberately NOT adopted yet: the mirror is
// e2e-verified working and the swap buys no user-visible change.
// (Each SKILL.md needs YAML frontmatter or codex rejects it — see SKILL_FORMAT_MIGRATION.md.)
const CODEX_SKILLS_ROOT = path.join(WORKSPACE_DIR, '.codex', 'skills');

function primeWorkspaceSkills(rpc: CodexRpc): void {
  mirrorSkillsInto(CODEX_SKILLS_ROOT, 'codex');
  rpc.request('skills/list', {
    cwds: [WORKSPACE_DIR],
    forceReload: true,
  }).then((result: any) => {
    const entry = result?.data?.[0];
    const all = entry?.skills ?? [];
    const repo = all.filter((s: any) => s.scope === 'repo');
    const errors = entry?.errors ?? [];
    log.ok(`[codex] skills primed: ${repo.length} workspace (repo), ${all.length - repo.length} user/system${errors.length ? `, ${errors.length} rejected` : ''}`);
    for (const err of errors) log.warn(`[codex] skill load error: ${err.path} — ${err.message}`);
  }).catch((err: any) => {
    log.warn(`[codex] skills/list failed: ${err.message}`);
  });
}

/* ── MCP wiring ────────────────────────────────────────────────────────── */

const MCP_CONFIG_FILE = path.join(WORKSPACE_DIR, 'MCP.json');

/**
 * Load MCP servers from workspace/MCP.json (the same file the Claude harness
 * reads). Accepts the canonical unwrapped map `{ name: { command, args, env } }`,
 * a `{ mcpServers: {...} }` wrapper, or a legacy array of single-key maps.
 * Returns {} when absent/invalid — so this is a no-op until the user populates MCP.json.
 */
function loadMcpServersForCodex(): Record<string, any> {
  try {
    const raw = JSON.parse(fs.readFileSync(MCP_CONFIG_FILE, 'utf-8'));
    let map: any = raw;
    if (raw && typeof raw === 'object' && raw.mcpServers && typeof raw.mcpServers === 'object') map = raw.mcpServers;
    else if (Array.isArray(raw)) map = Object.assign({}, ...raw);
    if (map && typeof map === 'object' && !Array.isArray(map)) return map;
  } catch {}
  return {};
}

/** Serialize a JS value as a TOML literal for a `-c key=value` override. */
function toToml(v: any): string {
  if (Array.isArray(v)) return `[${v.map(toToml).join(',')}]`;
  if (v && typeof v === 'object') return `{${Object.entries(v).map(([k, val]) => `${JSON.stringify(k)}=${toToml(val)}`).join(',')}}`;
  if (typeof v === 'number' || typeof v === 'boolean') return String(v);
  return JSON.stringify(String(v)); // TOML basic string — JSON escaping is compatible
}

/** codex's config layer requires STRING values for args items, env values and
 *  http_headers — a numeric `"PORT": 3000` in MCP.json (fine on claude) would
 *  otherwise kill the app-server at config load, bricking EVERY codex spawn.
 *  Coerce so non-string JSON values degrade to their string form instead. */
function toTomlStringArray(arr: any[]): string {
  return `[${arr.map((v) => JSON.stringify(String(v))).join(',')}]`;
}
function toTomlStringMap(obj: Record<string, any>): string {
  return `{${Object.entries(obj).map(([k, v]) => `${JSON.stringify(k)}=${JSON.stringify(String(v))}`).join(',')}}`;
}

/**
 * Translate MCP.json into `codex app-server -c mcp_servers.<name>.<field>=<toml>`
 * spawn flags. Codex sources MCP from its own config layer rather than a per-query
 * param (verified against 0.135.0: a `-c mcp_servers.X.command=...` override shows
 * up in both mcpServerStatus/list and config/read). Stdio entries translate
 * command/args/env; url entries (streamable HTTP — the Claude SDK's http/sse
 * form) translate url + headers→http_headers + bearer_token_env_var, supported
 * natively by the 0.138 binary. Names must be TOML-bare-key safe.
 */
function buildMcpConfigArgs(): string[] {
  const servers = loadMcpServersForCodex();
  const args: string[] = [];
  let wired = 0;
  for (const [name, cfg] of Object.entries(servers)) {
    if (!/^[A-Za-z0-9_-]+$/.test(name)) { log.warn(`[codex] skipping MCP server "${name}" — name not TOML-bare-key safe`); continue; }
    const c: any = cfg || {};
    if (c.command) {
      args.push('-c', `mcp_servers.${name}.command=${toToml(c.command)}`);
      if (Array.isArray(c.args) && c.args.length) args.push('-c', `mcp_servers.${name}.args=${toTomlStringArray(c.args)}`);
      if (c.env && typeof c.env === 'object' && Object.keys(c.env).length) args.push('-c', `mcp_servers.${name}.env=${toTomlStringMap(c.env)}`);
    } else if (typeof c.url === 'string' && c.url) {
      args.push('-c', `mcp_servers.${name}.url=${toToml(c.url)}`);
      if (c.headers && typeof c.headers === 'object' && Object.keys(c.headers).length) {
        args.push('-c', `mcp_servers.${name}.http_headers=${toTomlStringMap(c.headers)}`);
      }
      if (typeof c.bearer_token_env_var === 'string' && c.bearer_token_env_var) {
        args.push('-c', `mcp_servers.${name}.bearer_token_env_var=${toToml(c.bearer_token_env_var)}`);
      }
    } else {
      log.warn(`[codex] skipping MCP server "${name}" — no command or url`);
      continue;
    }
    wired++;
  }
  if (wired) log.info(`[codex] wiring ${wired} MCP server(s) from MCP.json via -c overrides`);
  return args;
}

/* ── Harness implementation ────────────────────────────────────────────── */

export function hasConversation(conversationId: string): boolean {
  return conversations.has(conversationId);
}

export function isConversationBusy(conversationId: string): boolean {
  return conversations.get(conversationId)?.busy ?? false;
}

/** True while any one-shot is in its init window (token refresh + spawn +
 *  initialize + thread/start). Once the conv registers in `conversations` and
 *  startTurn sets busy, anyConversationBusy() takes over seamlessly. */
export function anyOneShotActive(): boolean {
  return inFlightOneShots.size > 0;
}

export function anyConversationBusy(): boolean {
  for (const c of conversations.values()) if (c.busy) return true;
  return false;
}

export async function startConversation(
  conversationId: string,
  model: string,
  onMessage: OnAgentMessage,
  names?: { botName: string; humanName: string },
  recentMessages?: RecentMessage[],
): Promise<boolean> {
  if (conversations.has(conversationId)) endConversation(conversationId);
  // Typing dots NOW — the supervisor awaits this whole handshake before the
  // first pushMessage can run, and without an early signal the user stares at
  // a dead chat through spawn + initialize + thread/start on every session
  // start (boot, clear-context, every proactive recycle).
  onMessage('bot:typing', { conversationId });
  const baseInstructions = await assembleBaseInstructions(names, recentMessages);
  const conv = await spawnAndInitialize(conversationId, model, onMessage, baseInstructions, {
    oneShot: false,
    wantSkills: true,
    names,
  });
  if (!conv) {
    // bot:typing above set the supervisor's agentQueryActive; bot:error alone
    // doesn't clear it — conversation-ended does (and is safe for a conv that
    // never registered).
    onMessage('bot:conversation-ended', { conversationId });
    return false;
  }
  return true;
}

export function pushMessage(
  conversationId: string,
  content: string,
  attachments?: AgentAttachment[],
  savedFiles?: SavedFile[],
): boolean {
  const conv = conversations.get(conversationId);
  if (!conv) {
    log.warn(`[codex] pushMessage: no live conversation ${conversationId}`);
    return false;
  }
  queueOrStart(conv, content, savedFiles, attachments);
  return true;
}

export function endConversation(conversationId: string): void {
  const conv = conversations.get(conversationId);
  if (!conv) return;
  log.info(`[codex] ending conversation ${conversationId}`);
  if (conv.currentTurnId) {
    void conv.rpc.request('turn/interrupt', {
      threadId: conv.threadId,
      turnId: conv.currentTurnId,
    }).catch(() => {});
  }
  teardownConversation(conversationId);
}

export function endAllConversations(): void {
  for (const id of Array.from(conversations.keys())) endConversation(id);
  // The pre-warmed app-server may hold pre-re-auth state — drop it (a fresh
  // warm-up fires from the teardown re-warm path with current credentials).
  discardCodexWarmup();
}

export async function stopSubAgentTask(_conversationId: string, _taskId: string): Promise<void> {
  // Codex doesn't expose Claude-style sub-agent tasks (collab tools are still
  // experimental and not enabled by our initialize). No-op.
}

export async function startBlobyAgentQuery(
  conversationId: string,
  prompt: string,
  model: string,
  onMessage: OnAgentMessage,
  attachments?: AgentAttachment[],
  savedFiles?: SavedFile[],
  names?: { botName: string; humanName: string },
  recentMessages?: RecentMessage[],
  supportPrompt?: string,
  _maxTurns?: number, // no codex equivalent — bounded by ONE_SHOT_MAX_TURN_MS instead
): Promise<void> {
  inFlightOneShots.add(conversationId);
  try {
    if (conversations.has(conversationId)) endConversation(conversationId);
    onMessage('bot:typing', { conversationId });
    // Support personas ride supportPrompt as the FULL instructions (SCRIPT.md
    // governs them) — but the per-customer message buffer must still be
    // appended, exactly like claude: every one-shot is a fresh thread, so
    // recentMessages IS the agent's only memory of the conversation.
    let baseInstructions = supportPrompt ?? await assembleBaseInstructions(names, recentMessages);
    if (supportPrompt && recentMessages?.length) {
      baseInstructions += `\n\n---\n# Recent Conversation\n${formatConversationHistory(recentMessages)}`;
    }
    const conv = await spawnAndInitialize(conversationId, model, onMessage, baseInstructions, {
      oneShot: true,
      wantSkills: !supportPrompt,
      names,
    });
    if (!conv) {
      // Init failed (no auth / spawn / thread-start error — bot:error already
      // emitted). bot:done frees the caller's slot; without it each failed
      // customer message pins one of the 5 WhatsApp agent slots until restart.
      onMessage('bot:done', { conversationId, usedFileTools: false });
      return;
    }
    await startTurn(conv, prompt, savedFiles, attachments);
  } finally {
    inFlightOneShots.delete(conversationId);
  }
}

export function stopBlobyAgentQuery(conversationId: string): void {
  endConversation(conversationId); // teardown guarantees the one-shot's bot:done
}

// ── Workspace agent endpoint (POST /api/agent/query) ──────────────────────

/**
 * One-shot Codex query that spawns its own short-lived app-server, runs a
 * single turn, returns the accumulated response, and tears down. Mirrors
 * what `agent-api.ts` previously did directly with the Claude SDK.
 *
 * `sessionId` carries a Codex `threadId` — when present we issue a
 * `thread/resume` instead of `thread/start` to preserve server-side context.
 */
export async function runAgentQuery(req: AgentQueryRequest): Promise<AgentQueryResult> {
  const token = await getCodexAccessToken();
  if (!token) {
    return { ok: false, error: 'Codex credentials not found or expired. Re-authenticate via the dashboard.' };
  }

  // Pull the active model + parse effort suffix from the user's config —
  // agent-api callers don't get to pick.
  let model = 'gpt-5.5';
  let effort: string | undefined;
  try {
    const { loadConfig: loadCfg } = await import('../../shared/config.js');
    const cfg = loadCfg();
    if (cfg.ai?.model) {
      const parsed = parseModelString(cfg.ai.model);
      model = parsed.id;
      effort = parsed.effort;
    }
  } catch {}

  const timeout = Math.min(Math.max(req.timeout || 120_000, 5_000), 300_000);

  const rpc = new CodexRpc();
  rpc.start(buildMcpConfigArgs());

  let fullText = '';
  const usedTools = new Set<string>();
  let usedFileTools = false;
  let resolvedThreadId = req.sessionId || '';
  let resolveTurn: (() => void) | null = null;
  let turnError: string | null = null;
  const itemTexts = new Map<string, string>();

  const turnDone = new Promise<void>((r) => { resolveTurn = r; });

  rpc.onNotification((n) => {
    const p = n.params || {};
    switch (n.method) {
      case 'item/agentMessage/delta': {
        if (typeof p.delta === 'string') {
          fullText += p.delta;
          if (p.itemId) itemTexts.set(p.itemId, (itemTexts.get(p.itemId) || '') + p.delta);
        }
        break;
      }
      case 'item/started': {
        const item = p.item || {};
        // Same claude-vocabulary normalization as the live path — agent-api
        // callers see the identical toolsUsed names on both harnesses.
        if (item.type === 'commandExecution') usedTools.add('Bash');
        else if (item.type === 'mcpToolCall') usedTools.add(item.tool ? (item.server ? `mcp__${item.server}__${item.tool}` : item.tool) : 'mcp_tool');
        else if (item.type === 'fileChange') { usedTools.add('Edit'); usedFileTools = true; }
        else if (item.type === 'webSearch') usedTools.add('WebSearch');
        break;
      }
      case 'item/completed': {
        const item = p.item || {};
        if (item.type === 'fileChange') usedFileTools = true;
        // item/completed is authoritative per item — same per-itemId
        // reconciliation as the live path: append items that never streamed
        // deltas, and the remainder when the final text extends past them.
        if (item.type === 'agentMessage' && item.id) {
          const finalText: string = (typeof item.text === 'string' && item.text)
            ? item.text
            : ((item.content || []).map((c: any) => c.text || '').join(''));
          if (finalText) {
            const streamed = itemTexts.get(item.id) || '';
            let remainder = '';
            if (!streamed) {
              remainder = finalText;
            } else if (finalText.length > streamed.length && finalText.startsWith(streamed)) {
              remainder = finalText.slice(streamed.length);
            }
            if (remainder) {
              if (!streamed && fullText && !fullText.endsWith('\n')) fullText += '\n\n';
              fullText += remainder;
              itemTexts.set(item.id, finalText);
            }
          }
        }
        break;
      }
      case 'turn/completed': {
        const status = p.turn?.status || 'completed';
        if (status === 'failed') {
          turnError = humanizeCodexError(p.turn?.error?.message, p.turn?.error?.codexErrorInfo, p.turn?.error?.additionalDetails);
        }
        resolveTurn?.();
        break;
      }
      case 'error': {
        if (p.willRetry) break; // transient — codex retries itself
        turnError = humanizeCodexError(p.error?.message || 'Codex error', p.error?.codexErrorInfo, p.error?.additionalDetails);
        resolveTurn?.();
        break;
      }
    }
  });

  // App-server crash mid-query: without this, no notification ever resolves
  // turnDone and the caller burns the FULL timeout before getting a generic
  // "timed out" — surface the real exit error immediately instead.
  rpc.onClose((code, errMsg) => {
    if (!turnError) turnError = errMsg || `codex app-server exited unexpectedly (code=${code}).`;
    resolveTurn?.();
  });

  const timeoutHandle = setTimeout(() => {
    if (!turnError) turnError = `Query timed out after ${timeout}ms.`;
    resolveTurn?.();
  }, timeout);

  try {
    log.info(`[codex/agent-api] Query: msg="${req.message.slice(0, 80)}..." model=${model} resume=${req.sessionId || 'none'}`);
    await rpc.request('initialize', { clientInfo: CLIENT_INFO });
    rpc.notify('initialized', {});

    // Same stale-model/effort auto-repair as spawnAndInitialize — agent-api is
    // otherwise the one codex entry point that fails raw on a retired model.
    const validated = await validateModelSelection(rpc, model, effort);
    model = validated.modelId;
    effort = validated.effort;

    // Same execution posture on EVERY thread path — resume included.
    // ThreadResumeParams accepts all of these in 0.138; without them a resumed
    // session silently dropped the caller's systemPrompt and fell back to
    // config-default sandbox/approval settings.
    const threadOverrides = {
      cwd: WORKSPACE_DIR,
      model,
      ...(req.systemPrompt ? { developerInstructions: req.systemPrompt } : {}),
      personality: 'pragmatic',
      approvalPolicy: 'never',
      sandbox: 'danger-full-access',
    };

    if (req.sessionId) {
      // Resume an existing thread (if codex still has it). Caller must accept
      // failure here — we fall back to a fresh thread.
      try {
        const r = await rpc.request<{ thread: { id: string } }>('thread/resume', {
          threadId: req.sessionId,
          ...threadOverrides,
        });
        resolvedThreadId = r.thread.id;
      } catch (err: any) {
        log.warn(`[codex/agent-api] thread/resume failed (${err.message}); starting fresh thread`);
        const r = await rpc.request<{ thread: { id: string } }>('thread/start', threadOverrides);
        resolvedThreadId = r.thread.id;
      }
    } else {
      const r = await rpc.request<{ thread: { id: string } }>('thread/start', threadOverrides);
      resolvedThreadId = r.thread.id;
    }

    // Same priming as live conversations — workspace skills won't be visible
    // to codex's router otherwise.
    primeWorkspaceSkills(rpc);

    const turnParams: Record<string, any> = {
      threadId: resolvedThreadId,
      input: [{ type: 'text', text: req.message }],
    };
    if (effort) turnParams.effort = effort;
    await rpc.request('turn/start', turnParams);

    await turnDone;

    if (turnError) {
      return { ok: false, error: turnError, sessionId: resolvedThreadId, toolsUsed: Array.from(usedTools) };
    }

    log.info(`[codex/agent-api] Done: ${fullText.length} chars, tools=[${Array.from(usedTools).join(',')}], thread=${resolvedThreadId}`);
    return { ok: true, response: fullText, sessionId: resolvedThreadId, toolsUsed: Array.from(usedTools), usedFileTools };
  } catch (err: any) {
    return { ok: false, error: err?.message || String(err), sessionId: resolvedThreadId };
  } finally {
    clearTimeout(timeoutHandle);
    rpc.close();
  }
}