// RFC-030 Phase 0 — Minimal bridge for the shared-thread PoC.
//
// The bridge:
//   - Owns a single CodexAppServerClient connection.
//   - Binds one persistent thread (via `thread/resume`).
//   - Maps each Agent Network task_id ↔ one Codex `turnId`.
//   - Filters incoming events to the configured thread and `pendingTurns`
//     the bridge itself created (per RFC §7.5).
//   - Records `waiting_human` when a reverse request arrives, WITHOUT
//     responding — approvals belong to the human TUI (§7.6).
//   - Emits `task_reply` with the final agent text mapped back to the
//     originating `task_id` when its turn completes.
//
// Scope discipline (通信龙): Phase 0 is a PoC only. This bridge deliberately
// does NOT:
//   - Persist a durable ledger (§9 is Phase 2).
//   - Reconnect on transport loss (Phase 1 will add `thread/resume` retry).
//   - Wire into CommHub / cli.ts (§8.4 is Phase 1).
//   - Steer / cancel / interrupt (§7.4 is future scope).
//   - Run a tool proxy (§8.6 later).

import { EventEmitter } from "events";
import { CodexAppServerClient } from "./codex-app-server-client";

// ────────────────────────────────────────────────────────────────────────────
// Public shapes
// ────────────────────────────────────────────────────────────────────────────

export interface CodexAppServerBridgeOptions {
  client: CodexAppServerClient;
  /**
   * Persistent thread ID this bridge binds to (from node config). Leave
   * empty/undefined to have `bootstrap()` create a fresh thread and adopt
   * its id (network-runtime mode, where the bridge owns the thread). A
   * non-empty id is resumed; if the resume fails because the thread has no
   * persisted rollout yet, bootstrap falls back to creating a new one.
   */
  threadId?: string;
  /** Optional label for logs. */
  bridgeLabel?: string;
  /** Slow full-history fallback; production default avoids 5s multi-MB reads. */
  fullHistoryReconciliationIntervalMs?: number;
}

export type BridgeStatus =
  | "connecting"
  | "idle"
  | "working"
  | "waiting_human"
  | "offline";

export interface PendingTurn {
  taskId: string;
  clientUserMessageId: string;
  submittedAt: number;
  turnId?: string;
  /** Raw id returned by turn/start; may be superseded by a goal successor race. */
  responseTurnId?: string;
  /** True once an app-server userMessage echoes our exact client id. */
  identityConfirmed?: boolean;
  /** task_started must precede any terminal task event. */
  startAnnounced?: boolean;
  /** An unknown successor appeared before our client-id echo. */
  competingTurnObserved?: boolean;
  /** Terminal raced turn/start response or identity confirmation. */
  deferredTerminal?: { turnId: string; status?: string; error?: string };
  agentTextChunks: string[];
  finalText?: string;
}

interface SteeredTurn {
  tasks: Map<string, PendingTurn>;
  acceptedTaskIds: Set<string>;
  agentTextChunks: string[];
  finalText?: string;
  terminal?: { status?: string; error?: string };
}

export interface WaitingApproval {
  reverseRequestId: number;
  method: string;
  params: unknown;
  observedAt: number;
}

export interface ActiveTurnReconciliation {
  /** Whether a missed terminal notification was recovered from thread/read. */
  recovered: boolean;
  /** Active turn id observed when reconciliation started, if any. */
  turnId: string | null;
  /** Authoritative status returned by thread/read when a turn was found. */
  status?: string;
}

// ────────────────────────────────────────────────────────────────────────────
// Bridge
// ────────────────────────────────────────────────────────────────────────────

/**
 * Events:
 *   - "status_changed"    → { previous, current }
 *   - "waiting_human"     → WaitingApproval (bridge did not respond)
 *   - "approval_resolved" → { reverseRequestId } — from serverRequest/resolved
 *   - "task_started"      → { taskId, turnId, steered } — model budget starts
 *   - "task_turn_rebound" → { taskId, fromTurnId, toTurnId } — client-id repair
 *   - "task_reply"        → { taskId, text }  — final agent message
 *   - "task_error"        → { taskId, error }
 *   - "cross_thread_drop" → { event } — event for a thread we don't own
 *   - "unowned_turn_drop" → { turnId, event } — turn started outside this bridge
 */
export class CodexAppServerBridge extends EventEmitter {
  private client: CodexAppServerClient;
  private threadId: string;
  private readonly label: string;
  private status: BridgeStatus = "connecting";
  private activeTurnId: string | null = null;
  /**
   * A turn started by the human TUI. Dashboard chat is allowed to steer this
   * turn so a message sent while the human is actively using the TUI does not
   * sit behind a ten-minute FIFO timeout. Agent-to-agent tasks never steer a
   * human turn; callers must opt in explicitly.
   */
  private externalActiveTurnId: string | null = null;
  /** False when reconnect history proves the active turn came from anet. */
  private externalActiveTurnSteerable = false;
  /** Dashboard tasks accepted through turn/steer, grouped by human turn. */
  private steeredTurns = new Map<string, SteeredTurn>();
  /**
   * Synchronous claim taken BEFORE the `turn/start` RPC round-trip.
   * `activeTurnId` alone is insufficient as a mutual-exclusion guard: it is
   * only assigned after the server responds, so two concurrent
   * `startTaskTurn` calls could both pass the `activeTurnId === null` check
   * and double-start (通信龙 review finding #1).
   */
  private turnClaimed = false;
  private pendingTurns = new Map<string, PendingTurn>();
  /** Stable task identity across turn-id replacement races. */
  private pendingByClientUserMessageId = new Map<string, PendingTurn>();
  /** Reverse-request ids we've observed but not resolved (bridge policy). */
  private waitingApprovals = new Map<number, WaitingApproval>();
  /**
   * FIFO of tasks waiting for the thread to go idle (通信龙 review finding #2;
   * RFC §6.3 — Phase 0 gate requires the second concurrent task to queue
   * stably rather than error). Drained on turn completion/error.
   */
  private taskQueue: Array<{ taskId: string; text: string; from?: string }> = [];
  private draining = false;
  /** Coalesce watchdog reads so concurrent callers never stampede app-server. */
  private reconciliationInFlight: Promise<ActiveTurnReconciliation> | null = null;
  private readonly fullHistoryReconciliationIntervalMs: number;
  private lastFullHistoryReconciliationAt = Date.now();

  constructor(opts: CodexAppServerBridgeOptions) {
    super();
    this.client = opts.client;
    this.threadId = opts.threadId ?? "";
    this.label = opts.bridgeLabel ?? `bridge:${(this.threadId || "new").slice(0, 8)}`;
    this.fullHistoryReconciliationIntervalMs =
      opts.fullHistoryReconciliationIntervalMs ?? 60_000;
    this.attachClientListeners();
  }

  /** The thread this bridge is bound to (final id after bootstrap adoption). */
  getThreadId(): string {
    return this.threadId;
  }

  /**
   * Perform initialize → initialized → (thread/resume | thread/start).
   *
   * REAL-WIRE CORRECTION (通信龙, verified against codex-cli 0.144.0 live
   * two-client PoC): `initialize` is scoped to the **app-server**, NOT to
   * each connection. When the human TUI (`codex --remote`) has already
   * initialized the shared server, a second client's `initialize` returns
   * `-32600: Already initialized`. RFC §7.2's "每个连接必须单独初始化" is
   * wrong for the shared-server topology — the bridge tolerates an
   * already-initialized server and proceeds straight to thread binding.
   *
   * Thread binding is create-or-resume (mirrors the opencode runtime's
   * session/load-else-new):
   *   - empty threadId → thread/start, adopt the returned id.
   *   - non-empty threadId → thread/resume; if that fails because the
   *     thread has no persisted rollout yet ("no rollout found"), fall
   *     back to thread/start so a stale/never-persisted id can't wedge
   *     the node at boot.
   * Emits "thread_ready" { threadId, created } so the runtime can persist
   * a freshly-created id back to node config.
   */
  async bootstrap(): Promise<void> {
    try {
      await this.client.request("initialize", {
        clientInfo: {
          name: "anet_codex_bridge",
          title: "Agent Network Codex Bridge",
          version: "0.1.0",
        },
      });
      this.client.notify("initialized", {});
    } catch (e) {
      if (!isAlreadyInitialized(e)) throw e;
      // Shared server already initialized by the human TUI — expected in the
      // second-client bridge role. Do NOT re-send `initialized`.
    }

    let created = false;
    if (this.threadId) {
      try {
        await this.client.request("thread/resume", { threadId: this.threadId });
      } catch (e) {
        if (!isNoRollout(e)) throw e;
        // Persisted id never got a rollout (e.g. node created but never ran
        // a turn) — create a fresh thread rather than wedging at boot.
        this.threadId = await this.startNewThread();
        created = true;
      }
    } else {
      this.threadId = await this.startNewThread();
      created = true;
    }

    this.emit("thread_ready", { threadId: this.threadId, created });
    this.setStatus("idle");
  }

  private async startNewThread(): Promise<string> {
    const started = await this.client.request<{
      threadId?: string;
      thread?: { id?: string };
    }>("thread/start", {});
    const id = started?.threadId ?? started?.thread?.id;
    if (!id) {
      throw new Error(
        "thread/start returned no threadId: " + JSON.stringify(started).slice(0, 200),
      );
    }
    return id;
  }

  /**
   * Start a turn for an Agent Network task. Returns the turnId the server
   * assigned. The final agent reply is delivered later via `task_reply`.
   */
  async startTaskTurn(input: {
    taskId: string;
    text: string;
    from?: string;
  }): Promise<string> {
    // Bridge discipline: one active turn at a time. The claim is taken
    // SYNCHRONOUSLY (before any await) so concurrent callers race cleanly:
    // exactly one proceeds, the rest throw. Queueing callers should use
    // `submitTask()` instead of this low-level primitive.
    if (this.turnClaimed || this.activeTurnId) {
      throw new Error(
        `${this.label}: refusing to start turn for task=${input.taskId} while turn=${this.activeTurnId ?? "(starting)"} active`,
      );
    }
    this.turnClaimed = true;
    const clientUserMessageId = `anet:${input.taskId}`;
    const fromLabel = displaySender(input.from);
    const promptPrefix = fromLabel ? `[Agent Network/from=${fromLabel}/task=${input.taskId}] ` : `[Agent Network/task=${input.taskId}] `;
    const pending: PendingTurn = {
      taskId: input.taskId,
      clientUserMessageId,
      submittedAt: Date.now(),
      agentTextChunks: [],
    };
    // Install the stable identity before the RPC. Notifications can race the
    // turn/start response, and Codex may replace the response turn id when an
    // automatic goal successor starts at the same idle boundary.
    this.pendingByClientUserMessageId.set(clientUserMessageId, pending);
    // Optimistically claim active so a concurrent call races cleanly.
    this.setStatus("working");

    let resp: unknown;
    try {
      resp = await this.client.request("turn/start", {
        threadId: this.threadId,
        clientUserMessageId,
        input: [{ type: "text", text: promptPrefix + input.text }],
      });
    } catch (e) {
      // If the server accepted the turn but the RPC response was lost, the
      // echoed user-message client id is stronger evidence than the failed
      // response transport. Preserve the accepted task instead of duplicating
      // it on retry.
      if (pending.identityConfirmed && pending.turnId) {
        this.announcePendingStart(pending);
        return pending.turnId;
      }
      this.pendingByClientUserMessageId.delete(clientUserMessageId);
      this.turnClaimed = false;
      this.setStatus("idle");
      throw e;
    }
    const responseTurnId = extractTurnId(resp);
    pending.responseTurnId = responseTurnId ?? undefined;
    if (!responseTurnId && !pending.turnId) {
      this.pendingByClientUserMessageId.delete(clientUserMessageId);
      this.turnClaimed = false;
      this.setStatus("idle");
      throw new Error(
        `${this.label}: turn/start response did not include a turnId (task=${input.taskId})`,
      );
    }
    if (!pending.turnId && responseTurnId) {
      this.bindPendingToTurn(pending, responseTurnId, false);
    } else if (responseTurnId && pending.turnId !== responseTurnId) {
      this.emit("task_turn_rebound", {
        taskId: pending.taskId,
        fromTurnId: responseTurnId,
        toTurnId: pending.turnId,
        clientUserMessageId,
      });
    }
    this.announcePendingStart(pending);
    return pending.turnId!;
  }

  private bindPendingToTurn(
    pending: PendingTurn,
    turnId: string,
    identityConfirmed: boolean,
  ): void {
    const previousTurnId = pending.turnId;
    if (previousTurnId && previousTurnId !== turnId && this.pendingTurns.get(previousTurnId) === pending) {
      this.pendingTurns.delete(previousTurnId);
    }
    pending.turnId = turnId;
    if (identityConfirmed) pending.identityConfirmed = true;
    this.pendingTurns.set(turnId, pending);
    this.activeTurnId = turnId;
    if (this.externalActiveTurnId === turnId) {
      this.externalActiveTurnId = null;
      this.externalActiveTurnSteerable = false;
    }
    if (previousTurnId && previousTurnId !== turnId) {
      this.emit("task_turn_rebound", {
        taskId: pending.taskId,
        fromTurnId: previousTurnId,
        toTurnId: turnId,
        clientUserMessageId: pending.clientUserMessageId,
      });
    }
  }

  private announcePendingStart(pending: PendingTurn): void {
    if (pending.startAnnounced || !pending.turnId) return;
    pending.startAnnounced = true;
    this.emit("task_started", {
      taskId: pending.taskId,
      turnId: pending.turnId,
      steered: false,
    });
    const deferred = pending.deferredTerminal;
    if (deferred && deferred.turnId === pending.turnId) {
      pending.deferredTerminal = undefined;
      this.finishOwnedTurn(pending.turnId, deferred);
    }
  }

  /**
   * Queueing entry point (Phase 0 gate: "两个 Agent 同时派 task，只创建一个
   * active turn，另一个稳定排队"). If the thread is free the task starts
   * immediately; otherwise it joins a FIFO drained on turn completion/error.
   *
   * Human-priority note (§6.1): the bridge only ever starts a turn when the
   * shared thread is idle from ITS point of view; if the human TUI wins the
   * idle race, `turn/start` fails with an active-turn error and the task is
   * requeued at the FRONT (original order preserved) — the app-server is the
   * authority (§6.3).
   */
  async submitTask(input: {
    taskId: string;
    text: string;
    from?: string;
    steerIfExternalTurn?: boolean;
  }): Promise<
    { started: true; turnId: string; steered?: boolean } | { started: false; queuedAt: number }
  > {
    if (this.turnClaimed || this.activeTurnId) {
      this.taskQueue.push(input);
      this.emit("task_queued", { taskId: input.taskId, depth: this.taskQueue.length });
      return { started: false, queuedAt: this.taskQueue.length };
    }
    if (this.externalActiveTurnId) {
      if (input.steerIfExternalTurn && this.externalActiveTurnSteerable) {
        return this.steerExternalTurn(input, this.externalActiveTurnId);
      }
      this.taskQueue.push(input);
      this.emit("task_queued", { taskId: input.taskId, depth: this.taskQueue.length });
      return { started: false, queuedAt: this.taskQueue.length };
    }
    if (this.taskQueue.length > 0) {
      this.taskQueue.push(input);
      this.emit("task_queued", { taskId: input.taskId, depth: this.taskQueue.length });
      return { started: false, queuedAt: this.taskQueue.length };
    }
    const turnId = await this.startTaskTurn(input);
    return { started: true, turnId };
  }

  private async steerExternalTurn(
    input: { taskId: string; text: string; from?: string; steerIfExternalTurn?: boolean },
    expectedTurnId: string,
  ): Promise<{ started: true; turnId: string; steered: true } | { started: false; queuedAt: number }> {
    const state = this.steeredTurns.get(expectedTurnId) ?? {
      tasks: new Map<string, PendingTurn>(),
      acceptedTaskIds: new Set<string>(),
      agentTextChunks: [],
    };
    state.tasks.set(input.taskId, {
      taskId: input.taskId,
      clientUserMessageId: `anet:${input.taskId}`,
      submittedAt: Date.now(),
      turnId: expectedTurnId,
      agentTextChunks: [],
    });
    this.steeredTurns.set(expectedTurnId, state);

    const fromLabel = displaySender(input.from);
    const promptPrefix = fromLabel
      ? `[Agent Network/from=${fromLabel}/task=${input.taskId}] `
      : `[Agent Network/task=${input.taskId}] `;
    try {
      const response = await this.client.request("turn/steer", {
        threadId: this.threadId,
        input: [{ type: "text", text: promptPrefix + input.text }],
        expectedTurnId,
      });
      const acceptedTurnId = extractTurnId(response);
      if (acceptedTurnId !== expectedTurnId) {
        throw new Error(
          `${this.label}: turn/steer returned mismatched turnId=${acceptedTurnId ?? "(missing)"}; expected=${expectedTurnId}`,
        );
      }
      state.acceptedTaskIds.add(input.taskId);
      this.emit("task_started", {
        taskId: input.taskId,
        turnId: expectedTurnId,
        steered: true,
      });
      // If turn/completed raced the RPC response, attribution waits for this
      // acceptance proof and is emitted now rather than fail-open earlier.
      if (state.terminal) {
        const task = state.tasks.get(input.taskId);
        if (task) this.emitSteeredTask(state, task, state.terminal);
        state.tasks.delete(input.taskId);
        state.acceptedTaskIds.delete(input.taskId);
        if (state.tasks.size === 0) this.steeredTurns.delete(expectedTurnId);
      }
      this.emit("task_steered", { taskId: input.taskId, turnId: expectedTurnId });
      return { started: true, turnId: expectedTurnId, steered: true };
    } catch (error) {
      // The human turn may finish between turn/started and turn/steer. Remove
      // the optimistic mapping, preserve the task in FIFO, and immediately
      // try to drain: if the completion notification already raced past us,
      // no later event would otherwise wake the queue.
      state.tasks.delete(input.taskId);
      state.acceptedTaskIds.delete(input.taskId);
      if (state.tasks.size === 0) this.steeredTurns.delete(expectedTurnId);
      if (this.externalActiveTurnId === expectedTurnId) this.externalActiveTurnId = null;
      this.taskQueue.push(input);
      const queuedAt = this.taskQueue.length;
      this.emit("steer_deferred", {
        taskId: input.taskId,
        turnId: expectedTurnId,
        error: error instanceof Error ? error.message : String(error),
      });
      void this.drainQueue();
      return { started: false, queuedAt };
    }
  }

  /** Drain the FIFO after a turn finishes. One task per idle transition. */
  private async drainQueue(): Promise<void> {
    if (this.draining) return;
    if (this.turnClaimed || this.activeTurnId || this.externalActiveTurnId) return;
    const next = this.taskQueue.shift();
    if (!next) return;
    this.draining = true;
    try {
      await this.startTaskTurn(next);
    } catch (e) {
      // Lost the idle race to the human TUI (or transport error). Requeue at
      // the FRONT to preserve order; retry on the next idle transition.
      this.taskQueue.unshift(next);
      this.emit("drain_deferred", {
        taskId: next.taskId,
        error: e instanceof Error ? e.message : String(e),
      });
    } finally {
      this.draining = false;
    }
  }

  /** Read-only accessors (for testing / observability). */
  currentStatus(): BridgeStatus {
    return this.status;
  }
  activeTurn(): string | null {
    return this.activeTurnId;
  }
  externalActiveTurn(): string | null {
    return this.externalActiveTurnId;
  }

  /**
   * Recover an already-running shared-server turn after bridge reconnect.
   * A live turn/started notification is sufficient in the normal topology,
   * but a restarted bridge missed that frame. Persisted first-user input is
   * the provenance gate: `[Agent Network/…]` means an orphaned network turn
   * and is never steerable; missing/unknown input also fails closed.
   */
  async recoverSharedActiveTurn(): Promise<{ turnId: string | null; steerable: boolean }> {
    const result = await this.client.request<{
      thread?: {
        status?: string | { type?: string };
        turns?: Array<{
          id?: string;
          status?: string;
          items?: Array<{
            type?: string;
            content?: Array<{ type?: string; text?: string }>;
          }>;
        }>;
      };
    }>("thread/read", { threadId: this.threadId, includeTurns: true });
    this.lastFullHistoryReconciliationAt = Date.now();
    if (extractThreadStatus(result?.thread?.status) !== "active") {
      return { turnId: null, steerable: false };
    }
    const active = [...(result?.thread?.turns ?? [])]
      .reverse()
      .find((turn) => turn.status === "inProgress" && typeof turn.id === "string");
    if (!active?.id) return { turnId: null, steerable: false };
    const firstUserText = active.items
      ?.find((item) => item.type === "userMessage")
      ?.content
      ?.find((content) => content.type === "text" && typeof content.text === "string")
      ?.text;
    const steerable = typeof firstUserText === "string" && !/^\s*\[Agent Network(?:\/|\])/u.test(firstUserText);
    this.externalActiveTurnId = active.id;
    this.externalActiveTurnSteerable = steerable;
    if (this.waitingApprovals.size === 0) this.setStatus("working");
    this.emit("external_turn_recovered", { turnId: active.id, steerable });
    return { turnId: active.id, steerable };
  }
  pendingTurnCount(): number {
    return this.pendingTurns.size;
  }
  isWaitingHuman(): boolean {
    return this.waitingApprovals.size > 0;
  }

  // ──────────────────────────────────────────────────────────────────────
  // Internals
  // ──────────────────────────────────────────────────────────────────────

  private attachClientListeners(): void {
    // Notifications we care about (§7.5). We use `.on(method)` rather than
    // parsing a single "notification" stream because the client dispatches
    // notifications on their method name.
    this.client.on("thread/status/changed", (params) => this.onStatusChanged(params));
    this.client.on("turn/started", (params) => this.onTurnStarted(params));
    this.client.on("item/started", (params) => this.onItemStarted(params));
    this.client.on("item/agentMessage/delta", (params) => this.onAgentDelta(params));
    this.client.on("item/completed", (params) => this.onItemCompleted(params));
    this.client.on("turn/completed", (params) => this.onTurnCompleted(params));
    this.client.on("error", (err) => this.emit("client_error", err));
    this.client.on("serverRequest/resolved", (params) => this.onApprovalResolved(params));

    // Reverse requests: approvals must NOT be answered by this bridge.
    this.client.on("reverse_request", (rr: { id: number; method: string; params: unknown }) =>
      this.onReverseRequest(rr),
    );

    this.client.on("close", () => {
      // Transport gone: release the claim so a future reconnect (Phase 1)
      // can restart cleanly. Queue is intentionally preserved — those tasks
      // were never accepted by the server.
      this.turnClaimed = false;
      this.setStatus("offline");
    });
  }

  private onReverseRequest(rr: { id: number; method: string; params: unknown }): void {
    // The set of methods that indicate a human decision. We deliberately
    // widen this to anything that looks like an approval or user-input
    // request, per §7.6. Non-matching reverse requests are still recorded
    // (Phase 0 policy: don't respond to *anything* until a later phase
    // gates specific tool methods through a proxy).
    const waiting: WaitingApproval = {
      reverseRequestId: rr.id,
      method: rr.method,
      params: rr.params,
      observedAt: Date.now(),
    };
    this.waitingApprovals.set(rr.id, waiting);
    this.setStatus("waiting_human");
    this.emit("waiting_human", waiting);
  }

  private onApprovalResolved(params: unknown): void {
    // Server broadcasts `serverRequest/resolved` once *any* subscriber
    // responded. We just clean up our tracking and (if no more approvals
    // are outstanding, and a turn is active) return to working.
    const resolvedId = extractReverseRequestId(params);
    if (resolvedId === null) return;
    if (!this.waitingApprovals.delete(resolvedId)) return;
    this.emit("approval_resolved", { reverseRequestId: resolvedId });
    if (this.waitingApprovals.size === 0) {
      if (this.activeTurnId) this.setStatus("working");
      else this.setStatus("idle");
    }
  }

  private onStatusChanged(_params: unknown): void {
    // Phase 0 relies on the bridge's own accounting (activeTurnId +
    // waitingApprovals) rather than trusting server-reported status. But
    // we still forward as an event for observability.
    this.emit("status_notification", _params);
  }

  private onTurnStarted(params: unknown): void {
    // REAL-WIRE: turn/started params = { threadId, turn: { id, status, … } }.
    const p = params as { threadId?: string };
    if (!p || p.threadId !== this.threadId) {
      this.emit("cross_thread_drop", { event: "turn/started", params });
      return;
    }
    const turnId = extractTurnId(params);
    if (!turnId) return;
    if (!this.pendingTurns.has(turnId)) {
      const ownedTurnWasActive = this.activeTurnId !== null;
      const activePending = this.activeTurnId
        ? this.pendingTurns.get(this.activeTurnId)
        : undefined;
      if (activePending && !activePending.identityConfirmed) {
        activePending.competingTurnObserved = true;
      }
      // A turn we didn't start — e.g. the human TUI. Track its exact id so
      // authenticated Dashboard chat can use turn/steer. We still never map
      // a reply unless at least one task was explicitly and successfully
      // steered into this turn.
      this.externalActiveTurnId = turnId;
      this.externalActiveTurnSteerable = true;
      if (this.waitingApprovals.size === 0) this.setStatus("working");
      this.emit("unowned_turn_drop", { turnId, event: "turn/started" });
      if (ownedTurnWasActive) {
        // A successor can start immediately after our turn completes while
        // its terminal frame is lost. First join any in-flight cheap read,
        // then force one exact-history read. This avoids both a missed force
        // race and a 5-second loop over multi-megabyte thread history.
        void this.reconcileActiveTurn()
          .catch(() => ({ recovered: false, turnId: this.activeTurnId }))
          .then(() => this.reconcileActiveTurn(true))
          .catch((error) => this.emit("reconciliation_error", error));
      }
      return;
    }
  }

  private onItemStarted(_params: unknown): void {
    // No-op in Phase 0. Real bridge will use item ids to gate approval
    // acceptance in later phases.
  }

  private onAgentDelta(params: unknown): void {
    // REAL-WIRE: item/agentMessage/delta = { threadId, turnId, itemId, delta }
    // where `delta` is a plain string (NOT { text }).
    const p = params as { threadId?: string; turnId?: string; delta?: unknown };
    if (!p || p.threadId !== this.threadId) return;
    if (typeof p.turnId !== "string") return;
    const pending = this.pendingTurns.get(p.turnId);
    if (!pending) {
      const steered = this.steeredTurns.get(p.turnId);
      if (steered && typeof p.delta === "string") steered.agentTextChunks.push(p.delta);
      return; // Human TUI is receiving deltas too.
    }
    if (typeof p.delta === "string") pending.agentTextChunks.push(p.delta);
  }

  private onItemCompleted(params: unknown): void {
    // REAL-WIRE: item/completed = { item: { type, text, phase, … }, threadId,
    // turnId }. The final agent answer is the agentMessage item with
    // phase === "final_answer". Capture it as the authoritative reply text.
    const p = params as {
      threadId?: string;
      turnId?: string;
      item?: { type?: string; text?: string; phase?: string; clientId?: string };
    };
    if (!p || p.threadId !== this.threadId) return;
    if (typeof p.turnId !== "string") return;

    // Codex can accept turn/start with response id A, then let an automatic
    // goal successor win the same idle boundary and persist our user message
    // under turn id B. The echoed clientId is generated from the immutable Hub
    // task id and is therefore the only deterministic cross-surface join key.
    // Rebind before consuming answer/terminal events for B.
    if (p.item?.type === "userMessage" && typeof p.item.clientId === "string") {
      const byClientId = this.pendingByClientUserMessageId.get(p.item.clientId);
      if (byClientId) {
        this.bindPendingToTurn(byClientId, p.turnId, true);
        const deferred = byClientId.deferredTerminal;
        if (deferred) {
          byClientId.deferredTerminal = undefined;
          if (deferred.turnId === p.turnId && byClientId.startAnnounced) {
            this.finishOwnedTurn(p.turnId, deferred);
          }
        }
      }
    }
    const pending = this.pendingTurns.get(p.turnId);
    if (!pending) {
      const steered = this.steeredTurns.get(p.turnId);
      if (
        steered &&
        p.item?.type === "agentMessage" &&
        p.item.phase === "final_answer" &&
        typeof p.item.text === "string"
      ) {
        steered.finalText = p.item.text;
      }
      return;
    }
    if (
      p.item?.type === "agentMessage" &&
      p.item.phase === "final_answer" &&
      typeof p.item.text === "string"
    ) {
      pending.finalText = p.item.text;
    }
  }

  private onTurnCompleted(params: unknown): void {
    // REAL-WIRE: turn/completed = { threadId, turn: { id, status, error, … } }.
    const p = params as {
      threadId?: string;
      turn?: { id?: string; status?: string; error?: { message?: string } | null };
    };
    if (!p || p.threadId !== this.threadId) {
      this.emit("cross_thread_drop", { event: "turn/completed", params });
      return;
    }
    const turnId = extractTurnId(params);
    if (!turnId) return;
    const pending = this.pendingTurns.get(turnId);
    if (!pending) {
      // Human-TUI-initiated turn completed. Only tasks that were accepted by
      // turn/steer are mapped; an ordinary human-only turn remains invisible
      // to CommHub. All steered Dashboard tasks receive the same final answer
      // because app-server emits one final answer for the shared active turn.
      const finished = this.finishExternalTurn(turnId, {
        status: p.turn?.status,
        error: p.turn?.error?.message,
      });
      this.emit("unowned_turn_drop", { turnId, event: "turn/completed" });
      // Older app-server builds can omit turn/started for a competing human
      // turn. Preserve the historical idle wake-up for that notification.
      if (!finished) {
        if (this.waitingApprovals.size === 0) this.setStatus("idle");
        void this.drainQueue();
      }
      return;
    }
    // Preserve ordering (task_started before task_reply/error) when
    // notifications beat the turn/start response. Also distrust a terminal
    // for the response id after a distinct successor appeared but before the
    // exact client-id echo confirms which turn actually owns the task.
    const terminalNeedsIdentity =
      Boolean(p.turn?.error?.message) ||
      p.turn?.status === "failed" ||
      p.turn?.status === "interrupted";
    if (
      !pending.startAnnounced ||
      (!pending.identityConfirmed && (pending.competingTurnObserved || terminalNeedsIdentity))
    ) {
      // This cached terminal cannot wait forever: startOwnedTaskTurn announces
      // the response-id claim before returning, and codexAppServerThink arms
      // its bounded model-response timer from task_started (with the resolved
      // submit result as a compatibility fallback). The cross-layer
      // identity-never-confirms test protects that dependency.
      pending.deferredTerminal = {
        turnId,
        status: p.turn?.status,
        error: p.turn?.error?.message,
      };
      return;
    }
    this.finishOwnedTurn(turnId, {
      status: p.turn?.status,
      error: p.turn?.error?.message,
    });
  }

  /**
   * Reconcile the bridge's local active-turn claim against app-server's
   * authoritative persisted thread state.
   *
   * WebSocket notifications are best-effort. If one `turn/completed` frame is
   * missed, the old implementation retained `activeTurnId` forever: every
   * later Agent Network task entered the FIFO and timed out even though the
   * exact persisted turn was completed. The aggregate thread can be idle or
   * already active with a successor; only an explicit terminal status for the
   * exact claimed turn releases it.
   */
  async reconcileActiveTurn(forceFullHistory = false): Promise<ActiveTurnReconciliation> {
    if (this.reconciliationInFlight) return this.reconciliationInFlight;
    const ownedAtStart = this.activeTurnId;
    const externalAtStart = this.externalActiveTurnId;
    const activeAtStart = ownedAtStart ?? externalAtStart;
    if (!activeAtStart) return { recovered: false, turnId: null };
    const claimedTurnChanged = () => ownedAtStart !== null
      ? this.activeTurnId !== activeAtStart
      : this.externalActiveTurnId !== activeAtStart;

    this.reconciliationInFlight = (async () => {
      // Most watchdog ticks stay cheap. Full history on a real long-lived TUI
      // can be several megabytes, so hydrate it only when the thread is idle,
      // a successor event explicitly forces an exact check, or the slow
      // fallback interval expires (covering loss of both terminal and
      // successor notifications).
      const statusResult = await this.client.request<{
        thread?: { status?: string | { type?: string } };
      }>("thread/read", { threadId: this.threadId, includeTurns: false });

      if (claimedTurnChanged()) {
        return { recovered: false, turnId: activeAtStart };
      }

      const threadStatus = extractThreadStatus(statusResult?.thread?.status);
      const fullHistoryDue =
        forceFullHistory ||
        !threadStatus ||
        threadStatus === "idle" ||
        Date.now() - this.lastFullHistoryReconciliationAt >=
          this.fullHistoryReconciliationIntervalMs;
      if (!fullHistoryDue) {
        return { recovered: false, turnId: activeAtStart, status: threadStatus };
      }

      const result = await this.client.request<{
        thread?: {
          status?: string | { type?: string };
          turns?: Array<{
            id?: string;
            status?: string;
            error?: { message?: string } | null;
            items?: Array<{ type?: string; text?: string; phase?: string; clientId?: string }>;
          }>;
        };
      }>("thread/read", { threadId: this.threadId, includeTurns: true });
      this.lastFullHistoryReconciliationAt = Date.now();

      // A notification may have completed the turn while thread/read was in
      // flight. Never let an old snapshot release a newer active claim.
      if (claimedTurnChanged()) {
        return { recovered: false, turnId: activeAtStart };
      }

      const turns = result?.thread?.turns ?? [];
      const activePending = ownedAtStart
        ? this.pendingTurns.get(activeAtStart)
        : undefined;
      let turn = turns.find((candidate) => candidate.id === activeAtStart);
      if (activePending) {
        const clientMatchedTurn = turns.find((candidate) =>
          candidate.items?.some((item) =>
            item.type === "userMessage" &&
            item.clientId === activePending.clientUserMessageId
          )
        );
        if (clientMatchedTurn?.id) {
          this.bindPendingToTurn(activePending, clientMatchedTurn.id, true);
          turn = clientMatchedTurn;
        }
      }
      const resolvedTurnId = activePending?.turnId ?? activeAtStart;
      if (!turn || !isTerminalTurnStatus(turn.status)) {
        return {
          recovered: false,
          turnId: resolvedTurnId,
          status: turn?.status ?? extractThreadStatus(result?.thread?.status),
        };
      }

      const finalText = [...(turn.items ?? [])]
        .reverse()
        .find((item) =>
          item.type === "agentMessage" &&
          item.phase === "final_answer" &&
          typeof item.text === "string"
        )?.text;
      const recovered = ownedAtStart
        ? this.finishOwnedTurn(resolvedTurnId, {
          status: turn.status,
          error: turn.error?.message,
          finalText,
        })
        : this.finishExternalTurn(activeAtStart, {
          status: turn.status,
          error: turn.error?.message,
          finalText,
        });
      if (recovered) {
        this.emit("turn_reconciled", { turnId: resolvedTurnId, status: turn.status });
      }
      return { recovered, turnId: resolvedTurnId, status: turn.status };
    })().finally(() => {
      this.reconciliationInFlight = null;
    });
    return this.reconciliationInFlight;
  }

  private finishOwnedTurn(
    turnId: string,
    terminal: { status?: string; error?: string; finalText?: string },
  ): boolean {
    const pending = this.pendingTurns.get(turnId);
    if (!pending) return false;
    if (terminal.finalText) pending.finalText = terminal.finalText;
    this.pendingTurns.delete(turnId);
    this.pendingByClientUserMessageId.delete(pending.clientUserMessageId);
    if (this.activeTurnId === turnId) {
      this.activeTurnId = null;
      this.turnClaimed = false;
      this.setStatus(this.waitingApprovals.size > 0 ? "waiting_human" : "idle");
    }
    const turnErr = terminal.error ??
      (terminal.status === "failed"
        ? "Codex turn failed without an error message"
        : terminal.status === "interrupted"
          ? "Codex turn was interrupted without an error message"
          : undefined);
    if (turnErr) {
      this.emit("task_error", { taskId: pending.taskId, error: turnErr });
    } else {
      // Prefer the captured final_answer item; fall back to accumulated deltas.
      const text = pending.finalText && pending.finalText.length > 0
        ? pending.finalText
        : pending.agentTextChunks.join("");
      this.emit("task_reply", { taskId: pending.taskId, text });
    }
    // Either way the thread just went idle from our perspective → drain FIFO.
    void this.drainQueue();
    return true;
  }

  private finishExternalTurn(
    turnId: string,
    terminal: { status?: string; error?: string; finalText?: string },
  ): boolean {
    if (this.externalActiveTurnId !== turnId && !this.steeredTurns.has(turnId)) return false;
    if (this.externalActiveTurnId === turnId) {
      this.externalActiveTurnId = null;
      this.externalActiveTurnSteerable = false;
    }
    const steered = this.steeredTurns.get(turnId);
    if (steered) {
      if (terminal.finalText) steered.finalText = terminal.finalText;
      steered.terminal = { status: terminal.status, error: terminal.error };
      for (const [taskId, task] of steered.tasks) {
        if (!steered.acceptedTaskIds.has(taskId)) continue;
        this.emitSteeredTask(steered, task, steered.terminal);
        steered.tasks.delete(taskId);
        steered.acceptedTaskIds.delete(taskId);
      }
      if (steered.tasks.size === 0) this.steeredTurns.delete(turnId);
    }
    if (this.waitingApprovals.size === 0) this.setStatus("idle");
    void this.drainQueue();
    return true;
  }

  private emitSteeredTask(
    state: SteeredTurn,
    task: PendingTurn,
    terminal: { status?: string; error?: string },
  ): void {
    const turnErr = terminal.error ??
      (terminal.status === "failed"
        ? "Codex turn failed without an error message"
        : terminal.status === "interrupted"
          ? "Codex turn was interrupted without an error message"
          : undefined);
    const text = state.finalText && state.finalText.length > 0
      ? state.finalText
      : state.agentTextChunks.join("");
    if (turnErr) this.emit("task_error", { taskId: task.taskId, error: turnErr });
    else this.emit("task_reply", { taskId: task.taskId, text });
  }

  /** Read-only queue depth (for tests / observability). */
  queueDepth(): number {
    return this.taskQueue.length;
  }

  /**
   * Remove one task that has not begun `turn/start`/`turn/steer` yet.
   * Used by the runtime's independent queue-wait deadline so a timed-out
   * Dashboard row cannot silently execute later after its Hub lifecycle has
   * already been closed as failed.
   */
  cancelQueuedTask(taskId: string): boolean {
    const index = this.taskQueue.findIndex((task) => task.taskId === taskId);
    if (index < 0) return false;
    this.taskQueue.splice(index, 1);
    this.emit("task_queue_cancelled", { taskId, depth: this.taskQueue.length });
    return true;
  }

  private setStatus(next: BridgeStatus): void {
    if (this.status === next) return;
    const previous = this.status;
    this.status = next;
    this.emit("status_changed", { previous, current: next });
  }
}

// ────────────────────────────────────────────────────────────────────────────
// Field extractors — defensive against schema drift.
// ────────────────────────────────────────────────────────────────────────────

// REAL-WIRE (codex-cli 0.144.0): turn/start response and turn/{started,completed}
// events nest the id under `turn.id`; only item-level events use a flat
// `turnId`. Accept both so this survives either surface.
function extractTurnId(v: unknown): string | null {
  if (!v || typeof v !== "object") return null;
  const o = v as { turn?: { id?: unknown }; turnId?: unknown };
  if (o.turn && typeof o.turn === "object" && typeof o.turn.id === "string") return o.turn.id;
  if (typeof o.turnId === "string") return o.turnId;
  return null;
}

function isTerminalTurnStatus(status: unknown): status is "completed" | "failed" | "interrupted" {
  return status === "completed" || status === "failed" || status === "interrupted";
}

function extractThreadStatus(status: unknown): string | undefined {
  if (typeof status === "string") return status;
  if (status && typeof status === "object") {
    const type = (status as { type?: unknown }).type;
    if (typeof type === "string") return type;
  }
  return undefined;
}

/** Recognise the app-server's "already initialized" rejection (shared server). */
function isAlreadyInitialized(e: unknown): boolean {
  const code = (e as { code?: unknown })?.code;
  const msg = (e as { message?: unknown })?.message;
  if (code === -32600) return true;
  return typeof msg === "string" && /already initialized/i.test(msg);
}

/** thread/resume against an id the app-server has no persisted rollout for. */
function isNoRollout(e: unknown): boolean {
  const msg = (e as { message?: unknown })?.message;
  return typeof msg === "string" && /no rollout found|thread not found|unknown thread/i.test(msg);
}

/**
 * Human-readable sender label for the codex prompt prefix. The hub defaults
 * `from_session` to "api" for dashboard/REST-originated tasks (server/src/
 * index.ts), which reads as opaque "from=api" in the TUI — map it to a clear
 * "控制台(dashboard)" so a human at the TUI knows it came from the panel.
 * Everything else (agent aliases, "hub") passes through unchanged.
 */
function displaySender(from?: string): string | undefined {
  if (!from) return undefined;
  if (from === "api") return "控制台(dashboard)";
  return from;
}

function extractReverseRequestId(params: unknown): number | null {
  if (!params || typeof params !== "object") return null;
  const p = params as { reverseRequestId?: unknown; requestId?: unknown };
  if (typeof p.reverseRequestId === "number") return p.reverseRequestId;
  if (typeof p.requestId === "number") return p.requestId;
  return null;
}
