/**
 * Codex Adapter — WebSocket Proxy Mode
 *
 * Spawns `codex app-server --listen ws://127.0.0.1:<port>` and runs a proxy
 * on a second port. Codex TUI connects to the proxy; Bridge forwards all
 * traffic while intercepting agentMessages for Claude.
 *
 * Key design: app-server connection is PERSISTENT (never closed on TUI
 * disconnect), because TUI rapidly reconnects between bootstrap phases.
 */

import { spawn, execFileSync, type ChildProcess } from "node:child_process";
import { createInterface } from "node:readline";
import { EventEmitter } from "node:events";
import { StateDirResolver } from "./state-dir";
import { cleanupPorts, portPidsCommand } from "./port-cleanup";
import { createProcessLogger, type ProcessLogger } from "./process-log";
import type { BridgeMessage } from "./types";
import type { ServerWebSocket } from "bun";
import {
  APP_SERVER_RATE_LIMIT_ERROR_CODES,
  isAppServerNotification,
  isAppServerRequestMessage,
  isAppServerResponseMessage,
  isTrackedAppServerRequestMethod,
  parseAppServerVersion,
  type AppServerInfo,
  type AppServerInitializeResponse,
  type AppServerItem,
  type AppServerNotification,
  type AppServerRequest,
  type AppServerResponse,
  type AppServerServerRequestMethod,
  type AppServerTrackedRequestMethod,
  type TurnInterruptParams,
  type TurnStartParams,
  type TurnSteerParams,
} from "./app-server-protocol";
import { parsePositiveIntEnv } from "./env-utils";
import { DEFAULT_INTERRUPT_TIMEOUT_MS, clampInterruptTimeoutMs } from "./interrupt-timing";
import {
  CODEX_TRANSPORT_ENV,
  TcpToUnixRelay,
  type CodexTransport,
  codexListenArg,
  codexSocketPath,
  ensureSocketDir,
  parseTransportMode,
  removeSocketFile,
  resolveCodexTransport,
  waitForUnixWsReady,
} from "./codex-transport";
import {
  ADAPTER_DISCONNECT_REASON,
  APP_SERVER_RECONNECT_NEW_TUI_REASON,
} from "./turn-notices";
import { isAllowedWsUpgrade, wsOriginRejectedResponse } from "./ws-origin-guard";
import { PendingRequestRegistry } from "./pending-request-registry";

interface TuiSocketData {
  connId: number;
}

interface PendingServerRequest {
  raw: string;
  serverId: number | string;
  connId: number;
  method: AppServerServerRequestMethod | string;
  timestamp: number;
  threadId: string | null;
}

interface BufferedServerRequest {
  raw: string;
  serverId: number | string;
  method: AppServerServerRequestMethod | string;
  threadId: string | null;
}

interface PendingServerResponse {
  raw: string;
  serverId: number | string;
  method: AppServerServerRequestMethod | string;
  timestamp: number;
}

interface PendingRequest {
  method: AppServerTrackedRequestMethod;
  threadId?: string;
  /**
   * Monotonic "thread switch" sequence (latest-issued) for thread/start and
   * thread/resume. Used so an out-of-order OLD response can't clobber the active
   * thread the user most recently switched to (#70).
   */
  threadSwitchSeq?: number;
}

/**
 * ── Codex app-server PROTOCOL-VERSION-COUPLING CHECKLIST (P1 #5) ────────────
 *
 * The proxy intercepts and rewrites the Codex app-server protocol at a handful
 * of points. Every one of them encodes assumptions about a SPECIFIC protocol
 * snapshot — an upstream bump can silently break them. When bumping the
 * supported codex version, re-verify EACH site against codex-rs:
 *
 *   1. initialize capture — captureAppServerInfo() / handleAppServerResponse.
 *      Reads InitializeResponse.userAgent (codex-rs v1.rs:61-71). If the field
 *      set changes, the captured AppServerInfo / drift WARNING must follow.
 *   2. agentMessage aggregation — handleServerNotification (item/started +
 *      item/agentMessage/delta + item/completed). Coupled to the item/* event
 *      names and the `agentMessage` item type.
 *   3. turn lifecycle — turn/started + turn/completed notifications drive
 *      turnPhase, the inactivity watchdog, and busy-guard injection.
 *   4. rate-limit patch — patchResponse(). Matches the `account/rateLimits/read`
 *      error and hand-writes a GetAccountRateLimitsResponse-shaped mock
 *      (rateLimits / rateLimitsByLimitId). Coupled to BOTH the error shape and
 *      the success result shape (codex-rs v2/account.rs:254-258).
 *   5. thread/closed sniff — handleAppServerPayload. String-matches the
 *      `thread/closed` notification method to explain a silent TUI exit(0).
 *
 * The captured app-server version (exposed via `appServerInfo` →
 * DaemonStatus → /healthz + `abg doctor`) lets an operator see, at a glance,
 * whether the running server drifted from the version these sites were written
 * against.
 */
export class CodexAdapter extends EventEmitter {
  private static readonly RESPONSE_TRACKING_TTL_MS = 30000;

  private proc: ChildProcess | null = null;
  /** pid of the spawned `codex app-server` child; survives `stop()` nulling `proc`. */
  private appServerPid: number | null = null;
  private appServerWs: WebSocket | null = null;
  private tuiWs: ServerWebSocket<TuiSocketData> | null = null;
  private proxyServer: ReturnType<typeof Bun.serve> | null = null;
  // #85 transport: how Codex app-server is reached. In "unix" mode the adapter
  // still connects to ws://127.0.0.1:appPort, but a TcpToUnixRelay bridges that
  // TCP port to Codex's unix socket (Codex builds may drop ws:// listen support).
  private transport: CodexTransport = "ws";
  private socketPath: string | null = null;
  private relay: TcpToUnixRelay | null = null;
  private threadId: string | null = null;
  // Reserve negative ids for bridge-originated requests so they never collide
  // with proxy-rewritten TUI request ids.
  private nextInjectionId = -1;
  private appPort: number;
  private proxyPort: number;
  private readonly logFile: string;
  private readonly logger: ProcessLogger;
  private tuiConnId = 0; // tracks which TUI connection is "current" (primary)
  private connIdCounter = 0; // monotonically increasing counter for unique conn IDs
  // Secondary (picker) connections: each gets its own dedicated app-server WS
  // for raw passthrough. The primary connection stays untouched.
  private secondaryConnections = new Map<number, {
    tuiWs: ServerWebSocket<TuiSocketData>;
    appServerWs: WebSocket | null;
    buffer: string[];
    initialized: boolean;
    initializationReplayed: boolean;
  }>();

  private agentMessageBuffers = new Map<string, string[]>();
  private pendingRequests = new Map<string, PendingRequest>();
  private activeTurnIds = new Set<string>();
  turnInProgress = false;
  // #69: per-turn inactivity watchdog. A lost `turn/completed` would otherwise
  // leave turnInProgress=true forever and permanently block injection. Each
  // active turn gets a timer (keyed identically to activeTurnIds) that is
  // refreshed on any app-server notification and, on expiry, marks the turn as
  // stalled. It must NOT complete the turn: a long silent command is still a
  // live busy turn until Codex emits turn/completed or the connection is reset.
  private turnWatchdogs = new Map<string, ReturnType<typeof setTimeout>>();
  // turnStalled must fire AT MOST ONCE per turn: the watchdog is intentionally
  // NOT cleared on stall (the turn stays busy), so every subsequent app-server
  // notification re-arms it and a later expiry would emit a duplicate stalled
  // notification. This set records turnIds already reported stalled so we
  // suppress repeats; a turnId is removed when its turn completes/resets or a
  // new turn starts, so a genuinely new stall in a later turn still notifies.
  private stalledTurnIds = new Set<string>();
  // CURRENT stall state, distinct from stalledTurnIds (which is the
  // notify-at-most-once dedup set and is deliberately NOT cleared on activity).
  // A stalled turn that resumes streaming leaves this set, so turnPhase can
  // honestly report running again (protocol v2 PR A).
  private currentlyStalledTurnIds = new Set<string>();
  // Did the LAST turn end abnormally (turnAborted) with no new turn since?
  // Drives turnPhase === "aborted". Deliberately PERSISTS across app-server
  // reconnects with no intervening turn — the contract reads "no new turn
  // since", and a reconnect alone doesn't change what happened to that turn.
  private lastTurnEndedAbnormally = false;
  // Last phase reported via the unified turnPhaseChanged event — every state
  // mutation funnels through notifyPhaseIfChanged so consumers (the daemon's
  // status.json writer) can never miss a transition, including stalled→running
  // resumes that have no dedicated event of their own.
  private lastEmittedPhase: "idle" | "running" | "stalled" | "aborted" = "idle";
  // #70: latest-issued thread-switch sequence (see PendingRequest.threadSwitchSeq).
  private threadSwitchSeq = 0;

  // Proxy-layer id rewriting: upstream uses globally unique ids
  private nextProxyId = 100000;
  private upstreamToClient = new Map<number, { connId: number; clientId: number | string }>();
  private serverRequestToProxy = new Map<number, PendingServerRequest>();
  private pendingServerRequests: BufferedServerRequest[] = [];
  private pendingServerResponses = new Map<number, PendingServerResponse>();
  private staleProxyIds = new Map<number, ReturnType<typeof setTimeout>>();
  private bridgeRequestIds = new Map<number, ReturnType<typeof setTimeout>>();
  // What each bridge-originated request was (turn/start injection vs
  // turn/steer vs turn/interrupt) — error handling differs: a rejected
  // injection emits turnAborted (+ bridgeTurnRejected for correlation), a
  // rejected steer must NOT (the original turn is still running) and surfaces
  // steerFailed instead, and a rejected interrupt surfaces interruptFailed
  // (the original turn keeps running — response handlers never mutate turn
  // state; the notification flow owns it).
  private bridgeRequestKinds = new Map<number, "turn-start" | "steer" | "interrupt">();
  private intentionalDisconnect = false;
  // Fresh-session reconnection: buffer TUI messages while reconnecting app-server
  private pendingTuiMessages: string[] = [];
  private reconnectingForNewSession = false;
  private replayingBufferedMessages = false;
  // Generation counter to prevent stale app-server close handlers from interfering
  private appServerGeneration = 0;

  // Outage recovery window — when app-server is unintentionally down, buffer
  // outbound TUI messages briefly so short upstream blips don't kill the TUI.
  // If the outage runs past OUTAGE_TIMEOUT_MS or the queue overflows, we
  // fail-fast by closing the TUI WS with code 1011 so codex-rs raises a
  // visible FatalExitRequest rather than hanging silently.
  private outageQueue: Array<{ raw: string; connId: number }> = [];
  private outageTimer: ReturnType<typeof setTimeout> | null = null;
  private static readonly OUTAGE_QUEUE_MAX = 64;
  // Must outlast the first reconnect attempts (base 1s exponential: 1+2+4 ≈ 7s
  // before the third try) plus session restore — at 5s the fail-fast was firing
  // BEFORE the reconnect it exists to bridge, killing recoverable TUIs.
  private static readonly OUTAGE_TIMEOUT_MS = 10_000;

  // Session-state capture for unintentional-reconnect replay.
  //
  // When the upstream app-server drops unexpectedly and we silently
  // reconnect, the new session has no knowledge of initialize/thread
  // state. The TUI doesn't know the reconnect happened and continues
  // using its stale session. The first request that requires
  // initialized state (e.g. turn/steer) gets `"Not initialized"`,
  // codex-rs exits with `Error: turn/steer failed: Not initialized`.
  //
  // To avoid that, we capture the TUI's initialize/initialized payloads
  // and replay them against the fresh app-server session on
  // unintentional reconnect. See handleSessionRestoreAfterReconnect.
  private lastInitializeRaw: string | null = null;
  private lastInitializedRaw: string | null = null;
  // P1 #5: protocol-version awareness. Proxy ids of in-flight `initialize`
  // requests (TUI → app-server), so the matching RESPONSE can be recognized and
  // its InitializeResponse captured at the handleAppServerResponse funnel.
  private pendingInitializeProxyIds = new Set<number>();
  // Captured app-server identity (version / platform) from the last initialize
  // response. null until the first handshake completes. Surfaced on
  // DaemonStatus.appServerInfo via the getter below.
  private appServerInfo: AppServerInfo | null = null;
  // Fire the "unknown app-server version" WARNING at most once per distinct
  // captured version string, so a stable-but-unrecognized server does not spam
  // the log on every reconnect/initialize.
  private warnedAppServerVersions = new Set<string>();
  // Fire the patchResponse fragile-text-match WARNING at most once per distinct
  // error message, so a repeated legacy rate-limit error does not spam the log.
  private warnedFragileRateLimitMessages = new Set<string>();
  private sessionRestoreInProgress = false;
  // Session-replay waiter: id-keyed pending-request registry. REJECT semantics —
  // a timeout, a send failure, and an app-server error response all REJECT;
  // only a clean app-server response resolves. The timer is intentionally ref'd
  // (registry default), matching the prior raw setTimeout (no unref).
  private replayPending = new PendingRequestRegistry<unknown>();
  // Companion lookup so tryConsumeReplayResponse can name the method in its
  // reject message (`${method} rejected: ...`). Keyed by the same replay id and
  // cleared in lockstep with the registry entry on every exit path.
  private replayMethods = new Map<number | string, string>();
  private static readonly SESSION_REPLAY_TIMEOUT_MS = 5000;

  constructor(appPort = 4500, proxyPort = 4501, logFile = new StateDirResolver().logFile) {
    super();
    this.appPort = appPort;
    this.proxyPort = proxyPort;
    this.logFile = logFile;
    this.logger = createProcessLogger({ component: "CodexAdapter", logFile: this.logFile });
  }

  get appServerUrl() { return `ws://127.0.0.1:${this.appPort}`; }
  get proxyUrl() { return `ws://127.0.0.1:${this.proxyPort}`; }
  get activeThreadId() { return this.threadId; }
  /**
   * Captured Codex app-server identity (P1 #5). null until the first
   * `initialize` handshake response is observed. Read by the daemon's status
   * builder so /healthz + status.json + `abg doctor` can surface which
   * app-server build the proxy is coupled to.
   */
  get capturedAppServerInfo(): AppServerInfo | null { return this.appServerInfo; }

  // ── Lifecycle ──────────────────────────────────────────────

  async start() {
    this.intentionalDisconnect = false;
    // #85: pick the Codex transport. checkPorts still applies in both modes —
    // in unix mode the relay (not Codex) binds appPort, but the port must still
    // be free for the relay to bind it; proxyPort is always ours.
    await this.checkPorts();
    // Wrap the rest of bootstrap so a partial failure tears down whatever was
    // already stood up (relay/socket/spawned child/proxy) BEFORE rethrowing.
    // Otherwise — in unix mode especially — start() leaks the relay (a net.Server
    // bound on appPort inside the daemon's OWN process) plus the live codex child
    // when connectToAppServer rejects. bootCodex then retries start(), whose first
    // step checkPorts() sees the daemon's own pid holding appPort with a
    // `bun .../daemon.js` cmdline (NOT a `codex app-server` cmdline), classifies
    // it as a foreign process and THROWS "already in use by non-Codex process".
    // Every retry would then fail immediately at checkPorts, wasting the whole
    // backoff. Making start() idempotent on the failure path lets retries proceed
    // against a clean slate.
    try {
      this.resolveTransport();

      const listen = codexListenArg(this.transport, this.appPort, this.socketPath ?? "");
      if (this.transport === "unix" && this.socketPath) {
        ensureSocketDir(this.socketPath);
        removeSocketFile(this.socketPath); // clear any stale socket from a prior run
      }
      this.log(`Spawning codex app-server (transport=${this.transport}) --listen ${listen}`);
      this.spawnAppServer(listen);

      if (this.transport === "unix" && this.socketPath) {
        // Codex's unix listener does NOT serve HTTP /healthz — probe readiness via
        // a WS upgrade against the socket, then stand up the TCP↔unix relay so the
        // adapter's unchanged `new WebSocket(ws://127.0.0.1:appPort)` reaches it.
        await waitForUnixWsReady(this.socketPath);
        this.relay = new TcpToUnixRelay("127.0.0.1", this.appPort, this.socketPath, (m) => this.log(`[relay] ${m}`));
        await this.relay.start();
        this.log(`Transport relay ready: ws://127.0.0.1:${this.appPort} → unix://${this.socketPath}`);
      } else {
        await this.waitForHealthy();
      }

      // Connect to app-server once, keep it alive permanently
      await this.connectToAppServer();

      this.startProxy();
      this.log(`Proxy ready on ${this.proxyUrl}`);
    } catch (err) {
      const m = err instanceof Error ? err.message : String(err);
      this.log(`start() failed (${m}) — tearing down partial transport before rethrow`);
      this.cleanupAfterFailedStart();
      throw err; // bootCodex still sees the failure and retries — now clean.
    }
  }

  /** Spawn the `codex app-server` child and wire its lifecycle/log handlers. */
  private spawnAppServer(listen: string) {
    this.proc = spawn("codex", ["app-server", "--listen", listen], {
      stdio: ["pipe", "pipe", "pipe"],
    });
    // Retain the pid independently of `this.proc` (which stop() nulls): the
    // daemon's process.on("exit") last gasp uses it to guarantee the app-server
    // is killed even when process.exit() races ahead of stop()'s async SIGKILL
    // timer. Cleared only when the child truly exits.
    this.appServerPid = this.proc.pid ?? null;

    this.proc.on("error", (err) => this.emit("error", err));
    this.proc.on("exit", (code) => {
      this.appServerPid = null;
      this.emit("exit", code);
    });

    const stderrRl = createInterface({ input: this.proc.stderr! });
    stderrRl.on("line", (l) => this.log(`[codex-server] ${l}`));
    const stdoutRl = createInterface({ input: this.proc.stdout! });
    stdoutRl.on("line", (l) => this.log(`[codex-stdout] ${l}`));
  }

  /**
   * Tear down the proxy + transport relay + unix socket file. Shared by the
   * graceful disconnect() path and the start()-failure cleanup. Does NOT touch
   * the spawned codex child (callers decide whether the child should die).
   */
  private teardownTransport() {
    this.proxyServer?.stop();
    this.proxyServer = null;
    // #85: tear down the transport relay and remove the unix socket file.
    if (this.relay) {
      this.relay.stop();
      this.relay = null;
    }
    if (this.socketPath) removeSocketFile(this.socketPath);
  }

  /**
   * Idempotent teardown for a partially-constructed start(): tear down the
   * transport (proxy/relay/socket) AND kill the spawned codex child. Unlike
   * stop()'s graceful SIGTERM-then-SIGKILL dance, this synchronously SIGKILLs
   * the child via the retained pid: the daemon is about to retry start(), so the
   * port (and socket) must be free NOW — a lingering SIGTERM grace window would
   * keep the next checkPorts/relay.bind failing. Best-effort and non-throwing so
   * the original start() error is the one that propagates.
   */
  private cleanupAfterFailedStart() {
    try {
      this.teardownTransport();
    } catch (e) {
      this.log(`cleanupAfterFailedStart: teardownTransport error: ${(e as Error).message}`);
    }
    // Kill the spawned child synchronously (reuses the retained-pid SIGKILL the
    // daemon's exit handler relies on) and drop the proc reference.
    this.forceKillAppServerSync();
    this.proc = null;
  }

  /** #85: resolve the transport (env-driven, `auto` probes ws support) and the unix socket path. */
  private resolveTransport() {
    const mode = parseTransportMode(process.env[CODEX_TRANSPORT_ENV]);
    this.transport = resolveCodexTransport(mode);
    this.socketPath = this.transport === "unix" ? codexSocketPath(this.appPort) : null;
    this.log(`Codex transport mode=${mode} resolved=${this.transport}`);
  }

  /** Disconnect the bridge (proxy + app-server WS) without killing the Codex process. */
  disconnect() {
    this.intentionalDisconnect = true;

    // Cancel any pending reconnect
    if (this.reconnectTimer) {
      clearTimeout(this.reconnectTimer);
      this.reconnectTimer = null;
    }

    // Cancel any outage recovery in flight
    this.outageQueue = [];
    this.clearOutageTimer();

    this.appServerWs?.close();
    this.appServerWs = null;
    // Close all secondary (picker) connections
    for (const [id, sec] of this.secondaryConnections) {
      try { sec.appServerWs?.close(); } catch {}
      this.secondaryConnections.delete(id);
    }
    this.teardownTransport();
    this.clearResponseTrackingState();
    // #69: an intentional disconnect must synchronously drop turn state +
    // watchdog timers. We cannot rely on handleAppServerClose for this: we set
    // appServerWs=null above, and the close callback is async (and may be
    // skipped entirely if the socket was already gone), so without this the
    // turnInProgress flag and a pending watchdog timer leak past disconnect().
    // emitCompleted stays false — an intentional teardown is not a real turn
    // completion and must not signal "Codex finished" downstream.
    this.resetTurnState(ADAPTER_DISCONNECT_REASON);
  }

  /** Fully stop: disconnect bridge AND kill the Codex process. */
  stop() {
    this.intentionalDisconnect = true;
    this.disconnect();

    if (this.proc) {
      const proc = this.proc;
      this.proc = null;
      proc.kill("SIGTERM");
      // SIGKILL fallback if SIGTERM doesn't work within 2s. NOTE: this async
      // timer only fires if the daemon keeps running; when shutdown() calls
      // process.exit() right after stop(), the timer is destroyed before it
      // fires — that path is covered by forceKillAppServerSync() in the
      // daemon's process.on("exit") handler.
      const killTimer = setTimeout(() => {
        try { proc.kill("SIGKILL"); } catch {}
      }, 2000);
      proc.on("exit", () => clearTimeout(killTimer));
    }
  }

  /**
   * Synchronously force-kill the spawned app-server child if still alive.
   * Best-effort and NON-blocking — intended for the daemon's process.on("exit")
   * last gasp. `process.exit()` destroys stop()'s async SIGKILL fallback timer,
   * so without this an app-server that ignores or is slow to handle SIGTERM
   * would survive as an orphan holding the pair's port. A zombie left behind is
   * reaped by init/launchd once the daemon process fully exits.
   */
  forceKillAppServerSync(): void {
    const pid = this.appServerPid;
    if (pid === null) return;
    try {
      process.kill(pid, "SIGKILL");
    } catch {
      // already gone — nothing to do
    }
  }

  /**
   * Inject a message into the active Codex thread via turn/start.
   *
   * Returns the (negative) bridge-originated JSON-RPC request id when the
   * message was transport-accepted, or null on failure. Callers use the id to
   * correlate the later "bridgeTurnStarted"/"bridgeTurnRejected" events back
   * to the originating request (protocol v2 PR B turn_started ACK).
   *
   * Optional budget-tier overrides (plan v2.3 P4/R5) ride on the same turn/start:
   * `model`/`effort` are sticky for this thread's subsequent turns
   * (app-server protocol turn.rs: "this turn and subsequent turns"), so callers
   * own the restore lifecycle. No JSON-RPC error for the request means
   * transport-accepted — NOT confirmed-applied (turn/started carries no model).
   */
  injectMessage(text: string, overrides?: { model?: string; effort?: string }): number | null {
    if (!this.threadId) {
      this.log("Cannot inject: no active thread");
      return null;
    }
    if (!this.appServerWs || this.appServerWs.readyState !== WebSocket.OPEN) {
      this.log("Cannot inject: app-server WebSocket not connected");
      return null;
    }
    if (this.turnInProgress) {
      this.log(`Rejected injection: Codex turn is in progress (thread ${this.threadId})`);
      return null;
    }
    this.log(`Injecting message into Codex (${text.length} chars)`);
    const requestId = this.nextInjectionId--;
    this.trackBridgeRequestId(requestId);
    const params: TurnStartParams = { threadId: this.threadId, input: [{ type: "text", text }] };
    if (overrides?.model) params.model = overrides.model;
    if (overrides?.effort) params.effort = overrides.effort;
    if (overrides?.model || overrides?.effort) {
      this.log(
        `Budget tier override on turn/start (model=${overrides.model ?? "unchanged"}, effort=${overrides.effort ?? "unchanged"}) — sticky for subsequent turns; transport-accepted unless a JSON-RPC error follows`,
      );
    }
    try {
      this.appServerWs.send(JSON.stringify({
        method: "turn/start",
        id: requestId,
        params,
      } satisfies AppServerRequest<"turn/start", TurnStartParams>));
      return requestId;
    } catch (err: any) {
      this.untrackBridgeRequestId(requestId);
      this.log(`Injection send failed: ${err.message}`);
      return null;
    }
  }

  /**
   * Feed additional input into the CURRENTLY RUNNING turn via turn/steer —
   * Codex sees it mid-turn and adjusts without losing work (protocol v2 B0,
   * design consensus: explicit [STEER] framing is the CALLER's job).
   * Returns the (negative) bridge request id when transport-accepted, or null
   * on failure. The id correlates the later steerAccepted/steerFailed event
   * back to this dispatch so the daemon can release the steer's idempotency
   * key + reply expectation by id (NOT by FIFO) — a lost or out-of-order
   * response cannot then strand them onto the wrong turn. A later JSON-RPC
   * error surfaces as a "steerFailed" event (NOT turnAborted — the original
   * turn keeps running).
   */
  steerMessage(text: string): number | null {
    if (!this.threadId) {
      this.log("Cannot steer: no active thread");
      return null;
    }
    if (!this.appServerWs || this.appServerWs.readyState !== WebSocket.OPEN) {
      this.log("Cannot steer: app-server WebSocket not connected");
      return null;
    }
    if (!this.turnInProgress) {
      this.log("Cannot steer: no turn in progress (use injectMessage)");
      return null;
    }
    // turn/steer requires the active turn id as a precondition (expectedTurnId
    // — REQUIRED on every codex release that has turn/steer; omitting it gets
    // "missing field `expectedTurnId`", which is how the live E2E caught B0
    // sending steers that could never land). A turn tracked only under an
    // `unknown:` fallback key has no addressable id and cannot be steered.
    const expectedTurnId = this.currentSteerableTurnId();
    if (!expectedTurnId) {
      this.log("Cannot steer: no addressable active turn id (turn/started carried no id)");
      return null;
    }
    this.log(`Steering message into active Codex turn ${expectedTurnId} (${text.length} chars)`);
    const requestId = this.nextInjectionId--;
    this.trackBridgeRequestId(requestId, "steer");
    const params: TurnSteerParams = {
      threadId: this.threadId,
      expectedTurnId,
      input: [{ type: "text", text }],
    };
    try {
      this.appServerWs.send(JSON.stringify({
        method: "turn/steer",
        id: requestId,
        params,
      } satisfies AppServerRequest<"turn/steer", TurnSteerParams>));
      return requestId;
    } catch (err: any) {
      this.untrackBridgeRequestId(requestId);
      this.log(`Steer send failed: ${err.message}`);
      return null;
    }
  }

  /**
   * Interrupt EVERY addressable active turn via turn/interrupt (protocol v2
   * PR B; design consensus with Codex: multi-turn ambiguity is resolved as
   * "terminate ALL active turns"). `unknown:` fallback keys carry no
   * addressable id and cannot be interrupted — when no addressable id exists
   * this fails locally with a structured result instead of sending anything.
   *
   * Wire facts verified against codex-rs: params are {threadId, turnId}
   * (turn.rs TurnInterruptParams); a mismatching/inactive turnId gets an
   * immediate JSON-RPC error, while the SUCCESS response ({}) is deferred by
   * the app-server until the core's TurnAborted — arriving alongside the
   * terminal `turn/completed` (status "interrupted") notification. Callers
   * must therefore wait for the terminal boundary via waitForTurnsTerminal,
   * NOT for the interrupt responses.
   */
  interruptActiveTurns():
    | { ok: true; turnIds: string[] }
    | { ok: false; code: "interrupt_unavailable"; error: string } {
    if (!this.threadId) {
      this.log("Cannot interrupt: no active thread");
      return { ok: false, code: "interrupt_unavailable", error: "no active thread" };
    }
    if (!this.appServerWs || this.appServerWs.readyState !== WebSocket.OPEN) {
      this.log("Cannot interrupt: app-server WebSocket not connected");
      return { ok: false, code: "interrupt_unavailable", error: "app-server WebSocket not connected" };
    }
    const addressable = [...this.activeTurnIds].filter((id) => !id.startsWith("unknown:"));
    if (addressable.length === 0) {
      this.log("Cannot interrupt: no addressable active turn id (turn/started carried no id)");
      return {
        ok: false,
        code: "interrupt_unavailable",
        error: "no addressable active turn id (turn/started carried no id)",
      };
    }
    for (const turnId of addressable) {
      const requestId = this.nextInjectionId--;
      this.trackBridgeRequestId(requestId, "interrupt");
      const params: TurnInterruptParams = { threadId: this.threadId, turnId };
      try {
        this.appServerWs.send(JSON.stringify({
          method: "turn/interrupt",
          id: requestId,
          params,
        } satisfies AppServerRequest<"turn/interrupt", TurnInterruptParams>));
        this.log(`Sent turn/interrupt for active turn ${turnId} (request ${requestId})`);
      } catch (err: any) {
        // A send failure here means the socket just died mid-loop; earlier
        // interrupts MAY already be in flight, but the caller must not assume
        // the turns stopped — surface a loud structured failure (the close
        // path will reset turn state on its own if the socket is truly gone).
        this.untrackBridgeRequestId(requestId);
        this.log(`turn/interrupt send failed for ${turnId}: ${err.message}`);
        return {
          ok: false,
          code: "interrupt_unavailable",
          error: `turn/interrupt send failed (${err.message}); earlier interrupts may still land`,
        };
      }
    }
    return { ok: true, turnIds: addressable };
  }

  /**
   * Interrupt terminal-wait budget (env-overridable; default 10 seconds).
   *
   * INVARIANT (see interrupt-timing.ts): this budget must resolve BEFORE the
   * bridge client's reply timeout (daemon-client CLIENT_REPLY_TIMEOUT_MS), or
   * the client reports a false "Timed out waiting for daemon" while the daemon
   * still proceeds to inject — a Claude retry then double-turns. parsePositiveIntEnv
   * has no upper bound, so we CLAMP the env value below the client timeout to
   * make the relationship impossible to misconfigure.
   */
  private interruptTimeoutMs(): number {
    const requested = parsePositiveIntEnv(
      "AGENTBRIDGE_INTERRUPT_TIMEOUT_MS",
      DEFAULT_INTERRUPT_TIMEOUT_MS,
      (m) => this.log(m),
    );
    const clamped = clampInterruptTimeoutMs(requested);
    if (clamped !== requested) {
      this.log(
        `AGENTBRIDGE_INTERRUPT_TIMEOUT_MS=${requested}ms exceeds the safe ceiling — ` +
        `clamped to ${clamped}ms (must resolve before the client reply timeout to avoid a double-turn)`,
      );
    }
    return clamped;
  }

  /**
   * Resolve once every TARGETED turnId has reached its terminal boundary — i.e.
   * each is gone from activeTurnIds AND no longer flagged stalled — the safe
   * point after which a new turn/start cannot race the dying targeted turn(s).
   * On timeout the caller must NOT inject (double-turn race) — it gets a
   * structured failure instead.
   *
   * SCOPING (PR B fix): the predicate keys ONLY on the targeted ids, NOT on the
   * global turnPhase. turnPhase reports "running" whenever ANY turn is active,
   * so a coexisting NON-targeted turn — e.g. an unaddressable `unknown:` turn
   * that interruptActiveTurns can never target — would otherwise hold turnPhase
   * at "running" forever and guarantee a spurious timeout even after every
   * targeted turn ended. Per-id scoping makes the wait depend solely on the
   * turns we actually interrupted.
   *
   * Listens on the existing event funnels that mutate turn state:
   * turnIdCompleted (per-turn completion), turnTrackingReset (close /
   * reconnect / stop / abort resets) and turnPhaseChanged (covers
   * stalled↔running moves). The timeout timer is unref'd so a pending wait
   * never keeps the process alive.
   */
  waitForTurnsTerminal(
    turnIds: string[],
    timeoutMs = this.interruptTimeoutMs(),
    signal?: AbortSignal,
  ): Promise<{ ok: true } | { ok: false; code: "interrupt_timeout" | "interrupt_aborted" }> {
    const satisfied = () =>
      turnIds.every(
        (id) => !this.activeTurnIds.has(id) && !this.currentlyStalledTurnIds.has(id),
      );

    if (satisfied()) return Promise.resolve({ ok: true });
    // Cancelable (PR B recommend #4): when the caller already settled on another
    // signal (e.g. interruptFailed), aborting tears down THIS wait's listeners +
    // timer promptly instead of leaking them until timeoutMs elapses.
    if (signal?.aborted) return Promise.resolve({ ok: false, code: "interrupt_aborted" });

    return new Promise((resolve) => {
      let settled = false;
      const finish = (
        result: { ok: true } | { ok: false; code: "interrupt_timeout" | "interrupt_aborted" },
      ) => {
        if (settled) return;
        settled = true;
        clearTimeout(timer);
        this.off("turnIdCompleted", check);
        this.off("turnTrackingReset", check);
        this.off("turnPhaseChanged", check);
        signal?.removeEventListener("abort", onAbort);
        resolve(result);
      };
      const check = () => {
        if (satisfied()) finish({ ok: true });
      };
      const onAbort = () => finish({ ok: false, code: "interrupt_aborted" });
      const timer = setTimeout(() => {
        this.log(
          `waitForTurnsTerminal timed out after ${timeoutMs}ms (still active: ` +
          `${turnIds.filter((id) => this.activeTurnIds.has(id)).join(", ") || "none"}, phase=${this.turnPhase})`,
        );
        finish({ ok: false, code: "interrupt_timeout" });
      }, timeoutMs);
      timer.unref?.();
      this.on("turnIdCompleted", check);
      this.on("turnTrackingReset", check);
      this.on("turnPhaseChanged", check);
      signal?.addEventListener("abort", onAbort, { once: true });
    });
  }

  // ── Health Check ───────────────────────────────────────────

  private async waitForHealthy(maxRetries = 20, delayMs = 500) {
    for (let i = 0; i < maxRetries; i++) {
      try {
        const res = await fetch(`http://127.0.0.1:${this.appPort}/healthz`);
        if (res.ok) return;
      } catch {}
      await new Promise((r) => setTimeout(r, delayMs));
    }
    throw new Error("Codex app-server failed to become healthy");
  }

  // ── Persistent App-Server Connection ───────────────────────

  private connectToAppServer(isReconnect = false): Promise<void> {
    const generation = ++this.appServerGeneration;
    return new Promise((resolve, reject) => {
      const appWs = new WebSocket(this.appServerUrl);

      appWs.onopen = () => {
        if (this.appServerGeneration !== generation) {
          appWs.close();
          return;
        }
        this.appServerWs = appWs;
        this.intentionalDisconnect = false;
        this.reconnectAttempts = 0;
        this.log(isReconnect ? "Reconnected to app-server" : "Connected to app-server");
        this.flushPendingServerResponses();
        // On unintentional reconnect, restore session state (cached
        // initialize / thread/resume) BEFORE draining any TUI messages
        // that might require an initialized session. On first connect,
        // there's no state to restore so drain immediately.
        if (isReconnect) {
          this.handleSessionRestoreAfterReconnect()
            .finally(() => this.drainOutageQueue())
            .catch((e: unknown) => {
              const m = e instanceof Error ? e.message : String(e);
              this.log(`session restore unexpected error: ${m}`);
            });
        } else {
          this.drainOutageQueue();
        }
        resolve();
      };

      appWs.onmessage = (event) => {
        if (this.appServerGeneration !== generation) return;
        const data = typeof event.data === "string" ? event.data : event.data.toString();

        const forwarded = this.handleAppServerPayload(data);
        if (forwarded === null) return;

        // Forward to current TUI connection
        if (this.tuiWs) {
          try { this.tuiWs.send(forwarded); } catch (e: any) {
            this.log(`Failed to forward message to TUI: ${e.message}`);
          }
        } else {
          this.log("WARNING: response from app-server but no TUI connected, message dropped");
        }
      };

      appWs.onerror = () => {
        if (this.appServerGeneration !== generation) return;
        this.log("App-server connection error");
        if (!isReconnect) reject(new Error("Failed to connect to app-server"));
      };

      appWs.onclose = () => {
        if (this.appServerGeneration !== generation) return;
        this.handleAppServerClose();
      };
    });
  }

  /** Reconnect app-server WS to give TUI a fresh connection-scoped session. */
  private async reconnectAppServerForNewSession(tuiWs: ServerWebSocket<TuiSocketData>) {
    // Bump generation so stale handlers from the old connection become no-ops
    this.appServerGeneration++;
    this.intentionalDisconnect = true;
    if (this.reconnectTimer) {
      clearTimeout(this.reconnectTimer);
      this.reconnectTimer = null;
    }

    // Close old connection
    const oldWs = this.appServerWs;
    this.appServerWs = null;
    if (oldWs) {
      try { oldWs.close(); } catch {}
    }

    // Clear connection-scoped state (upstreamToClient etc.) but preserve
    // pendingServerRequests — those must survive the intentional reconnect
    // so they can be replayed after the TUI completes thread/resume.
    this.clearResponseTrackingStateForAppServerReconnect();
    this.resetTurnState(APP_SERVER_RECONNECT_NEW_TUI_REASON);

    try {
      await this.connectToAppServer(false);
      this.log("App-server reconnected for new TUI session — replaying buffered messages");

      // Replay raw buffered messages through onTuiMessage so they get
      // proper id-rewriting and response tracking on the fresh connection.
      const messages = this.pendingTuiMessages;
      this.pendingTuiMessages = [];
      this.reconnectingForNewSession = false;

      this.replayingBufferedMessages = true;
      try {
        for (const msg of messages) {
          this.onTuiMessage(tuiWs, msg);
        }
      } finally {
        this.replayingBufferedMessages = false;
      }
    } catch (err: any) {
      this.log(`Failed to reconnect app-server for new session: ${err.message}`);
      this.pendingTuiMessages = [];
      this.reconnectingForNewSession = false;
      this.intentionalDisconnect = false;
      this.scheduleReconnect();
    }
  }

  // ── Auto-Reconnect ──────────────────────────────────────────

  private reconnectAttempts = 0;
  private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
  private static readonly MAX_RECONNECT_ATTEMPTS = 10;
  private static readonly RECONNECT_BASE_DELAY_MS = 1000;

  private scheduleReconnect() {
    // Don't reconnect if we're shutting down or process is dead
    if (!this.proc) return;

    if (this.reconnectAttempts >= CodexAdapter.MAX_RECONNECT_ATTEMPTS) {
      this.log(`App-server reconnect failed after ${this.reconnectAttempts} attempts. Giving up.`);
      this.emit("error", new Error("App-server connection lost and reconnect failed"));
      return;
    }

    const delay = Math.min(
      CodexAdapter.RECONNECT_BASE_DELAY_MS * Math.pow(2, this.reconnectAttempts),
      30000,
    );
    this.reconnectAttempts++;
    this.log(`Scheduling app-server reconnect attempt ${this.reconnectAttempts}/${CodexAdapter.MAX_RECONNECT_ATTEMPTS} in ${delay}ms...`);

    this.reconnectTimer = setTimeout(async () => {
      try {
        await this.connectToAppServer(true);
        this.log("App-server reconnect successful");
      } catch {
        this.log("App-server reconnect attempt failed");
        this.scheduleReconnect();
      }
    }, delay);
  }

  private handleAppServerClose() {
    // Extra detail helps diagnose the "TUI silent exit" path: if app-server
    // closes first (upstream-side failure) and TUI sends anything afterwards,
    // the outage queue absorbs the blow for up to OUTAGE_TIMEOUT_MS; after
    // that we close the TUI with 1011 so codex-rs raises a visible
    // FatalExitRequest. The timestamp here pairs with codex-wrapper.log.
    const intentional = this.intentionalDisconnect;
    const tuiConnected = this.tuiWs !== null;
    this.log(
      `App-server connection closed (intentional=${intentional}, tuiConnected=${tuiConnected}, turnInProgress=${this.turnInProgress})`,
    );
    this.appServerWs = null;
    // Approval request/response ids are scoped to the current app-server session.
    // If the socket reconnects, replaying old approval state would forward stale ids.
    this.clearResponseTrackingState();
    this.resetTurnState("app-server connection closed");
    if (!intentional) {
      this.scheduleReconnect();
    }
  }

  // ── Outage recovery queue ─────────────────────────────────────

  /**
   * Buffer a TUI → app-server message during an upstream outage.
   *
   * Called when `onTuiMessage` would otherwise silently drop the message.
   * Starts a fail-fast timer on first use; overflows close the TUI with
   * 1011. The buffer is drained in-order when app-server reconnects.
   */
  private bufferDuringOutage(ws: ServerWebSocket<TuiSocketData>, raw: string): void {
    if (this.outageQueue.length >= CodexAdapter.OUTAGE_QUEUE_MAX) {
      this.log(
        `ERROR: outage queue overflow (${this.outageQueue.length}/${CodexAdapter.OUTAGE_QUEUE_MAX}) — closing TUI with 1011`,
      );
      this.outageQueue = [];
      this.clearOutageTimer();
      if (this.tuiWs && this.tuiWs === ws) {
        try {
          ws.close(1011, "agentbridge: app-server unavailable; pending TUI queue overflow");
        } catch (e: any) {
          this.log(`Failed to close TUI WS after outage queue overflow: ${e.message}`);
        }
      }
      return;
    }

    this.outageQueue.push({ raw, connId: ws.data.connId });
    this.log(
      `DIAGNOSTIC: buffered TUI message while app-server unavailable (queue size=${this.outageQueue.length}/${CodexAdapter.OUTAGE_QUEUE_MAX})`,
    );
    this.ensureOutageTimer();
  }

  private ensureOutageTimer(): void {
    if (this.outageTimer !== null) return;
    this.outageTimer = setTimeout(() => {
      this.outageTimer = null;
      const buffered = this.outageQueue.length;
      this.outageQueue = [];
      this.log(
        `ERROR: app-server did not return within ${CodexAdapter.OUTAGE_TIMEOUT_MS}ms (buffered=${buffered}) — closing TUI with 1011`,
      );
      const ws = this.tuiWs;
      if (ws) {
        try {
          ws.close(
            1011,
            `agentbridge: app-server unavailable after ${CodexAdapter.OUTAGE_TIMEOUT_MS}ms; buffered=${buffered}`,
          );
        } catch (e: any) {
          this.log(`Failed to close TUI WS on outage timeout: ${e.message}`);
        }
      }
    }, CodexAdapter.OUTAGE_TIMEOUT_MS);
  }

  private clearOutageTimer(): void {
    if (this.outageTimer !== null) {
      clearTimeout(this.outageTimer);
      this.outageTimer = null;
    }
  }

  // ── Session restore after unintentional reconnect ────────────

  /**
   * Restore codex-rs session state on the fresh app-server WS.
   *
   * Flow:
   *   1. If no cached initialize, bail (no-op, usually because the
   *      reconnect happened before the TUI's initial handshake).
   *   2. Send cached initialize, await response by id.
   *   3. Send cached `initialized` notification (fire-and-forget).
   *   4. If we know an active threadId, send `thread/resume` and await.
   *   5. On any failure, close the TUI with 1011 so codex-rs surfaces a
   *      visible FatalExitRequest instead of deadlocking on the next
   *      `"Not initialized"` response.
   *
   * Open questions for Codex source review:
   *   - Is `initialize` idempotent in codex app-server?
   *   - Does `thread/resume` work against a fresh session, or must it
   *     follow a session handshake specific to the thread's creator?
   *   - Are there session-unique fields in initialize that would cause
   *     replay to be rejected?
   * See docs/issues-2026-04-24-codex-not-initialized-after-silent-reconnect.md.
   */
  private async handleSessionRestoreAfterReconnect(): Promise<void> {
    if (!this.lastInitializeRaw) {
      this.log("DIAGNOSTIC: no cached initialize to replay after unintentional reconnect");
      return;
    }
    if (!this.appServerWs || this.appServerWs.readyState !== WebSocket.OPEN) {
      this.log("DIAGNOSTIC: app-server not open at session restore start — skipping");
      return;
    }

    this.sessionRestoreInProgress = true;
    try {
      this.log(
        `DIAGNOSTIC: replaying cached initialize to restore session (threadId=${this.threadId ?? "none"})`,
      );
      await this.sendReplayAndAwait(this.lastInitializeRaw, "initialize");

      if (this.lastInitializedRaw && this.appServerWs.readyState === WebSocket.OPEN) {
        this.appServerWs.send(this.lastInitializedRaw);
      }

      if (this.threadId && this.appServerWs.readyState === WebSocket.OPEN) {
        const replayId = `agentbridge-replay-thread-resume-${Date.now()}`;
        const resumeRaw = JSON.stringify({
          jsonrpc: "2.0",
          id: replayId,
          method: "thread/resume",
          params: { threadId: this.threadId },
        });
        await this.sendReplayAndAwait(resumeRaw, "thread/resume");
      }

      this.log(
        `DIAGNOSTIC: session restored after unintentional reconnect (threadId=${this.threadId ?? "none"})`,
      );
    } catch (err: unknown) {
      const msg = err instanceof Error ? err.message : String(err);
      this.log(
        `ERROR: session restore failed (${msg}) — closing TUI with 1011`,
      );
      const tuiWs = this.tuiWs;
      if (tuiWs) {
        try {
          tuiWs.close(1011, `agentbridge: session restore failed: ${msg}`);
        } catch (closeErr: unknown) {
          const cm = closeErr instanceof Error ? closeErr.message : String(closeErr);
          this.log(`Failed to close TUI after session restore failure: ${cm}`);
        }
      }
    } finally {
      this.sessionRestoreInProgress = false;
    }
  }

  /**
   * Send a raw JSON-RPC payload to app-server and await the response by id.
   *
   * Short-circuits the normal response handler via `replayPending`, so the
   * response is captured here and NOT forwarded to the TUI (which would
   * confuse it — the TUI didn't send this request). Rejects on timeout
   * or if the response carries an `error` field.
   */
  private sendReplayAndAwait(raw: string, method: string): Promise<unknown> {
    if (!this.appServerWs || this.appServerWs.readyState !== WebSocket.OPEN) {
      return Promise.reject(new Error("app-server not open"));
    }
    let id: number | string;
    try {
      const parsed = JSON.parse(raw) as { id?: number | string };
      if (parsed.id === undefined) {
        return Promise.reject(new Error(`replay payload for ${method} has no id`));
      }
      id = parsed.id;
    } catch (e: unknown) {
      const m = e instanceof Error ? e.message : String(e);
      return Promise.reject(new Error(`replay parse failed for ${method}: ${m}`));
    }

    // Read the (test-mutable) static ONCE for both the arm value and the
    // timeout message, matching the prior raw setTimeout which interpolated
    // the same field at fire time.
    const timeoutMs = CodexAdapter.SESSION_REPLAY_TIMEOUT_MS;
    this.replayMethods.set(id, method);
    const pending = this.replayPending.register(id, {
      timeoutMs,
      onTimeout: ({ reject }) => {
        this.replayMethods.delete(id);
        reject(new Error(`replay timeout (${timeoutMs}ms) for ${method} id=${JSON.stringify(id)}`));
      },
    });
    try {
      this.appServerWs!.send(raw);
    } catch (e: unknown) {
      // reject() clears the timer + drops the registry entry; keep the
      // companion method map in lockstep.
      this.replayMethods.delete(id);
      const m = e instanceof Error ? e.message : String(e);
      this.replayPending.reject(id, new Error(`replay send failed for ${method}: ${m}`));
    }
    return pending;
  }

  /**
   * Intercept an app-server response that matches a pending replay.
   *
   * Returns true if the response was consumed (and should NOT be forwarded
   * to TUI). Returns false if no replay is awaiting this id.
   */
  private tryConsumeReplayResponse(payload: Record<string, unknown>): boolean {
    const id = payload.id;
    if (id === undefined) return false;
    const key = id as number | string;
    if (!this.replayPending.has(key)) return false;

    const method = this.replayMethods.get(key) ?? "replay";
    this.replayMethods.delete(key);

    if (payload.error !== undefined) {
      const errMsg = typeof payload.error === "object" && payload.error !== null && "message" in payload.error
        ? String((payload.error as { message?: unknown }).message ?? "unknown")
        : JSON.stringify(payload.error);
      // reject() clears the timer + drops the registry entry.
      this.replayPending.reject(key, new Error(`${method} rejected: ${errMsg}`));
    } else {
      // settle() clears the timer + drops the registry entry.
      this.replayPending.settle(key, payload);
    }
    return true;
  }

  /**
   * Replay the outage queue after app-server reconnects.
   *
   * Messages are replayed in original order. We stored RAW TUI bytes
   * (pre-rewrite), so re-entering `onTuiMessage` produces a fresh proxy id
   * and a fresh `upstreamToClient` mapping against the new app-server
   * session — avoiding the response-tracking race where the original
   * mapping was wiped by `clearResponseTrackingState` on close.
   */
  private drainOutageQueue(): void {
    if (this.outageQueue.length === 0) {
      this.clearOutageTimer();
      return;
    }
    if (!this.appServerWs || this.appServerWs.readyState !== WebSocket.OPEN) return;
    const ws = this.tuiWs;
    if (!ws) {
      // TUI disconnected between buffer and drain — safe to discard.
      this.outageQueue = [];
      this.clearOutageTimer();
      return;
    }

    const messages = this.outageQueue;
    this.outageQueue = [];
    this.clearOutageTimer();
    this.log(
      `DIAGNOSTIC: replaying ${messages.length} buffered TUI messages after app-server reconnect`,
    );
    for (const msg of messages) {
      try {
        // Re-enter the full TUI message pipeline so rewriting and tracking
        // run against the fresh app-server session.
        this.onTuiMessage(ws, msg.raw);
      } catch (e: any) {
        this.log(`Failed to replay buffered TUI message (conn #${msg.connId}): ${e.message}`);
      }
    }
  }

  // ── Proxy Server ───────────────────────────────────────────

  private startProxy() {
    const self = this;
    this.proxyServer = Bun.serve({
      port: this.proxyPort,
      hostname: "127.0.0.1",
      fetch(req, server) {
        const url = new URL(req.url);
        const isUpgrade = req.headers.get("upgrade")?.toLowerCase() === "websocket";
        self.log(`HTTP ${req.method} ${url.pathname} (upgrade=${isUpgrade})`);
        if (url.pathname === "/healthz" || url.pathname === "/readyz") {
          // unix transport: appPort is a raw byte relay into Codex's unix
          // socket, which serves NO HTTP endpoints (verified fact in
          // codex-transport.ts) — forwarding would hang/500 and `abg codex`
          // would always die at waitForProxyReady. Answer from the adapter's
          // own upstream-connection state instead.
          if (self.transport === "unix") {
            const up = self.appServerWs?.readyState === WebSocket.OPEN;
            return new Response(up ? "ok" : "upstream not connected", { status: up ? 200 : 503 });
          }
          return fetch(`http://127.0.0.1:${self.appPort}${url.pathname}`);
        }
        // CSWSH guard: reject any WS upgrade carrying an Origin header (browser
        // page) before upgrading. The legitimate Codex TUI proxy client uses the
        // Bun global WebSocket and sends no Origin — empirically verified, see
        // ws-origin-guard.ts. The /healthz and /readyz GET endpoints above are
        // not gated (they are plain HTTP, not upgrades).
        if (isUpgrade && !isAllowedWsUpgrade(req)) {
          self.log("Rejected WS upgrade on proxy port: Origin header present (possible CSWSH)");
          return wsOriginRejectedResponse();
        }
        if (server.upgrade(req, { data: { connId: 0 } })) return undefined;
        self.log(`WARNING: non-upgrade HTTP request not handled: ${req.method} ${url.pathname}`);
        return new Response("AgentBridge Codex Proxy");
      },
      websocket: {
        open: (ws: ServerWebSocket<TuiSocketData>) => self.onTuiConnect(ws),
        close: (ws: ServerWebSocket<TuiSocketData>, code: number, reason: string) => {
          self.log(`WebSocket close event: conn #${ws.data.connId}, code=${code}, reason=${reason || "none"}`);
          self.onTuiDisconnect(ws);
        },
        message: (ws: ServerWebSocket<TuiSocketData>, msg) => self.onTuiMessage(ws, msg),
      },
    });
  }

  private onTuiConnect(ws: ServerWebSocket<TuiSocketData>) {
    const connId = ++this.connIdCounter;
    ws.data.connId = connId;

    // If a primary TUI connection is still alive, this is a secondary (picker)
    // connection. The Codex TUI opens a parallel WS for the resume picker while
    // the main session connection stays open. Give the secondary its own
    // app-server WS so the primary is not disturbed.
    if (this.tuiWs) {
      this.log(`Secondary TUI connected (conn #${connId}, primary is #${this.tuiConnId})`);
      this.setupSecondaryConnection(ws, connId);
      return;
    }

    // No active primary — this becomes the new primary connection
    const previousConnId = this.tuiConnId > 0 ? this.tuiConnId : null;
    this.tuiConnId = connId;
    this.tuiWs = ws;
    // Reset threadId to prevent premature message injection before TUI completes
    // its handshake (initialize → thread/start or thread/resume).
    this.threadId = null;
    this.log(`TUI connected (conn #${this.tuiConnId})`);
    this.emit("tuiConnected", this.tuiConnId);
    if (previousConnId !== null) {
      this.retireConnectionState(previousConnId);
    }
    // Do not replay pendingServerRequests here — defer until TUI completes its
    // handshake (thread/resume to matching thread, or thread/start which drops
    // orphan entries). See handleTrackedResponse and dropOrphanPendingRequests.
  }

  /** Set up a secondary (picker) connection with its own dedicated app-server WS. */
  private setupSecondaryConnection(ws: ServerWebSocket<TuiSocketData>, connId: number) {
    const appWs = new WebSocket(this.appServerUrl);
    // Store the appWs handle immediately (not just in onopen) so that
    // onTuiDisconnect can close it even if the picker disconnects before
    // the app-server WS finishes connecting.
    const entry = {
      tuiWs: ws,
      appServerWs: appWs,
      buffer: [] as string[],
      initialized: false,
      initializationReplayed: false,
    };
    this.secondaryConnections.set(connId, entry);

    appWs.onopen = () => {
      // Picker may have already disconnected — if so, close immediately
      if (!this.secondaryConnections.has(connId)) {
        appWs.close();
        return;
      }
      this.log(`Secondary conn #${connId}: app-server WS connected, flushing ${entry.buffer.length} buffered messages`);
      for (const msg of entry.buffer) {
        try { appWs.send(msg); } catch {}
      }
      entry.buffer = [];
    };
    appWs.onmessage = (event) => {
      if (!this.secondaryConnections.has(connId)) return;
      const data = typeof event.data === "string" ? event.data : event.data.toString();
      try { ws.send(data); } catch {}
    };
    appWs.onerror = () => {
      this.log(`Secondary conn #${connId}: app-server WS error`);
    };
    appWs.onclose = () => {
      this.log(`Secondary conn #${connId}: app-server WS closed`);
      const sec = this.secondaryConnections.get(connId);
      if (sec) {
        // App-server died — tear down the picker TUI socket too so it
        // doesn't become a zombie that silently drops messages.
        this.secondaryConnections.delete(connId);
        try { sec.tuiWs.close(); } catch {}
      }
    };
  }

  /**
   * Replay buffered server requests that belong to the given thread.
   *
   * Entries are replayed when their threadId matches `resumedThreadId`, or
   * when they have no threadId (fallback: cannot attribute, so assume they
   * belong to whatever the TUI is resuming). Non-matching entries survive
   * THIS function, but the thread/resume response handler then calls
   * dropOrphanPendingRequests(resumedThreadId), which deletes them — they do
   * NOT persist to a subsequent resume. (Their server ids would be stale on a
   * later session anyway; see the new-TUI initialize handling.)
   */
  private replayPendingForThread(resumedThreadId: string, ws: ServerWebSocket<TuiSocketData>) {
    const remaining: BufferedServerRequest[] = [];
    for (const buffered of this.pendingServerRequests) {
      const belongsToThread = buffered.threadId === null || buffered.threadId === resumedThreadId;
      if (!belongsToThread) {
        remaining.push(buffered);
        continue;
      }
      const proxyId = this.nextProxyId++;
      try {
        const parsed = JSON.parse(buffered.raw);
        parsed.id = proxyId;
        ws.send(JSON.stringify(parsed));
        this.serverRequestToProxy.set(proxyId, {
          raw: buffered.raw,
          serverId: buffered.serverId,
          connId: this.tuiConnId,
          method: buffered.method,
          timestamp: Date.now(),
          threadId: buffered.threadId,
        });
        if (buffered.threadId === null) {
          this.log(`WARNING: Replaying pending server request with unknown threadId (experimental fallback, may surface orphan UI on wrong thread): ${buffered.method} (server id=${buffered.serverId} → proxy id=${proxyId})`);
        } else {
          this.log(`Replayed buffered server request on thread/resume: ${buffered.method} (server id=${buffered.serverId} → proxy id=${proxyId}, threadId=${buffered.threadId})`);
        }
      } catch (e: any) {
        this.log(`Failed to replay buffered server request: ${buffered.method} (server id=${buffered.serverId}): ${e.message}`);
        remaining.push(buffered);
      }
    }
    this.pendingServerRequests = remaining;
  }

  /**
   * Drop buffered server requests that cannot be delivered because the TUI
   * has started a brand-new thread (thread/start), or has resumed a thread
   * that does not match the entry's threadId. The original listener in the
   * app-server is already dead for these entries, so keeping them would
   * just produce orphan approval UI on subsequent reconnects.
   */
  private dropOrphanPendingRequests(reason: string, matchThreadId: string | null = null) {
    if (this.pendingServerRequests.length === 0) return;
    const remaining: BufferedServerRequest[] = [];
    for (const buffered of this.pendingServerRequests) {
      const shouldDrop = matchThreadId === null
        ? true
        : buffered.threadId !== null && buffered.threadId !== matchThreadId;
      if (shouldDrop) {
        this.log(`Dropped orphan pending server request: ${buffered.method} (server id=${buffered.serverId}, threadId=${buffered.threadId ?? "unknown"}, reason=${reason})`);
        continue;
      }
      remaining.push(buffered);
    }
    this.pendingServerRequests = remaining;
  }

  private onTuiDisconnect(ws: ServerWebSocket<TuiSocketData>) {
    const connId = ws.data.connId;

    // Check if this is a secondary (picker) connection
    const secondary = this.secondaryConnections.get(connId);
    if (secondary) {
      this.log(`Secondary TUI disconnected (conn #${connId})`);
      this.secondaryConnections.delete(connId);
      if (secondary.appServerWs) {
        try { secondary.appServerWs.close(); } catch {}
      }
      return;
    }

    // Only clear tuiWs if this is still the current connection
    if (this.tuiWs === ws) {
      // Capture app-server state at the moment TUI disconnects — critical
      // for diagnosing "silent exit": did the TUI drop before or after we
      // lost app-server? This timestamp lines up with codex-wrapper.log's
      // exit line so we can reconstruct the sequence.
      const appServerOpen = this.appServerWs?.readyState === WebSocket.OPEN;
      this.log(
        `TUI disconnected (conn #${connId}, appServerOpen=${appServerOpen}, turnInProgress=${this.turnInProgress}, pendingTuiMessages=${this.pendingTuiMessages.length}, outageQueue=${this.outageQueue.length}, reconnectingForNewSession=${this.reconnectingForNewSession})`,
      );
      this.tuiWs = null;
      // Clear any pending reconnection state
      if (this.reconnectingForNewSession) {
        this.log("Clearing pending TUI message buffer (TUI disconnected during app-server reconnect)");
        this.pendingTuiMessages = [];
        this.reconnectingForNewSession = false;
      }
      // Clear outage buffer and timer — no TUI to replay to, and stale
      // messages would be wrong for whatever TUI reconnects next.
      if (this.outageQueue.length > 0 || this.outageTimer !== null) {
        this.log(
          `Clearing outage queue on TUI disconnect (buffered=${this.outageQueue.length})`,
        );
        this.outageQueue = [];
        this.clearOutageTimer();
      }
      this.emit("tuiDisconnected", connId);
    } else {
      this.log(`Stale TUI disconnected (conn #${connId}, current is #${this.tuiConnId})`);
    }
    this.retireConnectionState(connId);
  }

  private onTuiMessage(ws: ServerWebSocket<TuiSocketData>, msg: string | Buffer) {
    const data = typeof msg === "string" ? msg : msg.toString();
    const connId = ws.data.connId;

    // Route secondary (picker) connection messages to their own app-server WS
    const secondary = this.secondaryConnections.get(connId);
    if (secondary) {
      const method = this.detectJsonMethod(data);
      if (method === "initialize" || method === "initialized") {
        secondary.initialized = true;
      } else if (!secondary.initialized) {
        this.ensureSecondaryInitialized(secondary, connId);
      }
      this.sendOrBufferSecondary(secondary, data);
      return;
    }

    // Ignore messages from stale connections
    if (connId !== this.tuiConnId) {
      this.log(`Dropping message from stale TUI conn #${connId} (current is #${this.tuiConnId})`);
      return;
    }

    // Check if this is a response to a server-originated request
    try {
      const parsed = JSON.parse(data);
      if (parsed.id !== undefined && !parsed.method) {
        const normalizedId = this.normalizeNumericId(parsed.id);
        if (!isNaN(normalizedId) && this.pendingServerResponses.has(normalizedId)) {
          this.log(`Ignoring duplicate approval response while app-server reconnect is pending (proxy id=${normalizedId})`);
          return;
        }
        const pending = !isNaN(normalizedId) ? this.serverRequestToProxy.get(normalizedId) : undefined;
        if (pending !== undefined) {
          if (pending.connId !== connId) {
            this.log(`Dropping stale server request response (proxy id=${normalizedId}, expected conn #${pending.connId}, got #${connId})`);
            return;
          }
          parsed.id = pending.serverId;
          const forwardedResponse = JSON.stringify(parsed);
          if (!this.appServerWs || this.appServerWs.readyState !== WebSocket.OPEN) {
            this.bufferPendingServerResponse(normalizedId, pending, forwardedResponse, "app-server disconnected");
            return;
          }
          try {
            this.appServerWs.send(forwardedResponse);
            this.serverRequestToProxy.delete(normalizedId);
            this.log(`TUI → app-server: ${pending.method} response (proxy id=${normalizedId} → server id=${pending.serverId})`);
          } catch (e: any) {
            this.bufferPendingServerResponse(normalizedId, pending, forwardedResponse, `send failed: ${e.message}`);
          }
          return;
        }
      }
    } catch {}

    // Detect method before id-rewriting so we can intercept `initialize`
    // and buffer the RAW (un-rewritten) message for replay after reconnect.
    let detectedMethod: string | undefined;
    try {
      const parsed = JSON.parse(data);
      detectedMethod = typeof parsed.method === "string" ? parsed.method : undefined;
    } catch {}

    // When TUI sends `initialize`, reconnect app-server to give it a fresh
    // connection-scoped session (the upstream protocol requires per-connection init).
    // Buffer the raw message — id-rewriting happens after reconnect during replay.
    // Skip this check during replay to avoid infinite recursion.
    if (!this.replayingBufferedMessages) {
      if (detectedMethod === "initialize") {
        // Cache the raw payload for use by unintentional-reconnect replay.
        // Replays use the TUI's original id as-is — safe because the new
        // app-server session has no prior mapping for it.
        this.lastInitializeRaw = data;
        this.log("Detected initialize — reconnecting app-server for fresh session");
        this.reconnectingForNewSession = true;
        this.pendingTuiMessages = [data];
        this.reconnectAppServerForNewSession(ws);
        return;
      }

      // Buffer raw messages while reconnecting app-server for a fresh session
      if (this.reconnectingForNewSession) {
        this.pendingTuiMessages.push(data);
        return;
      }
    }

    // Cache `initialized` notification raw payload — used by
    // handleSessionRestoreAfterReconnect to recover session state after
    // an unintentional upstream reconnect.
    if (detectedMethod === "initialized") {
      this.lastInitializedRaw = data;
    }

    // Outage-path early return: buffer RAW before any id-rewriting or
    // upstreamToClient/serverRequestToProxy mutation. This avoids a
    // response-tracking race where `handleAppServerClose` clears mappings
    // that our buffered `forwarded` bytes would then point to. On drain
    // we re-enter this method with the raw bytes and do a fresh rewrite
    // against the new app-server session.
    //
    // We also treat `sessionRestoreInProgress` as an outage — while we're
    // replaying cached initialize / thread/resume, we must not let TUI
    // requests leak onto the still-uninitialized session.
    if (
      !this.appServerWs
      || this.appServerWs.readyState !== WebSocket.OPEN
      || this.sessionRestoreInProgress
    ) {
      if (this.tuiWs && this.tuiWs === ws) {
        this.bufferDuringOutage(ws, data);
      } else {
        this.log(
          `WARNING: non-primary TUI attempted to send while app-server down — dropped (connId=${connId})`,
        );
      }
      return;
    }

    let forwarded = data;
    try {
      const parsed = JSON.parse(data);
      const method = parsed.method ?? `response:${parsed.id}`;
      this.log(`TUI → app-server: ${method}`);

      // Rewrite request id to globally unique proxy id
      if (parsed.id !== undefined && parsed.method) {
        const proxyId = this.nextProxyId++;
        this.upstreamToClient.set(proxyId, { connId, clientId: parsed.id });
        this.trackPendingRequest(parsed, connId, proxyId);
        // P1 #5: remember initialize request proxy ids so the matching response
        // can be recognized in handleAppServerResponse and its InitializeResponse
        // captured (version / platform). Bounded: cleared on capture and on
        // response-tracking resets.
        if (parsed.method === "initialize") {
          this.pendingInitializeProxyIds.add(proxyId);
        }
        parsed.id = proxyId;
        forwarded = JSON.stringify(parsed);
      } else {
        this.trackPendingRequest(parsed, connId);
      }
    } catch {
      this.log(`TUI → app-server: (unparseable)`);
    }

    // Forward to app-server. The OPEN check above guarantees readyState was
    // OPEN when we started processing; the narrow race where upstream closes
    // between that check and here is handled here as a best-effort warning.
    if (this.appServerWs?.readyState === WebSocket.OPEN) {
      this.appServerWs.send(forwarded);
    } else {
      this.log(
        `WARNING: app-server closed between OPEN check and send — message lost (connId=${ws.data.connId})`,
      );
    }
  }

  private detectJsonMethod(raw: string): string | undefined {
    try {
      const parsed = JSON.parse(raw);
      return typeof parsed?.method === "string" ? parsed.method : undefined;
    } catch {
      return undefined;
    }
  }

  private ensureSecondaryInitialized(
    secondary: {
      appServerWs: WebSocket | null;
      buffer: string[];
      initialized: boolean;
      initializationReplayed: boolean;
    },
    connId: number,
  ) {
    if (secondary.initializationReplayed) return;
    secondary.initializationReplayed = true;

    if (!this.lastInitializeRaw) {
      this.log(`Secondary conn #${connId}: no cached initialize available before first non-initialize request`);
      return;
    }

    this.log(`Secondary conn #${connId}: replaying cached initialize before picker request`);
    this.sendOrBufferSecondary(secondary, this.lastInitializeRaw);
    if (this.lastInitializedRaw) {
      this.sendOrBufferSecondary(secondary, this.lastInitializedRaw);
    }
    secondary.initialized = true;
  }

  private sendOrBufferSecondary(
    secondary: {
      appServerWs: WebSocket | null;
      buffer: string[];
    },
    raw: string,
  ) {
    if (secondary.appServerWs && secondary.appServerWs.readyState === WebSocket.OPEN) {
      try { secondary.appServerWs.send(raw); } catch {}
    } else {
      secondary.buffer.push(raw);
    }
  }

  // ── Response Patching ──────────────────────────────────────

  private handleAppServerPayload(raw: string): string | null {
    try {
      const parsed: unknown = JSON.parse(raw);
      // #69: ANY inbound app-server message proves the turn is alive — refresh
      // the inactivity watchdog(s) here, at the single inbound funnel. This
      // MUST be broader than the modeled notification set: Codex streams many
      // notification types we deliberately do NOT forward (item/agentReasoning/
      // delta, item/commandExecution/*, item/fileChange/*, …). A long but
      // actively-streaming turn (e.g. a multi-minute build) emits only these
      // "unknown" notifications between turn/started and turn/completed; if we
      // refreshed only on the handful in isAppServerNotification() we would
      // mark a live turn stalled. Responses and server requests count as
      // activity too. (refreshTurnWatchdogs() no-ops when no turn is active.)
      if (typeof parsed === "object" && parsed !== null) {
        this.refreshTurnWatchdogs();
      }
      // Short-circuit: response carries an id that matches a pending
      // session-restore replay? Consume it and suppress forwarding to
      // the TUI (which didn't send the request and would be confused).
      if (typeof parsed === "object" && parsed !== null && "id" in parsed) {
        if (this.tryConsumeReplayResponse(parsed as Record<string, unknown>)) {
          return null;
        }
      }
      if (isAppServerNotification(parsed) || (typeof parsed === "object" && parsed !== null && !("id" in parsed))) {
        const notificationLike = parsed as Record<string, unknown>;
        // Sniff thread/closed → the ONLY signal that correlates with a
        // silent TUI exit (code 0, empty stderr, no ERROR line). Verified
        // via Codex's PTY reproduction: server emitting this notification
        // causes TUI to run ExitMode::Immediate with no output. We log a
        // high-signal line so codex-wrapper.log's `exit_0_empty_stderr`
        // classification can be paired up by timestamp with this event.
        if (notificationLike.method === "thread/closed") {
          const params = notificationLike.params as Record<string, unknown> | undefined;
          const threadId = typeof params?.threadId === "string" ? params.threadId : "unknown";
          this.log(
            `DIAGNOSTIC: app-server emitted thread/closed (threadId=${threadId}) — TUI will exit(0) silently`,
          );
        }
        const forwarded = this.patchResponse(notificationLike, raw);
        this.interceptServerMessage(notificationLike);
        return forwarded;
      }

      if (isAppServerRequestMessage(parsed)) {
        // Intentionally uses the broad isAppServerRequestMessage (any id+method) instead of
        // the narrow isAppServerServerRequest (only known approval methods). This ensures
        // unknown future server-to-client requests are forwarded rather than silently dropped
        // (lesson from issue #37).
        this.handleServerRequest(parsed, raw);
        return null;
      }

      if (isAppServerResponseMessage(parsed)) {
        return this.handleAppServerResponse(parsed, raw);
      }

      // Drop unclassifiable messages (not notification, not request, not response).
      this.log(`Dropping unclassifiable app-server message: ${raw.slice(0, 100)}`);
      return null;
    } catch {
      return raw;
    }
  }

  private handleServerRequest(parsed: AppServerRequest, raw: string): void {
    const serverId = parsed.id;
    const method = parsed.method;
    const threadId = this.extractThreadIdFromParams(parsed.params);

    if (!this.tuiWs) {
      this.pendingServerRequests.push({ raw, serverId, method, threadId });
      this.log(`Server request buffered (no TUI): ${method} (server id=${serverId}, threadId=${threadId ?? "unknown"})`);
      return;
    }

    const proxyId = this.nextProxyId++;
    parsed.id = proxyId;

    try {
      this.tuiWs.send(JSON.stringify(parsed));
    } catch (e: any) {
      this.log(`Server request send failed, buffering: ${method} (server id=${serverId}): ${e.message}`);
      this.pendingServerRequests.push({ raw, serverId, method, threadId });
      return;
    }

    this.serverRequestToProxy.set(proxyId, {
      raw,
      serverId,
      connId: this.tuiConnId,
      method,
      timestamp: Date.now(),
      threadId,
    });
    this.log(`Server request: ${method} (server id=${serverId} → proxy id=${proxyId}, conn #${this.tuiConnId}, threadId=${threadId ?? "unknown"})`);
  }

  private extractThreadIdFromParams(params: unknown): string | null {
    if (typeof params !== "object" || params === null) return null;
    const tid = (params as Record<string, unknown>).threadId;
    return typeof tid === "string" && tid.length > 0 ? tid : null;
  }

  /** Normalize a JSON-RPC id to a number. Returns NaN for non-numeric strings. */
  private normalizeNumericId(id: unknown): number {
    if (typeof id === "number") return id;
    if (typeof id === "string" && /^-?\d+$/.test(id)) return Number(id);
    return NaN;
  }

  private bufferPendingServerResponse(
    proxyId: number,
    pending: PendingServerRequest,
    forwardedResponse: string,
    reason: string,
  ) {
    this.pendingServerResponses.set(proxyId, {
      raw: forwardedResponse,
      serverId: pending.serverId,
      method: pending.method,
      timestamp: Date.now(),
    });
    this.serverRequestToProxy.delete(proxyId);
    // Honest wording: reconnect paths clear this buffer before any flush can
    // run, and server ids are session-scoped anyway — so this response is
    // best-effort at most and usually LOST. The old text ("buffered until
    // reconnect") sent incident debugging down a dead end.
    this.log(
      `Approval response could not reach the app-server (${reason}) — buffered best-effort, but it is ` +
        `likely lost (session-scoped id; reconnects clear this buffer). The TUI may need to re-approve. ` +
        `(proxy id=${proxyId} → server id=${pending.serverId})`,
    );
  }

  private flushPendingServerResponses() {
    if (!this.appServerWs || this.appServerWs.readyState !== WebSocket.OPEN) return;

    for (const [proxyId, pending] of this.pendingServerResponses.entries()) {
      try {
        this.appServerWs.send(pending.raw);
        this.pendingServerResponses.delete(proxyId);
        this.log(`Flushed buffered approval response after app-server reconnect (proxy id=${proxyId} → server id=${pending.serverId})`);
      } catch (e: any) {
        this.log(`Failed to flush buffered approval response (proxy id=${proxyId}): ${e.message}`);
        break;
      }
    }
  }

  private handleAppServerResponse(parsed: AppServerResponse, raw: string): string | null {
    const responseId = parsed.id;
    const numericId = this.normalizeNumericId(responseId);
    const mapping = !isNaN(numericId) ? this.upstreamToClient.get(numericId) : undefined;

    if (mapping) {
      this.upstreamToClient.delete(numericId);

      // P1 #5: capture the initialize response (version / platform) BEFORE the
      // staleness drop below — the server identity is connection-independent, so
      // even a response to a now-superseded conn still tells us the truth about
      // the app-server we're proxying. Always clears the pending-id bookkeeping.
      if (!isNaN(numericId) && this.pendingInitializeProxyIds.delete(numericId)) {
        this.captureAppServerInfo(parsed.result);
      }

      if (mapping.connId !== this.tuiConnId) {
        this.log(`Dropping stale response (upstream id ${responseId}, from conn #${mapping.connId}, current #${this.tuiConnId})`);
        return null;
      }

      parsed.id = mapping.clientId;
      this.log(`app-server → TUI: response (proxy id=${numericId} → client id=${String(mapping.clientId)}, conn #${mapping.connId})`);
      const forwarded = this.patchResponse(parsed, JSON.stringify(parsed));
      this.interceptServerMessage(parsed, mapping.connId);
      return forwarded;
    }

    const bridgeKind = !isNaN(numericId) ? this.consumeBridgeRequestId(numericId) : null;
    if (bridgeKind) {
      if (parsed.error) {
        this.log(`Bridge-originated ${bridgeKind} request failed (id ${responseId}): ${parsed.error.message ?? "unknown error"}`);
        if (bridgeKind === "steer") {
          // A rejected steer leaves the ORIGINAL turn running untouched —
          // no abort, no phase change. Surface it so the daemon can tell
          // Claude the mid-turn message did NOT reach Codex (e.g.
          // ActiveTurnNotSteerable on Review/Compact turns, or the turn ended
          // in the NoActiveTurn race window). The bridge requestId (numericId)
          // correlates this verdict back to its dispatch so the daemon can
          // release the steer's idempotency key + reply expectation by id
          // instead of relying on FIFO ordering (lost-response orphan guard).
          this.emit("steerFailed", { requestId: numericId, reason: parsed.error.message ?? "unknown error" });
        } else if (bridgeKind === "interrupt") {
          // A rejected turn/interrupt ("expected active turn id X but found Y",
          // "no active turn to interrupt") means the ORIGINAL turn keeps
          // running. Like steer rejections, the response handler must NOT
          // touch turn state — the notification flow owns it. Surface the
          // failure so the daemon can abort its terminal wait loudly.
          this.emit("interruptFailed", parsed.error.message ?? "unknown error");
        } else {
          // An injected turn/start was REJECTED before any turn/started — no
          // turn ever goes in-progress, so resetTurnState's wasInProgress-gated
          // turnAborted never fires. Signal the abort here so daemon-level
          // per-turn state (the require_reply tracker) is not stranded.
          // turnPhase must agree with the emitted event (PR A contract).
          // bridgeTurnRejected adds per-request correlation so the daemon can
          // mark the matching idempotency key terminal `rejected` (PR B).
          this.lastTurnEndedAbnormally = true;
          this.emit("turnAborted", `injected turn/start rejected: ${parsed.error.message ?? "unknown error"}`);
          this.emit("bridgeTurnRejected", {
            requestId: numericId,
            error: parsed.error.message ?? "unknown error",
          });
          this.notifyPhaseIfChanged();
        }
      } else {
        this.log(`Bridge-originated ${bridgeKind} request completed (id ${responseId})`);
        if (bridgeKind === "steer") {
          // Carry the bridge requestId so the daemon correlates the accept
          // back to its dispatch by id (same orphan guard as steerFailed).
          this.emit("steerAccepted", { requestId: numericId });
        } else if (bridgeKind === "turn-start") {
          // turn_started ACK correlation (PR B): the turn/start response
          // carries the created turn (TurnStartResponse.turn.id). Emit the
          // correlation so the daemon can send the turn_started control event
          // and advance the idempotency machine to started(turnId).
          const result = parsed.result as { turn?: { id?: unknown } } | undefined;
          const turnId = result?.turn?.id;
          if (typeof turnId === "string" && turnId.length > 0) {
            this.emit("bridgeTurnStarted", { requestId: numericId, turnId });
          } else {
            this.log(
              `Bridge-originated turn/start response carried no turn id (id ${responseId}) — turn_started ACK skipped`,
            );
          }
        }
        // interrupt success (TurnInterruptResponse is {}): log only — the
        // app-server defers it until TurnAborted, and the terminal
        // turn/completed notification is what drives state.
      }
      return null;
    }

    if (!isNaN(numericId) && this.consumeStaleProxyId(numericId)) {
      this.log(`Dropping stale response for retired upstream id ${responseId}`);
      return null;
    }

    this.log(`Dropping unmatched app-server response id ${String(responseId)}`);
    return null;
  }

  /**
   * P1 #5: capture the app-server identity from an `initialize` response result
   * (version / platform) and expose it on the status getter. Logs the captured
   * identity, and WARNs once-per-distinct-version when the userAgent is missing
   * or its version is unparseable — a signal the app-server drifted from the
   * shape these intercept points were written against (see the class-level
   * version-coupling checklist).
   *
   * Behavior-preserving: capture-only, never mutates the forwarded payload.
   */
  private captureAppServerInfo(result: unknown): void {
    const init = (typeof result === "object" && result !== null ? result : {}) as AppServerInitializeResponse;
    const userAgent = typeof init.userAgent === "string" ? init.userAgent : null;
    const version = parseAppServerVersion(userAgent);
    const info: AppServerInfo = {
      version,
      userAgent,
      platformFamily: typeof init.platformFamily === "string" ? init.platformFamily : null,
      platformOs: typeof init.platformOs === "string" ? init.platformOs : null,
    };
    this.appServerInfo = info;
    this.log(
      `Captured app-server initialize: version=${version ?? "unknown"} ` +
        `userAgent=${userAgent ?? "none"} platform=${info.platformOs ?? "?"}/${info.platformFamily ?? "?"}`,
    );
    if (version === null) {
      // Key the at-most-once dedup on the raw userAgent (or a sentinel) so a
      // stable-but-unparseable server warns once, not on every reconnect.
      const dedupKey = userAgent ?? "<missing-userAgent>";
      if (!this.warnedAppServerVersions.has(dedupKey)) {
        this.warnedAppServerVersions.add(dedupKey);
        this.log(
          `WARNING: app-server initialize response carried no parseable version ` +
            `(userAgent=${userAgent ?? "missing"}). The proxy's intercept points assume a ` +
            `known protocol snapshot — verify the version-coupling checklist if Codex was upgraded.`,
        );
      }
    }
  }

  private patchResponse(parsed: AppServerNotification | AppServerResponse | Record<string, unknown>, raw: string): string {
    if (isAppServerResponseMessage(parsed) && parsed.error && parsed.id !== undefined) {
      const errMsg: string = parsed.error.message ?? "";
      const errCode = parsed.error.code;
      const textMatchesRateLimit = errMsg.includes("rate limits") || errMsg.includes("rateLimits");
      // P1 #5: structured-code-FIRST. The app-server's rate-limit read path uses
      // GENERIC JSON-RPC error codes (no rate-limit-specific code exists —
      // codex-rs error_code.rs + account_processor.rs:967-990), so the
      // structured signal we DO have is "the error code is one of the recognized
      // codes that path emits". When the code is recognized AND the text agrees,
      // we patch via the structured path with NO fragile-match warning. When the
      // code is absent/unrecognized but the legacy text still matches, we keep
      // patching (behavior-preserving) but log a once-per-message 'fragile-match'
      // warning so an upstream wording change becomes visible instead of silent.
      const codeRecognized =
        typeof errCode === "number" && APP_SERVER_RATE_LIMIT_ERROR_CODES.has(errCode);
      const structuredMatch = codeRecognized && textMatchesRateLimit;

      if (structuredMatch || textMatchesRateLimit) {
        if (structuredMatch) {
          this.log(`Patching rateLimits error → mock success via structured code ${errCode} (id: ${parsed.id})`);
        } else {
          this.warnFragileRateLimitMatch(errMsg, errCode);
          this.log(`Patching rateLimits error → mock success via fragile text fallback (id: ${parsed.id})`);
        }
        return JSON.stringify({
          id: parsed.id,
          result: {
            rateLimits: {
              limitId: null,
              limitName: null,
              primary: { usedPercent: 0, windowDurationMins: 60, resetsAt: null },
              secondary: null,
              credits: null,
              planType: null,
            },
            rateLimitsByLimitId: null,
          },
        });
      }
      // "Already initialized" is no longer patched — we reconnect the app-server
      // WS on each TUI `initialize` request, giving it a fresh connection scope.
    }
    return raw;
  }

  /**
   * P1 #5: emit the 'fragile-match' warning at most once per distinct error
   * message, so a repeated legacy rate-limit error does not spam the log.
   */
  private warnFragileRateLimitMatch(errMsg: string, errCode: number | undefined): void {
    if (this.warnedFragileRateLimitMessages.has(errMsg)) return;
    this.warnedFragileRateLimitMessages.add(errMsg);
    this.log(
      `WARNING: fragile-match — patched a rate-limit error by human-readable text ` +
        `(code=${errCode ?? "none"} not in the recognized set). If Codex changed the ` +
        `error wording or code, update patchResponse / APP_SERVER_RATE_LIMIT_ERROR_CODES. ` +
        `Message: ${errMsg.slice(0, 120)}`,
    );
  }

  // ── Server Message Interception (for Bridge) ───────────────

  private interceptServerMessage(msg: AppServerNotification | AppServerResponse | Record<string, unknown>, connId?: number) {
    this.handleTrackedResponse(msg, connId);
    if ("method" in msg && typeof msg.method === "string" && isAppServerNotification(msg)) {
      this.handleServerNotification(msg);
    }
  }

  private handleServerNotification(msg: AppServerNotification) {
    const { method, params } = msg;
    // Note: the inactivity watchdog is refreshed for ALL inbound messages at
    // the handleAppServerPayload funnel (see #69 there), not here — this method
    // only runs for the modeled subset, which would miss the long-build case.
    switch (method) {
      case "turn/started":
        this.markTurnStarted(params?.turn?.id);
        break;
      case "item/started": {
        const item: AppServerItem | undefined = params?.item;
        if (item?.type === "agentMessage") this.agentMessageBuffers.set(item.id, []);
        break;
      }
      case "item/agentMessage/delta": {
        const itemId = params?.itemId;
        if (typeof itemId !== "string") break;
        const buf = this.agentMessageBuffers.get(itemId);
        if (buf && params?.delta) buf.push(params.delta);
        break;
      }
      case "item/completed": {
        const item: AppServerItem | undefined = params?.item;
        if (item?.type === "agentMessage") {
          const content = this.extractContent(item);
          this.agentMessageBuffers.delete(item.id);
          if (content) {
            this.log(`Agent message completed (${content.length} chars)`);
            this.emit("agentMessage", {
              id: item.id, source: "codex" as const, content, timestamp: Date.now(),
            } satisfies BridgeMessage);
          }
        }
        break;
      }
      case "turn/completed": {
        const wasInProgress = this.turnInProgress;
        this.markTurnCompleted(params?.turn?.id);
        // Only emit when all turns are done (symmetric with turnStarted)
        if (wasInProgress && !this.turnInProgress) {
          this.emit("turnCompleted");
        }
        break;
      }
    }
  }

  private extractContent(item: AppServerItem): string {
    if (item.content?.length) {
      return item.content.filter((c) => c.type === "text" && c.text).map((c) => c.text!).join("");
    }
    return this.agentMessageBuffers.get(item.id)?.join("") ?? "";
  }

  /** Build a generation-scoped key: "connId:rpcId" to prevent cross-reconnect collisions. */
  private pendingKey(rpcId: unknown, connId?: number): string | null {
    const base = this.requestKey(rpcId);
    if (!base) return null;
    return `${connId ?? this.tuiConnId}:${base}`;
  }

  private trackPendingRequest(message: AppServerRequest | Record<string, unknown>, connId: number, _proxyId?: number) {
    const rpcId = "id" in message ? message.id : undefined;
    const method = "method" in message && typeof message.method === "string" ? message.method : undefined;
    const key = this.pendingKey(rpcId, connId);

    if (!key || !isTrackedAppServerRequestMethod(method)) return;

    const pending: PendingRequest = { method };
    if (method === "turn/start") {
      const params = "params" in message && typeof message.params === "object" && message.params !== null
        ? message.params as Record<string, unknown>
        : undefined;
      const threadId = params?.threadId;
      if (typeof threadId === "string" && threadId.length > 0) {
        pending.threadId = threadId;
      }
    }
    // #70: stamp explicit thread switches (thread/start, thread/resume) with the
    // latest-issued sequence so an out-of-order OLD response can't clobber the
    // thread the user most recently switched to.
    if (method === "thread/start" || method === "thread/resume") {
      pending.threadSwitchSeq = ++this.threadSwitchSeq;
    }

    if (this.pendingRequests.has(key)) {
      this.log(`WARNING: overwriting pending request for key ${key}`);
    }

    this.pendingRequests.set(key, pending);
  }

  // TODO: narrow this type further once response shapes are fully typed (#47 follow-up)
  private handleTrackedResponse(message: any, connId?: number) {
    const key = this.pendingKey(message?.id, connId);
    if (!key) return;

    const pending = this.pendingRequests.get(key);
    if (!pending) {
      // Log responses that have result.thread.id for debugging
      if (message?.result?.thread?.id) {
        this.log(`[track-resp] Unmatched response with thread.id=${message.result.thread.id}, key=${key}, pending keys=[${[...this.pendingRequests.keys()].join(",")}]`);
      }
      return;
    }

    this.pendingRequests.delete(key);

    if (message?.error) {
      this.log(
        `Tracked request failed (${pending.method}, id ${key}): ${message.error.message ?? "unknown error"}`,
      );
      return;
    }

    switch (pending.method) {
      case "thread/start": {
        // #70: ignore an out-of-order OLD thread switch. Only the response to
        // the most-recently-issued thread/start|resume may mutate activeThreadId
        // or drop orphans — a stale one would clobber the thread the user just
        // switched to and re-orphan its in-flight approvals.
        if (!this.isLatestThreadSwitch(pending)) {
          this.log(
            `Ignoring stale thread/start response ${key} (seq=${pending.threadSwitchSeq} < latest=${this.threadSwitchSeq})`,
          );
          break;
        }
        const threadId = message?.result?.thread?.id;
        if (typeof threadId === "string" && threadId.length > 0) {
          this.setActiveThreadId(threadId, `thread/start response ${key}`);
        }
        // User started a brand-new session — any buffered server requests
        // belong to a thread the user has abandoned. Drop them so they do
        // not surface as orphan approval UI.
        this.dropOrphanPendingRequests(`thread/start (new session)`);
        break;
      }
      case "thread/resume": {
        // #70: same staleness guard as thread/start — a late resume response
        // must not steal activeThreadId back from a newer switch.
        if (!this.isLatestThreadSwitch(pending)) {
          this.log(
            `Ignoring stale thread/resume response ${key} (seq=${pending.threadSwitchSeq} < latest=${this.threadSwitchSeq})`,
          );
          break;
        }
        const threadId = message?.result?.thread?.id;
        if (typeof threadId === "string" && threadId.length > 0) {
          this.setActiveThreadId(threadId, `thread/resume response ${key}`);
          if (this.tuiWs) {
            this.replayPendingForThread(threadId, this.tuiWs);
          }
          // Drop any entries that belong to a different thread than the
          // resumed one — the user did not return to them.
          this.dropOrphanPendingRequests(`thread/resume to ${threadId}`, threadId);
        }
        break;
      }
      case "turn/start":
        // #70: a turn/start response carries the threadId the turn ran against.
        // Only adopt it when it agrees with the active thread (or none is set
        // yet). If the user has already switched threads, a late turn/start must
        // not drag activeThreadId back to the abandoned thread.
        if (pending.threadId) {
          if (this.threadId === null || this.threadId === pending.threadId) {
            this.setActiveThreadId(pending.threadId, `turn/start response ${key}`);
          } else {
            this.log(
              `Ignoring turn/start response ${key} threadId=${pending.threadId} (active thread is ${this.threadId})`,
            );
          }
        }
        break;
    }
  }

  /**
   * #70: true only if this pending request is the most-recently-issued explicit
   * thread switch (thread/start | thread/resume). Out-of-order responses to
   * superseded switches return false and must not mutate activeThreadId.
   */
  private isLatestThreadSwitch(pending: PendingRequest): boolean {
    return pending.threadSwitchSeq === this.threadSwitchSeq;
  }

  private setActiveThreadId(threadId: string, reason: string) {
    if (this.threadId === threadId) return;

    const previousThreadId = this.threadId;
    this.threadId = threadId;
    this.emit("threadChanged", { threadId, previousThreadId, reason });

    if (previousThreadId) {
      this.log(`Active thread changed: ${previousThreadId} → ${threadId} (${reason})`);
      return;
    }

    this.log(`Thread detected: ${threadId} (${reason})`);
    this.emit("ready", threadId);
  }

  /**
   * The most recently started REAL active turn id — the value turn/steer must
   * pass as expectedTurnId. `unknown:` fallback keys (turn/started without an
   * id) are not addressable and are skipped. Insertion order of the Set makes
   * the last real entry the newest turn.
   */
  private currentSteerableTurnId(): string | null {
    let newest: string | null = null;
    for (const id of this.activeTurnIds) {
      if (!id.startsWith("unknown:")) newest = id;
    }
    return newest;
  }

  /**
   * Public view of the steer target (PR B): the daemon reads it BEFORE calling
   * steerMessage so an idempotency key dispatched as a steer can be bound to
   * the turn it joined (started(turnId)) without changing steerMessage's
   * boolean contract.
   */
  get steerableTurnId(): string | null {
    return this.currentSteerableTurnId();
  }

  /**
   * Turn lifecycle phase (protocol v2 PR A). Strictly the turn axis — the
   * bridge's attention/routing window is reported separately by the daemon.
   */
  get turnPhase(): "idle" | "running" | "stalled" | "aborted" {
    if (this.activeTurnIds.size > 0) {
      const allStalled = [...this.activeTurnIds].every((id) => this.currentlyStalledTurnIds.has(id));
      return allStalled ? "stalled" : "running";
    }
    return this.lastTurnEndedAbnormally ? "aborted" : "idle";
  }

  /** Emit turnPhaseChanged when the derived phase moved — single funnel for
   *  ALL transitions so status persistence cannot drift from /healthz. */
  private notifyPhaseIfChanged() {
    const phase = this.turnPhase;
    if (phase === this.lastEmittedPhase) return;
    const previous = this.lastEmittedPhase;
    this.lastEmittedPhase = phase;
    this.emit("turnPhaseChanged", { phase, previous });
  }

  private markTurnStarted(turnId?: string) {
    const wasInProgress = this.turnInProgress;
    // Compute the key ONCE and use it for both the set and the watchdog so the
    // two never diverge (incl. the no-turn-id `unknown:` fallback).
    const turnKey = typeof turnId === "string" && turnId.length > 0 ? turnId : `unknown:${Date.now()}`;
    // delete-before-add keeps Set insertion order = recency even if the
    // app-server re-emits turn/started for an already-active id —
    // currentSteerableTurnId relies on "last entry = newest turn".
    this.activeTurnIds.delete(turnKey);
    this.activeTurnIds.add(turnKey);
    // A new turn must be eligible to notify on stall even if a prior turn
    // reused the same id (e.g. the `unknown:` fallback is time-stamped, but
    // real ids could repeat across reconnects).
    this.stalledTurnIds.delete(turnKey);
    this.currentlyStalledTurnIds.delete(turnKey);
    this.lastTurnEndedAbnormally = false;
    this.scheduleTurnWatchdog(turnKey);

    this.turnInProgress = this.activeTurnIds.size > 0;
    if (!wasInProgress && this.turnInProgress) {
      this.emit("turnStarted");
    }
    this.notifyPhaseIfChanged();
  }

  private markTurnCompleted(turnId?: string) {
    const completedId = typeof turnId === "string" && turnId.length > 0 ? turnId : null;
    if (completedId !== null) {
      // LOW-7: symmetric eviction. A turn STARTED without an id is keyed
      // `unknown:<ts>`; its COMPLETED event may carry a real id. Evicting only
      // the real id strands the placeholder, pinning turnInProgress=true forever
      // (every later reply gets busy-rejected). Conservatively reclaim a SINGLE
      // dangling `unknown:` placeholder here — only when exactly one exists, so a
      // legitimate concurrent real turn is never collateral. We reclaim whether
      // or not the real id matched: if the id-keyed turn existed it was a normal
      // completion, and any lone placeholder still belongs to the
      // started-without-id pair this completion closes out.
      const idWasTracked = this.activeTurnIds.has(completedId);
      this.activeTurnIds.delete(completedId);
      this.clearTurnWatchdog(completedId);
      this.stalledTurnIds.delete(completedId);
      this.currentlyStalledTurnIds.delete(completedId);

      // Reclaim the placeholder ONLY when the completed real id was NOT already
      // tracked: that means the matching turn/started carried no id and was
      // keyed `unknown:` — this completion closes it out. If the id WAS tracked
      // it was a normal id-keyed turn, so any lingering placeholder belongs to a
      // different, possibly still-live, started-without-id turn — leave it for
      // the watchdog / reset path. Require exactly one placeholder so a real turn
      // is never collateral when correlation is ambiguous.
      if (!idWasTracked) {
        const placeholders = [...this.activeTurnIds].filter((id) => id.startsWith("unknown:"));
        if (placeholders.length === 1) {
          const placeholder = placeholders[0]!;
          this.activeTurnIds.delete(placeholder);
          this.clearTurnWatchdog(placeholder);
          this.stalledTurnIds.delete(placeholder);
          this.currentlyStalledTurnIds.delete(placeholder);
        }
      }
    } else {
      this.activeTurnIds.clear();
      this.clearAllTurnWatchdogs();
      this.stalledTurnIds.clear();
      this.currentlyStalledTurnIds.clear();
      // A completion with no turn id clears ALL active turns, so any
      // agentMessage still mid-stream will never see its item/completed —
      // drop the dangling buffers instead of leaking them.
      this.agentMessageBuffers.clear();
    }

    this.lastTurnEndedAbnormally = false;
    this.turnInProgress = this.activeTurnIds.size > 0;
    // Per-turn completion signal (PR B): unlike the aggregate "turnCompleted"
    // (which fires only when ALL turns are done), this carries the specific
    // turn id so the daemon can terminate the matching idempotency key and
    // waitForTurnsTerminal can re-check after EVERY completion. null = the
    // notification carried no id and ALL active turns were cleared.
    this.emit("turnIdCompleted", completedId);
    this.notifyPhaseIfChanged();
  }

  /** Inactivity timeout for a turn (env-overridable; default 5 minutes). */
  private turnWatchdogMs(): number {
    const v = Number(process.env.AGENTBRIDGE_TURN_WATCHDOG_MS);
    return Number.isFinite(v) && v > 0 ? v : 300_000;
  }

  /** (Re)arm the inactivity watchdog for a turn. Replaces any existing timer. */
  private scheduleTurnWatchdog(turnKey: string) {
    this.clearTurnWatchdog(turnKey);
    const timer = setTimeout(() => {
      if (!this.activeTurnIds.has(turnKey)) return;
      this.log(
        `WARNING: turn ${turnKey} watchdog fired after ${this.turnWatchdogMs()}ms of inactivity — ` +
        `marking stalled but keeping Codex busy until a real completion or reconnect`,
      );
      this.markTurnStalled(turnKey);
    }, this.turnWatchdogMs());
    // Never keep the process/test runner alive on account of a watchdog.
    timer.unref?.();
    this.turnWatchdogs.set(turnKey, timer);
  }

  private clearTurnWatchdog(turnKey: string) {
    const timer = this.turnWatchdogs.get(turnKey);
    if (timer) {
      clearTimeout(timer);
      this.turnWatchdogs.delete(turnKey);
    }
  }

  private clearAllTurnWatchdogs() {
    for (const timer of this.turnWatchdogs.values()) clearTimeout(timer);
    this.turnWatchdogs.clear();
  }

  /**
   * Refresh every active turn's watchdog — called on any app-server notification
   * so the watchdog measures INACTIVITY, not total turn duration (a genuinely
   * long but active turn keeps streaming events and is never marked stalled).
   */
  private refreshTurnWatchdogs() {
    if (this.turnWatchdogs.size === 0) return;
    for (const turnKey of [...this.turnWatchdogs.keys()]) {
      this.scheduleTurnWatchdog(turnKey);
    }
    // Activity arrived — any turn previously marked stalled is moving again.
    // (stalledTurnIds, the notify-once set, intentionally stays.)
    this.currentlyStalledTurnIds.clear();
    this.notifyPhaseIfChanged();
  }

  /** Mark a turn stalled without clearing busy state (watchdog expiry path). */
  private markTurnStalled(turnKey: string) {
    if (!this.activeTurnIds.has(turnKey)) return;
    this.turnInProgress = true;
    this.currentlyStalledTurnIds.add(turnKey);
    // Phase notify must precede the at-most-once dedup return below: a turn
    // that stalls, resumes, then stalls AGAIN re-enters the stalled phase but
    // is suppressed from a second turnStalled notification — the phase event
    // must still fire or status persistence drifts from the getter.
    this.notifyPhaseIfChanged();
    // Emit at most once per turn: a stalled turn keeps its watchdog armed, so
    // later notifications can re-fire it. Suppress repeats for the same turnId.
    if (this.stalledTurnIds.has(turnKey)) return;
    this.stalledTurnIds.add(turnKey);
    this.emit("turnStalled", {
      turnId: turnKey,
      inactivityMs: this.turnWatchdogMs(),
    });
  }

  /**
   * Clear all turn state (active ids + watchdogs) and recompute busy. Single
   * source of truth used by app-server close / disconnect / stop so no path
   * clears the ids without also clearing the watchdog timers.
   */
  private resetTurnState(reason: string, emitCompleted = false) {
    const wasInProgress = this.turnInProgress;
    this.activeTurnIds.clear();
    this.clearAllTurnWatchdogs();
    this.stalledTurnIds.clear();
    this.currentlyStalledTurnIds.clear();
    // app-server close / reconnect / stop interrupts any in-flight agentMessage
    // stream: the item/completed that would normally delete its buffer never
    // arrives, so clear here (single reset boundary) to prevent unbounded growth
    // across the daemon's long lifetime. Matches the per-turn map discipline.
    this.agentMessageBuffers.clear();
    this.turnInProgress = false;
    if (wasInProgress) {
      this.lastTurnEndedAbnormally = !emitCompleted;
      if (emitCompleted) {
        this.emit("turnCompleted");
      } else {
        // The turn ended WITHOUT a normal completion (app-server close / reconnect
        // / stop). Signal an abort so daemon-level per-turn state (e.g. the
        // require_reply tracker) can be cleared instead of stranded onto a later
        // turn — turnCompleted is the ONLY other reset signal the daemon gets.
        this.emit("turnAborted", reason);
      }
      this.log(`Turn state reset (${reason})`);
    }
    this.notifyPhaseIfChanged();
    // Unconditional (NOT wasInProgress-gated) reset signal (PR B): an
    // app-server close / reconnect / stop also invalidates bridge requests
    // that were transport-accepted but never answered — the daemon must clear
    // its per-injection correlation state and terminate idempotency keys even
    // when no turn/started ever arrived for them.
    this.emit("turnTrackingReset", reason);
  }

  private requestKey(id: unknown): string | null {
    if (typeof id === "number" || typeof id === "string") return String(id);
    return null;
  }

  private retireConnectionState(connId: number) {
    const prefix = `${connId}:`;
    for (const key of this.pendingRequests.keys()) {
      if (key.startsWith(prefix)) this.pendingRequests.delete(key);
    }

    for (const [upId, mapping] of this.upstreamToClient.entries()) {
      if (mapping.connId !== connId) continue;
      this.upstreamToClient.delete(upId);
      this.trackStaleProxyId(upId);
    }

    const requeuedServerRequests: BufferedServerRequest[] = [];
    for (const [proxyId, pending] of this.serverRequestToProxy.entries()) {
      if (pending.connId === connId) {
        this.serverRequestToProxy.delete(proxyId);
        requeuedServerRequests.push({
          raw: pending.raw,
          serverId: pending.serverId,
          method: pending.method,
          threadId: pending.threadId,
        });
        this.log(`Requeued in-flight server request after TUI disconnect (proxy id=${proxyId}, server id=${pending.serverId}, method=${pending.method}, threadId=${pending.threadId ?? "unknown"})`);
      }
    }

    if (requeuedServerRequests.length === 0) return;

    this.pendingServerRequests.push(...requeuedServerRequests);

    // Do not replay immediately even if a new primary is already connected.
    // Replay is deferred until the new TUI completes thread/resume to a
    // matching thread — see handleTrackedResponse.
  }

  private trackStaleProxyId(proxyId: number) {
    this.clearTrackedId(this.staleProxyIds, proxyId);

    const timer = setTimeout(() => {
      this.staleProxyIds.delete(proxyId);
    }, CodexAdapter.RESPONSE_TRACKING_TTL_MS);
    timer.unref?.();
    this.staleProxyIds.set(proxyId, timer);
  }

  private consumeStaleProxyId(proxyId: number) {
    return this.clearTrackedId(this.staleProxyIds, proxyId);
  }

  private trackBridgeRequestId(requestId: number, kind: "turn-start" | "steer" | "interrupt" = "turn-start") {
    this.clearTrackedId(this.bridgeRequestIds, requestId);

    const timer = setTimeout(() => {
      this.bridgeRequestIds.delete(requestId);
      this.bridgeRequestKinds.delete(requestId);
    }, CodexAdapter.RESPONSE_TRACKING_TTL_MS);
    timer.unref?.();
    this.bridgeRequestIds.set(requestId, timer);
    this.bridgeRequestKinds.set(requestId, kind);
  }

  /** Returns the request kind when this id was bridge-originated, else null. */
  private consumeBridgeRequestId(requestId: number): "turn-start" | "steer" | "interrupt" | null {
    const kind = this.bridgeRequestKinds.get(requestId) ?? "turn-start";
    this.bridgeRequestKinds.delete(requestId);
    return this.clearTrackedId(this.bridgeRequestIds, requestId) ? kind : null;
  }

  private untrackBridgeRequestId(requestId: number) {
    this.bridgeRequestKinds.delete(requestId);
    this.clearTrackedId(this.bridgeRequestIds, requestId);
  }

  private clearTrackedId(store: Map<number, ReturnType<typeof setTimeout>>, id: number) {
    const timer = store.get(id);
    if (!timer) return false;
    clearTimeout(timer);
    store.delete(id);
    return true;
  }

  private clearTransientResponseTrackingState() {
    this.pendingRequests.clear();
    this.upstreamToClient.clear();
    // P1 #5: pending initialize ids share upstreamToClient's lifecycle — a proxy
    // id dropped from upstreamToClient can never match a response, so clear here
    // to avoid an unbounded leak of orphaned ids. The captured appServerInfo
    // itself PERSISTS (it is a fact about the server, not a transient mapping).
    this.pendingInitializeProxyIds.clear();

    for (const timer of this.staleProxyIds.values()) {
      clearTimeout(timer);
    }
    this.staleProxyIds.clear();

    for (const timer of this.bridgeRequestIds.values()) {
      clearTimeout(timer);
    }
    this.bridgeRequestIds.clear();
    // The kinds map shares its lifecycle with bridgeRequestIds — clearing one
    // without the other leaks kind entries forever, because the TTL timer that
    // would have deleted them was just cancelled above.
    this.bridgeRequestKinds.clear();
  }

  private clearResponseTrackingState() {
    this.clearTransientResponseTrackingState();
    this.serverRequestToProxy.clear();
    this.pendingServerRequests = [];
    this.pendingServerResponses.clear();
  }

  /**
   * Variant used by the intentional app-server reconnect triggered by a TUI
   * `initialize` message. Preserves `pendingServerRequests` so they can be
   * replayed after the TUI completes `thread/resume`. In-flight server
   * requests are moved into the pending buffer so they get the same
   * replay-on-resume treatment. The serverIds are stale with respect to
   * the new app-server session — the subsequent response will most likely
   * be rejected upstream, which is precisely what the A' experiment tests.
   */
  private clearResponseTrackingStateForAppServerReconnect() {
    this.clearTransientResponseTrackingState();
    for (const pending of this.serverRequestToProxy.values()) {
      this.pendingServerRequests.push({
        raw: pending.raw,
        serverId: pending.serverId,
        method: pending.method,
        threadId: pending.threadId,
      });
      this.log(`Requeued in-flight server request on app-server reconnect (server id=${pending.serverId}, method=${pending.method}, threadId=${pending.threadId ?? "unknown"})`);
    }
    this.serverRequestToProxy.clear();
    // Intentionally preserve pendingServerRequests across the reconnect.
    this.pendingServerResponses.clear();
  }

  /**
   * Build the lsof command used to find LISTENing processes on a TCP port.
   * Kept as the POSIX single source of truth (delegates to port-cleanup.ts);
   * see portPidsCommand for why the LISTEN restriction matters.
   *
   * Platform is hardcoded to "linux" on purpose: this method exists only for
   * the POSIX lsof test fixture (string-equality asserted in tests). The
   * production path is checkPorts(), which uses process.platform.
   */
  static buildPortListenLsofCommand(port: number): string {
    const { cmd, args } = portPidsCommand(port, "linux");
    return [cmd, ...args].join(" ");
  }

  /**
   * Clean up stale ports before starting.
   * Only kills `codex app-server` processes (our own spawns). If the port is
   * occupied by something else, throws with a clear message.
   *
   * Platform-aware (issue #76): POSIX uses lsof/ps/kill, win32 uses
   * PowerShell — previously the lsof failure on Windows was swallowed and the
   * spawn ran straight into "os error 10048".
   */
  private async checkPorts() {
    await cleanupPorts({
      ports: [
        { port: this.appPort, envVar: "CODEX_WS_PORT" },
        { port: this.proxyPort, envVar: "CODEX_PROXY_PORT" },
      ],
      run: ({ cmd, args }) => execFileSync(cmd, args, { encoding: "utf-8" }),
      log: (message) => this.log(message),
      sleep: (ms) => new Promise((r) => setTimeout(r, ms)),
    });
  }

  private log(msg: string) {
    this.logger.log(msg);
  }
}
