use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicI64, Ordering};

use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite, AsyncWriteExt};
use tokio::process::ChildStdin;
use tokio::sync::broadcast;
use tokio::sync::mpsc;
use tokio::sync::oneshot;

use super::environment::build_agent_command;
use super::{
    AgentEvent, AssistantMessage, ContentBlock, ControlRequestInner, Delta, FileAttachment,
    InnerStreamEvent, StartContentBlock, StreamEvent, TokenUsage, TurnHandle,
};

type CodexStdin = Arc<tokio::sync::Mutex<ChildStdin>>;
type PendingRequests = Arc<tokio::sync::Mutex<BTreeMap<JsonRpcId, PendingCodexRequest>>>;
type TurnOutputBuffer = Arc<tokio::sync::Mutex<CodexTurnOutput>>;

struct PendingCodexRequest {
    method: String,
    tx: oneshot::Sender<Result<JsonRpcResponse, JsonRpcError>>,
}

#[derive(Debug, Default)]
struct CodexTurnOutput {
    text: String,
    thinking: String,
    /// Item-id of the agentMessage item currently accumulating into
    /// `text`. When a delta with a different item_id arrives (or that
    /// item's `item/completed` fires), we drain `text` as its own
    /// `ContentBlock::Text` so the persisted turn keeps item boundaries
    /// instead of gluing every assistant utterance into one wall.
    text_item_id: Option<String>,
    /// Same idea for reasoning items: each discrete reasoning item is
    /// drained as its own `ContentBlock::Thinking` so the UI's
    /// `Thinking…` collapsible streams progressively rather than
    /// arriving as one mega-block after the turn ends.
    thinking_item_id: Option<String>,
    command_outputs: BTreeMap<String, String>,
}

#[derive(Debug, Clone)]
pub struct CodexAppServerOptions {
    pub model: Option<String>,
    pub permission_level: CodexPermissionLevel,
    pub plan_mode: bool,
    pub fast_mode: bool,
    pub reasoning_effort: Option<String>,
    pub resume_thread_id: Option<String>,
    pub custom_instructions: Option<String>,
    pub mcp_config: Option<String>,
    pub workspace_env: Option<crate::env::WorkspaceEnv>,
    pub resolved_env: Option<crate::env_provider::ResolvedEnv>,
}

impl Default for CodexAppServerOptions {
    fn default() -> Self {
        Self {
            model: None,
            permission_level: CodexPermissionLevel::Readonly,
            plan_mode: false,
            fast_mode: false,
            reasoning_effort: None,
            resume_thread_id: None,
            custom_instructions: None,
            mcp_config: None,
            workspace_env: None,
            resolved_env: None,
        }
    }
}

pub struct CodexAppServerSession {
    pid: u32,
    stdin: Option<CodexStdin>,
    event_tx: broadcast::Sender<AgentEvent>,
    /// Side-channel for `account/rateLimits/updated` notifications and
    /// the initial `account/rateLimits/read` seed. Lives separately
    /// from `event_tx` so the usage cache can subscribe without having
    /// to filter every chat event. The Tauri host wires this to
    /// `AppState.codex_rate_limits` in `chat::send` when it spawns the
    /// session; non-chat callers (discovery probe, model listing) can
    /// ignore it.
    rate_limits_tx: broadcast::Sender<CodexRateLimitSnapshot>,
    pending: PendingRequests,
    /// Set the first time the stderr reader recognises an unrecoverable
    /// Codex auth-expiry message. Once set, the session is considered
    /// terminally broken: in-flight requests are failed with a user-facing
    /// re-auth hint, and subsequent matching stderr lines are silenced so
    /// the log isn't flooded with the same line per second. Persists for
    /// the lifetime of the session; the user must re-spawn (e.g. by
    /// resending after running `codex login`) to clear it.
    auth_expired: Arc<AtomicBool>,
    next_request_id: AtomicI64,
    working_dir: PathBuf,
    model: Option<String>,
    permission_level: CodexPermissionLevel,
    plan_mode: bool,
    fast_mode: bool,
    reasoning_effort: Option<String>,
    resume_thread_id: Option<String>,
    custom_instructions: Option<String>,
    mcp_config: Option<String>,
    thread_id: Arc<tokio::sync::Mutex<Option<String>>>,
    active_turn_id: Arc<tokio::sync::Mutex<Option<String>>>,
    /// True only for an explicit `/compact` turn started through
    /// `thread/compact/start`. Auto-compaction happens inside a normal
    /// Codex turn and must not synthesize a terminal Result.
    manual_compaction_in_flight: Arc<AtomicBool>,
    turn_output: TurnOutputBuffer,
}

impl CodexAppServerSession {
    pub async fn start(working_dir: &Path, client_version: &str) -> Result<Self, String> {
        Self::start_with_options(
            working_dir,
            client_version,
            CodexAppServerOptions::default(),
        )
        .await
    }

    pub async fn start_with_options(
        working_dir: &Path,
        client_version: &str,
        options: CodexAppServerOptions,
    ) -> Result<Self, String> {
        crate::missing_cli::precheck_cwd(working_dir)?;

        let codex_path = super::binary::resolve_codex_path().await;
        let args = codex_app_server_args()
            .into_iter()
            .map(str::to_string)
            .collect::<Vec<_>>();
        let built_command = build_agent_command(
            codex_path.as_os_str(),
            &args,
            working_dir,
            options.resolved_env.as_ref(),
        );
        let mut cmd = built_command.command;
        cmd.stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped());

        if let Some(env) = options.workspace_env.as_ref() {
            env.apply(&mut cmd);
        }

        let mut child = cmd.spawn().map_err(|e| {
            crate::missing_cli::map_spawn_err(&e, "codex", || {
                format!("Failed to spawn Codex app-server: {e}")
            })
        })?;
        let pid = child
            .id()
            .ok_or_else(|| "Codex app-server exited immediately".to_string())?;
        let stdin = child
            .stdin
            .take()
            .ok_or_else(|| "Failed to capture Codex app-server stdin".to_string())?;
        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| "Failed to capture Codex app-server stdout".to_string())?;
        let stderr = child
            .stderr
            .take()
            .ok_or_else(|| "Failed to capture Codex app-server stderr".to_string())?;

        let (event_tx, _) = broadcast::channel(2048);
        // Rate-limit updates are infrequent (a few per session). A
        // small buffer is plenty; late subscribers can miss earlier
        // snapshots without harm — the cache always reflects whichever
        // value arrived last.
        let (rate_limits_tx, _) = broadcast::channel(16);
        let session = Self {
            pid,
            stdin: Some(Arc::new(tokio::sync::Mutex::new(stdin))),
            event_tx,
            rate_limits_tx,
            pending: Arc::new(tokio::sync::Mutex::new(BTreeMap::new())),
            auth_expired: Arc::new(AtomicBool::new(false)),
            next_request_id: AtomicI64::new(1),
            working_dir: working_dir.to_path_buf(),
            model: options.model,
            permission_level: options.permission_level,
            plan_mode: options.plan_mode,
            fast_mode: options.fast_mode,
            reasoning_effort: normalize_codex_reasoning_effort(options.reasoning_effort.as_deref()),
            resume_thread_id: options.resume_thread_id,
            custom_instructions: options.custom_instructions,
            mcp_config: options.mcp_config,
            thread_id: Arc::new(tokio::sync::Mutex::new(None)),
            active_turn_id: Arc::new(tokio::sync::Mutex::new(None)),
            manual_compaction_in_flight: Arc::new(AtomicBool::new(false)),
            turn_output: Arc::new(tokio::sync::Mutex::new(CodexTurnOutput::default())),
        };

        session.spawn_stdout_reader(stdout);
        session.spawn_stderr_reader(stderr);
        session.spawn_exit_watcher(child);
        let _ = session.event_tx.send(codex_command_line_event());
        if let Err(err) = session.initialize(client_version).await {
            fail_pending_requests(&session.pending, "Codex app-server initialize failed").await;
            if let Err(stop_err) = super::process::stop_agent_graceful(pid).await {
                tracing::warn!(
                    target: "claudette::agent",
                    subsystem = "codex-app-server",
                    pid,
                    error = %stop_err,
                    "failed to stop codex app-server after initialize failure"
                );
            }
            return Err(err);
        }
        if session.resume_thread_id.is_some()
            && let Err(err) = session.ensure_thread().await
        {
            fail_pending_requests(&session.pending, "Codex app-server resume failed").await;
            if let Err(stop_err) = super::process::stop_agent_graceful(pid).await {
                tracing::warn!(
                    target: "claudette::agent",
                    subsystem = "codex-app-server",
                    pid,
                    error = %stop_err,
                    "failed to stop codex app-server after resume failure"
                );
            }
            return Err(err);
        }

        // Note: we intentionally do NOT call `seed_rate_limits()`
        // here. The seed would broadcast on `rate_limits_tx` before
        // any caller has had a chance to subscribe (the
        // `subscribe_rate_limits()` receiver is created *after*
        // `start_with_options` returns), so the value would be
        // dropped, paying an extra RPC for nothing. Callers that
        // actually want a seed call `seed_rate_limits()` themselves
        // after subscribing; callers that want a one-shot read
        // (e.g. `prefetch_codex_rate_limits`) call
        // [`read_rate_limits`] directly and use the return value.
        Ok(session)
    }

    #[doc(hidden)]
    pub fn new_for_test(pid: u32) -> Self {
        let (event_tx, _) = broadcast::channel(128);
        let (rate_limits_tx, _) = broadcast::channel(16);
        Self {
            pid,
            stdin: None,
            event_tx,
            rate_limits_tx,
            pending: Arc::new(tokio::sync::Mutex::new(BTreeMap::new())),
            auth_expired: Arc::new(AtomicBool::new(false)),
            next_request_id: AtomicI64::new(1),
            working_dir: PathBuf::from("/tmp"),
            model: None,
            permission_level: CodexPermissionLevel::Readonly,
            plan_mode: false,
            fast_mode: false,
            reasoning_effort: None,
            resume_thread_id: None,
            custom_instructions: None,
            mcp_config: None,
            thread_id: Arc::new(tokio::sync::Mutex::new(None)),
            active_turn_id: Arc::new(tokio::sync::Mutex::new(None)),
            manual_compaction_in_flight: Arc::new(AtomicBool::new(false)),
            turn_output: Arc::new(tokio::sync::Mutex::new(CodexTurnOutput::default())),
        }
    }

    pub fn pid(&self) -> u32 {
        self.pid
    }

    pub fn subscribe(&self) -> broadcast::Receiver<AgentEvent> {
        self.event_tx.subscribe()
    }

    /// Subscribe to live Codex rate-limit snapshots. Receives every
    /// `account/rateLimits/updated` notification the app-server
    /// pushes, plus the one-shot seed value sent by [`Self::seed_rate_limits`]
    /// after a successful `initialize`. The Tauri host drains this
    /// receiver into `AppState.codex_rate_limits` so the composer
    /// usage meter renders live quota data without polling.
    pub fn subscribe_rate_limits(&self) -> broadcast::Receiver<CodexRateLimitSnapshot> {
        self.rate_limits_tx.subscribe()
    }

    /// Fire `account/rateLimits/read` and publish the result onto the
    /// `subscribe_rate_limits` channel so existing subscribers get a
    /// snapshot before the first `account/rateLimits/updated`
    /// notification arrives. Best-effort: a failure here just means
    /// the meter waits for the next push, so we log and move on.
    pub async fn seed_rate_limits(&self) {
        match self.read_rate_limits().await {
            Ok(response) => {
                let _ = self.rate_limits_tx.send(response.rate_limits);
            }
            Err(err) => {
                tracing::debug!(
                    target: "claudette::agent",
                    subsystem = "codex-app-server",
                    pid = self.pid,
                    error = %err,
                    "Codex rateLimits/read seed failed; will rely on push notifications",
                );
            }
        }
    }

    pub fn publish_notification_event(&self, event: CodexNotificationEvent) {
        for event in map_notification_to_agent_events_with_compact_state(
            event,
            Some(&self.manual_compaction_in_flight),
        ) {
            let _ = self.event_tx.send(event);
        }
    }

    pub async fn send_turn(
        &self,
        prompt: &str,
        attachments: &[FileAttachment],
    ) -> Result<TurnHandle, String> {
        validate_codex_attachments(attachments)?;
        // The codex-app-server subprocess is still running, but its
        // refresh token has rotated and every websocket call will 401.
        // Without this guard the user types a message and we sit on
        // `turn/start` forever waiting for a notification that will
        // never arrive (Codex's responses websocket can't open, but
        // the JSON-RPC channel stays nominally healthy).
        if self.is_auth_expired() {
            return Err(CODEX_AUTH_EXPIRED_MESSAGE.to_string());
        }
        let mut broadcast_rx = self.event_tx.subscribe();
        let thread_id = self.ensure_thread().await?;
        let response = self
            .send_request(build_turn_start_request(CodexTurnStartRequest {
                id: self.next_id(),
                thread_id: &thread_id,
                prompt,
                cwd: &self.working_dir,
                model: self.model.as_deref(),
                permission_level: self.permission_level,
                plan_mode: self.plan_mode,
                fast_mode: self.fast_mode,
                reasoning_effort: self.reasoning_effort.as_deref(),
                attachments,
            }))
            .await?;
        let turn_id = turn_id_from_response(&response)?;
        *self.active_turn_id.lock().await = Some(turn_id);
        for event in codex_turn_start_events(&thread_id) {
            let _ = self.event_tx.send(event);
        }

        let (mpsc_tx, mpsc_rx) = mpsc::channel::<AgentEvent>(128);
        tokio::spawn(async move {
            loop {
                match broadcast_rx.recv().await {
                    Ok(event) => {
                        let is_turn_end =
                            matches!(&event, AgentEvent::Stream(StreamEvent::Result { .. }));
                        let is_process_exit = matches!(&event, AgentEvent::ProcessExited(_));
                        if mpsc_tx.send(event).await.is_err() {
                            break;
                        }
                        if is_turn_end || is_process_exit {
                            break;
                        }
                    }
                    Err(broadcast::error::RecvError::Closed) => break,
                    Err(broadcast::error::RecvError::Lagged(n)) => {
                        tracing::warn!(
                            target: "claudette::agent",
                            subsystem = "codex-app-server",
                            dropped_events = n,
                            "broadcast lag — codex per-turn receiver missed events"
                        );
                    }
                }
            }
        });

        Ok(TurnHandle {
            event_rx: mpsc_rx,
            pid: self.pid,
        })
    }

    pub async fn steer_turn(
        &self,
        prompt: &str,
        attachments: &[FileAttachment],
    ) -> Result<(), String> {
        validate_codex_attachments(attachments)?;
        let thread_id = self.ensure_thread().await?;
        let turn_id = self
            .active_turn_id
            .lock()
            .await
            .clone()
            .ok_or_else(|| "Codex app-server has no active turn to steer".to_string())?;
        self.send_request(build_turn_steer_request(
            self.next_id(),
            &thread_id,
            &turn_id,
            prompt,
            attachments,
        ))
        .await?;
        Ok(())
    }

    pub async fn interrupt_turn(&self) -> Result<(), String> {
        let thread_id = self
            .thread_id
            .lock()
            .await
            .clone()
            .ok_or_else(|| "Codex app-server has no thread to interrupt".to_string())?;
        let turn_id = self
            .active_turn_id
            .lock()
            .await
            .clone()
            .ok_or_else(|| "Codex app-server has no active turn to interrupt".to_string())?;
        self.send_request(build_turn_interrupt_request(
            self.next_id(),
            &thread_id,
            &turn_id,
        ))
        .await?;
        *self.active_turn_id.lock().await = None;
        Ok(())
    }

    /// Trigger Codex's native context compaction via the `thread/compact/start`
    /// JSON-RPC method. Returns a [`TurnHandle`] shaped exactly like
    /// [`Self::send_turn`]'s so the caller (`send_chat_message`) can plug it
    /// into the same per-turn event pump without branching.
    ///
    /// The compaction itself is asynchronous: Codex enqueues a non-steerable
    /// turn and, on completion, emits an `item/completed` notification
    /// carrying a `contextCompaction` thread item. That item is translated
    /// into a `compact_boundary` + synthetic `Result` event pair by
    /// [`map_codex_item_completed_to_agent_events`]. The boundary persists
    /// the existing `COMPACTION:...` sentinel; the Result terminates the
    /// per-turn pump so `session.agent_status` flips back to `"Idle"`.
    ///
    /// Codex's app-server protocol does not surface the compaction summary
    /// text — `CompactedItem.message` is empty by design upstream, and the
    /// `ContextCompaction` item carries only `{id}`. There is no equivalent
    /// of Claude CLI's `SYNTHETIC_SUMMARY` here; the boundary divider is the
    /// only user-visible artifact.
    ///
    /// Immediately after the RPC accepts, broadcasts a synthetic
    /// `subtype: "status", status: "compacting"` event so the frontend
    /// `useAgentStream` listener flips `workspace.agent_status` to
    /// `"Compacting"` and shows the "Compacting context…" affordance the
    /// Claude CLI emits natively. Without this, the UI sees nothing happen
    /// for the entire wait between the RPC and the eventual
    /// `ContextCompaction` item.
    pub async fn start_compact(&self) -> Result<TurnHandle, String> {
        if self.is_auth_expired() {
            return Err(CODEX_AUTH_EXPIRED_MESSAGE.to_string());
        }
        // Subscribe BEFORE the RPC so neither the synthetic "compacting"
        // status nor the eventual ContextCompaction item slip through
        // before the per-turn pump exists.
        let mut broadcast_rx = self.event_tx.subscribe();
        let thread_id = self.ensure_thread().await?;
        self.manual_compaction_in_flight
            .store(true, Ordering::SeqCst);
        if let Err(err) = self
            .send_request(build_thread_compact_start_request(
                self.next_id(),
                &thread_id,
            ))
            .await
        {
            self.manual_compaction_in_flight
                .store(false, Ordering::SeqCst);
            return Err(err);
        }
        if self.manual_compaction_in_flight.load(Ordering::SeqCst) {
            let _ = self.event_tx.send(codex_compacting_status_event());
        }

        let (mpsc_tx, mpsc_rx) = mpsc::channel::<AgentEvent>(128);
        tokio::spawn(async move {
            loop {
                match broadcast_rx.recv().await {
                    Ok(event) => {
                        let is_turn_end =
                            matches!(&event, AgentEvent::Stream(StreamEvent::Result { .. }));
                        let is_process_exit = matches!(&event, AgentEvent::ProcessExited(_));
                        if mpsc_tx.send(event).await.is_err() {
                            break;
                        }
                        if is_turn_end || is_process_exit {
                            break;
                        }
                    }
                    Err(broadcast::error::RecvError::Closed) => break,
                    Err(broadcast::error::RecvError::Lagged(n)) => {
                        tracing::warn!(
                            target: "claudette::agent",
                            subsystem = "codex-app-server",
                            dropped_events = n,
                            "broadcast lag — codex compact pump missed events"
                        );
                    }
                }
            }
        });

        Ok(TurnHandle {
            event_rx: mpsc_rx,
            pid: self.pid,
        })
    }

    pub async fn read_account(
        &self,
        refresh_token: bool,
    ) -> Result<CodexAppServerAccountStatus, String> {
        let response = self
            .send_request(build_account_read_request(self.next_id(), refresh_token))
            .await?;
        account_status_from_response(&response)
    }

    /// Issue `account/rateLimits/read` against the live Codex app-server.
    /// Returns the current quota snapshot keyed by limit id (typically
    /// just `"codex"`) plus the backward-compatible single-bucket view.
    /// Notifications (`account/rateLimits/updated`) keep the snapshot
    /// fresh after this initial seed; see [`CodexNotificationEvent::RateLimitsUpdated`].
    pub async fn read_rate_limits(&self) -> Result<GetAccountRateLimitsResponse, String> {
        let response = self
            .send_request(build_account_rate_limits_read_request(self.next_id()))
            .await?;
        rate_limits_from_response(&response)
    }

    pub async fn list_models(&self) -> Result<Vec<CodexAppServerModel>, String> {
        let mut cursor: Option<String> = None;
        let mut models = Vec::new();
        loop {
            let response = self
                .send_request(build_model_list_request(self.next_id(), cursor.as_deref()))
                .await?;
            let page = model_list_from_response(&response)?;
            models.extend(page.models);
            cursor = page.next_cursor;
            if cursor.is_none() {
                break;
            }
        }
        Ok(models)
    }

    pub async fn send_control_response(
        &self,
        request_id: &str,
        response: Value,
    ) -> Result<(), String> {
        let server_response =
            build_codex_server_response_from_control_response(request_id, response)?;
        let stdin = self
            .stdin
            .as_ref()
            .ok_or_else(|| "Codex app-server stdin is not available".to_string())?;
        let mut stdin = stdin.lock().await;
        write_jsonrpc_message(&mut *stdin, &server_response).await
    }

    async fn initialize(&self, client_version: &str) -> Result<(), String> {
        let request = build_initialize_request(self.next_id(), client_version);
        self.send_request(request).await?;
        self.send_notification(build_initialized_notification())
            .await
    }

    async fn ensure_thread(&self) -> Result<String, String> {
        if let Some(thread_id) = self.thread_id.lock().await.clone() {
            return Ok(thread_id);
        }
        let params = self.thread_request_params();
        let response = if let Some(resume_thread_id) = self.resume_thread_id.as_deref() {
            self.send_request(build_thread_resume_request(
                self.next_id(),
                resume_thread_id,
                params,
            ))
            .await?
        } else {
            self.send_request(build_thread_start_request(self.next_id(), params))
                .await?
        };
        let thread_id = thread_id_from_response(&response)?;
        *self.thread_id.lock().await = Some(thread_id.clone());
        Ok(thread_id)
    }

    fn thread_request_params(&self) -> CodexThreadRequestParams<'_> {
        CodexThreadRequestParams {
            model: self.model.as_deref(),
            cwd: &self.working_dir,
            permission_level: self.permission_level,
            fast_mode: self.fast_mode,
            reasoning_effort: self.reasoning_effort.as_deref(),
            custom_instructions: self.custom_instructions.as_deref(),
            mcp_config: self.mcp_config.as_deref(),
        }
    }

    async fn send_request(&self, request: JsonRpcRequest) -> Result<JsonRpcResponse, String> {
        let stdin = self
            .stdin
            .as_ref()
            .ok_or_else(|| "Codex app-server stdin is not available".to_string())?;
        let (tx, rx) = oneshot::channel();
        {
            let mut pending = self.pending.lock().await;
            pending.insert(
                request.id.clone(),
                PendingCodexRequest {
                    method: request.method.clone(),
                    tx,
                },
            );
        }
        let write_result = {
            let mut stdin = stdin.lock().await;
            write_jsonrpc_message(&mut *stdin, &JsonRpcMessage::Request(request.clone())).await
        };
        if let Err(err) = write_result {
            let mut pending = self.pending.lock().await;
            pending.remove(&request.id);
            return Err(err);
        }
        match rx.await.map_err(|_| {
            format!(
                "Codex app-server response channel closed for `{}`",
                request.method
            )
        })? {
            Ok(response) => Ok(response),
            Err(error) => Err(format!(
                "Codex app-server `{}` failed: {}",
                request.method, error.error.message
            )),
        }
    }

    async fn send_notification(&self, notification: JsonRpcNotification) -> Result<(), String> {
        let stdin = self
            .stdin
            .as_ref()
            .ok_or_else(|| "Codex app-server stdin is not available".to_string())?;
        let mut stdin = stdin.lock().await;
        write_jsonrpc_message(&mut *stdin, &JsonRpcMessage::Notification(notification)).await
    }

    fn next_id(&self) -> i64 {
        self.next_request_id.fetch_add(1, Ordering::Relaxed)
    }

    fn spawn_stdout_reader(&self, stdout: tokio::process::ChildStdout) {
        let event_tx = self.event_tx.clone();
        let rate_limits_tx = self.rate_limits_tx.clone();
        let pending = self.pending.clone();
        let stdin = self.stdin.clone();
        let active_turn_id = self.active_turn_id.clone();
        let manual_compaction_in_flight = self.manual_compaction_in_flight.clone();
        let turn_output = self.turn_output.clone();
        let pid = self.pid;
        tokio::spawn(async move {
            let mut reader = tokio::io::BufReader::new(stdout);
            loop {
                match read_jsonrpc_message(&mut reader).await {
                    Ok(Some(message)) => {
                        route_app_server_message(
                            pid,
                            &event_tx,
                            &rate_limits_tx,
                            &pending,
                            stdin.as_ref(),
                            Some(&active_turn_id),
                            Some(&manual_compaction_in_flight),
                            Some(&turn_output),
                            message,
                        )
                        .await;
                    }
                    Ok(None) => {
                        fail_pending_requests(
                            &pending,
                            "Codex app-server stdout closed before responding",
                        )
                        .await;
                        break;
                    }
                    Err(err) => {
                        tracing::warn!(
                            target: "claudette::agent",
                            subsystem = "codex-app-server",
                            pid,
                            error = %err,
                            "failed to read codex stdout"
                        );
                        let _ = event_tx.send(AgentEvent::Stderr(err.clone()));
                        fail_pending_requests(&pending, &err).await;
                        break;
                    }
                }
            }
        });
    }

    fn spawn_stderr_reader(&self, stderr: tokio::process::ChildStderr) {
        let event_tx = self.event_tx.clone();
        let pending = self.pending.clone();
        let auth_expired = self.auth_expired.clone();
        let pid = self.pid;
        tokio::spawn(async move {
            let mut reader = tokio::io::BufReader::new(stderr);
            let mut line = String::new();
            loop {
                line.clear();
                match reader.read_line(&mut line).await {
                    Ok(0) => break,
                    Ok(_) => {
                        let line = line.trim();
                        if line.is_empty() {
                            continue;
                        }
                        // Codex's own login subsystem emits these phrases on
                        // stderr when the cached refresh token has been
                        // consumed (e.g. another `codex login` session
                        // rotated it) and the websocket to ChatGPT 401s
                        // before any notification flows back through stdout.
                        // From Claudette's side this looks like a silent
                        // hang on `turn/start`. Detect once, fail in-flight
                        // RPCs with a user-readable hint, then mute further
                        // matching lines so we don't WARN-flood the log at
                        // ~5 Hz while Codex keeps retrying.
                        let already_signalled = auth_expired.load(Ordering::SeqCst);
                        if codex_stderr_indicates_auth_expiry(line) {
                            if !already_signalled
                                && auth_expired
                                    .compare_exchange(
                                        false,
                                        true,
                                        Ordering::SeqCst,
                                        Ordering::SeqCst,
                                    )
                                    .is_ok()
                            {
                                tracing::warn!(
                                    target: "claudette::agent",
                                    subsystem = "codex-app-server",
                                    pid,
                                    line,
                                    "Codex auth expired — failing pending requests"
                                );
                                fail_pending_requests(&pending, CODEX_AUTH_EXPIRED_MESSAGE).await;
                                let _ = event_tx.send(AgentEvent::Stderr(
                                    CODEX_AUTH_EXPIRED_MESSAGE.to_string(),
                                ));
                            }
                            // Either we just signalled or a previous line
                            // already did — silence the duplicate.
                            continue;
                        }
                        tracing::warn!(
                            target: "claudette::agent",
                            subsystem = "codex-app-server",
                            pid,
                            line,
                            "codex stderr"
                        );
                        let _ = event_tx.send(AgentEvent::Stderr(line.to_string()));
                    }
                    Err(err) => {
                        tracing::warn!(
                            target: "claudette::agent",
                            subsystem = "codex-app-server",
                            pid,
                            error = %err,
                            "failed to read codex stderr"
                        );
                        break;
                    }
                }
            }
        });
    }

    /// True once the stderr reader has flagged the session as terminally
    /// auth-broken. Caller can use this to skip a doomed `turn/start` RPC
    /// or to decide whether to tear down the session and respawn after a
    /// successful `codex login`.
    pub fn is_auth_expired(&self) -> bool {
        self.auth_expired.load(Ordering::SeqCst)
    }

    fn spawn_exit_watcher(&self, mut child: tokio::process::Child) {
        let event_tx = self.event_tx.clone();
        let pending = self.pending.clone();
        tokio::spawn(async move {
            let status = child.wait().await.ok().and_then(|status| status.code());
            fail_pending_requests(&pending, "Codex app-server process exited").await;
            let _ = event_tx.send(AgentEvent::ProcessExited(status));
        });
    }
}

#[allow(clippy::too_many_arguments)]
async fn route_app_server_message(
    pid: u32,
    event_tx: &broadcast::Sender<AgentEvent>,
    rate_limits_tx: &broadcast::Sender<CodexRateLimitSnapshot>,
    pending: &PendingRequests,
    stdin: Option<&CodexStdin>,
    active_turn_id: Option<&Arc<tokio::sync::Mutex<Option<String>>>>,
    manual_compaction_in_flight: Option<&Arc<AtomicBool>>,
    turn_output: Option<&TurnOutputBuffer>,
    message: JsonRpcMessage,
) {
    match message {
        JsonRpcMessage::Response(response) => {
            let request = {
                let mut pending = pending.lock().await;
                response.id.as_ref().and_then(|id| pending.remove(id))
            };
            if let Some(request) = request {
                let method = request.method.clone();
                if request.tx.send(Ok(response)).is_err() {
                    tracing::warn!(
                        target: "claudette::agent",
                        subsystem = "codex-app-server",
                        pid,
                        method,
                        "codex response receiver dropped"
                    );
                }
            } else {
                tracing::warn!(
                    target: "claudette::agent",
                    subsystem = "codex-app-server",
                    pid,
                id = ?response.id,
                    "orphan codex response"
                );
            }
        }
        JsonRpcMessage::Error(error) => {
            let request = {
                let mut pending = pending.lock().await;
                error.id.as_ref().and_then(|id| pending.remove(id))
            };
            if let Some(request) = request {
                let method = request.method.clone();
                if request.tx.send(Err(error)).is_err() {
                    tracing::warn!(
                        target: "claudette::agent",
                        subsystem = "codex-app-server",
                        pid,
                        method,
                        "codex error receiver dropped"
                    );
                }
            } else {
                tracing::warn!(
                    target: "claudette::agent",
                    subsystem = "codex-app-server",
                    pid,
                id = ?error.id,
                    "orphan codex error"
                );
            }
        }
        JsonRpcMessage::Notification(notification) => {
            let notification = decode_notification(notification);
            // Rate-limit snapshots also go to a dedicated side-channel
            // so the AppState cache writer can subscribe without
            // walking every AgentEvent. `send` only fails when there
            // are no subscribers, which is the expected steady state
            // for non-chat sessions (discovery probe, model listing).
            if let CodexNotificationEvent::RateLimitsUpdated { rate_limits } = &notification {
                let _ = rate_limits_tx.send(rate_limits.clone());
            }
            // Any per-item drains triggered by this notification (item-id
            // change inside a delta, or item/completed for an
            // agentMessage/reasoning item) must be emitted BEFORE the
            // delta events from `map_notification_to_agent_events_for_route`
            // so the flushed assistant content stays chronologically
            // ordered with the new item's first delta.
            if let Some(turn_output) = turn_output {
                for event in update_turn_output_buffer(turn_output, &notification).await {
                    let _ = event_tx.send(event);
                }
            }
            if notification_finishes_turn(&notification)
                && let Some(active_turn_id) = active_turn_id
            {
                *active_turn_id.lock().await = None;
            }
            if notification_finishes_turn(&notification)
                && let Some(manual_compaction_in_flight) = manual_compaction_in_flight
            {
                manual_compaction_in_flight.store(false, Ordering::SeqCst);
            }
            if notification_finishes_turn(&notification)
                && let Some(turn_output) = turn_output
            {
                for event in drain_turn_output_buffer(turn_output).await {
                    let _ = event_tx.send(event);
                }
            }
            let events = map_notification_to_agent_events_for_route(
                notification,
                turn_output,
                manual_compaction_in_flight,
            )
            .await;
            for event in events {
                let _ = event_tx.send(event);
            }
        }
        JsonRpcMessage::Request(request) => {
            let method = request.method.clone();
            match codex_server_request_to_control_event(&request) {
                Ok(Some(event)) => {
                    let _ = event_tx.send(event);
                    tracing::debug!(
                        target: "claudette::agent",
                        subsystem = "codex-app-server",
                        pid,
                        method = %method,
                        "routed codex app-server server request to host approval prompt"
                    );
                }
                Ok(None) => {
                    if let Some(stdin) = stdin {
                        let mut stdin = stdin.lock().await;
                        let error = JsonRpcMessage::Error(JsonRpcError {
                            id: Some(request.id),
                            error: JsonRpcErrorBody {
                                code: -32601,
                                message: format!(
                                    "Codex app-server request `{method}` is not implemented"
                                ),
                                data: None,
                            },
                        });
                        if let Err(err) = write_jsonrpc_message(&mut *stdin, &error).await {
                            tracing::warn!(
                                target: "claudette::agent",
                                subsystem = "codex-app-server",
                                pid,
                                error = %err,
                                "failed to write codex server-request error response"
                            );
                        }
                    }
                    let _ = event_tx.send(AgentEvent::Stderr(format!(
                        "Codex app-server request `{method}` is not handled yet."
                    )));
                }
                Err(err) => {
                    tracing::warn!(
                        target: "claudette::agent",
                        subsystem = "codex-app-server",
                        pid,
                        error = %err,
                        method = %method,
                        "failed to route codex app-server server request"
                    );
                    if let Some(stdin) = stdin {
                        let mut stdin = stdin.lock().await;
                        let error = JsonRpcMessage::Error(JsonRpcError {
                            id: Some(request.id),
                            error: JsonRpcErrorBody {
                                code: -32603,
                                message: err,
                                data: None,
                            },
                        });
                        let _ = write_jsonrpc_message(&mut *stdin, &error).await;
                    }
                }
            }
        }
    }
}

/// User-facing message shown in chat when the Codex app-server's refresh
/// token has been rotated out from under us. The exact phrasing matches
/// what the UI surfaces verbatim — keep it short and actionable.
pub(crate) const CODEX_AUTH_EXPIRED_MESSAGE: &str = "Codex authentication expired. Run /login (or `codex login` in a terminal), then send the message again.";

/// Recognise stderr lines emitted by Codex's `codex_login::auth::manager`
/// and `codex_api::endpoint::responses_websocket` subsystems that indicate
/// the cached OAuth credentials are dead and a re-login is required.
/// Substring match because the lines are wrapped in ANSI escapes and a
/// timestamp prefix that varies per call.
pub(crate) fn codex_stderr_indicates_auth_expiry(line: &str) -> bool {
    line.contains("Failed to refresh token")
        || line.contains("Please log out and sign in again")
        || (line.contains("401 Unauthorized")
            && line.contains("wss://chatgpt.com/backend-api/codex/responses"))
}

async fn fail_pending_requests(pending: &PendingRequests, reason: &str) {
    let drained = {
        let mut pending = pending.lock().await;
        std::mem::take(&mut *pending)
    };
    for (id, request) in drained {
        let _ = request.tx.send(Err(JsonRpcError {
            id: Some(id),
            error: JsonRpcErrorBody {
                code: -32000,
                message: reason.to_string(),
                data: Some(json!({ "method": request.method })),
            },
        }));
    }
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(untagged)]
pub enum JsonRpcId {
    Integer(i64),
    String(String),
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(untagged)]
pub enum JsonRpcMessage {
    Request(JsonRpcRequest),
    Notification(JsonRpcNotification),
    Response(JsonRpcResponse),
    Error(JsonRpcError),
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct JsonRpcRequest {
    pub id: JsonRpcId,
    pub method: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub params: Option<Value>,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct JsonRpcNotification {
    pub method: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub params: Option<Value>,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct JsonRpcResponse {
    pub id: Option<JsonRpcId>,
    pub result: Value,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct JsonRpcError {
    pub id: Option<JsonRpcId>,
    pub error: JsonRpcErrorBody,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct JsonRpcErrorBody {
    pub code: i64,
    pub message: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub data: Option<Value>,
}

#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct JsonRpcServerResponse {
    pub id: JsonRpcId,
    pub result: Value,
}

#[derive(Debug, Clone, PartialEq)]
pub enum CodexRoutedMessage {
    Response {
        method: String,
        response: JsonRpcResponse,
    },
    Error {
        method: String,
        error: JsonRpcError,
    },
    Notification(CodexNotificationEvent),
    ServerRequest(JsonRpcRequest),
    OrphanResponse(JsonRpcResponse),
    OrphanError(JsonRpcError),
}

#[derive(Debug, Default)]
pub struct CodexResponseRouter {
    pending: BTreeMap<JsonRpcId, String>,
}

impl CodexResponseRouter {
    pub fn track_request(&mut self, request: &JsonRpcRequest) {
        self.pending
            .insert(request.id.clone(), request.method.clone());
    }

    pub fn pending_len(&self) -> usize {
        self.pending.len()
    }

    pub fn route(&mut self, message: JsonRpcMessage) -> CodexRoutedMessage {
        match message {
            JsonRpcMessage::Response(response) => {
                let request = response.id.as_ref().and_then(|id| self.pending.remove(id));
                match request {
                    Some(method) => CodexRoutedMessage::Response { method, response },
                    None => CodexRoutedMessage::OrphanResponse(response),
                }
            }
            JsonRpcMessage::Error(error) => {
                let request = error.id.as_ref().and_then(|id| self.pending.remove(id));
                match request {
                    Some(method) => CodexRoutedMessage::Error { method, error },
                    None => CodexRoutedMessage::OrphanError(error),
                }
            }
            JsonRpcMessage::Notification(notification) => {
                CodexRoutedMessage::Notification(decode_notification(notification))
            }
            JsonRpcMessage::Request(request) => CodexRoutedMessage::ServerRequest(request),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CodexPermissionLevel {
    Readonly,
    Standard,
    Full,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum CodexApprovalPolicy {
    #[serde(rename = "untrusted")]
    UnlessTrusted,
    OnRequest,
    Never,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum CodexSandboxMode {
    ReadOnly,
    WorkspaceWrite,
    DangerFullAccess,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum CodexSandboxPolicy {
    DangerFullAccess,
    #[serde(rename_all = "camelCase")]
    ReadOnly {
        network_access: bool,
    },
    #[serde(rename_all = "camelCase")]
    WorkspaceWrite {
        network_access: bool,
    },
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CodexPermissionMapping {
    pub approval_policy: CodexApprovalPolicy,
    pub thread_sandbox: CodexSandboxMode,
    pub turn_sandbox_policy: CodexSandboxPolicy,
}

impl CodexPermissionLevel {
    pub fn from_claudette_level(level: &str) -> Self {
        match level {
            "readonly" => Self::Readonly,
            "standard" => Self::Standard,
            "full" => Self::Full,
            _ => Self::Readonly,
        }
    }

    pub fn mapping(self) -> CodexPermissionMapping {
        match self {
            Self::Readonly => CodexPermissionMapping {
                approval_policy: CodexApprovalPolicy::UnlessTrusted,
                thread_sandbox: CodexSandboxMode::ReadOnly,
                turn_sandbox_policy: CodexSandboxPolicy::ReadOnly {
                    network_access: false,
                },
            },
            Self::Standard => CodexPermissionMapping {
                approval_policy: CodexApprovalPolicy::OnRequest,
                thread_sandbox: CodexSandboxMode::WorkspaceWrite,
                turn_sandbox_policy: CodexSandboxPolicy::WorkspaceWrite {
                    network_access: false,
                },
            },
            Self::Full => CodexPermissionMapping {
                approval_policy: CodexApprovalPolicy::Never,
                thread_sandbox: CodexSandboxMode::DangerFullAccess,
                turn_sandbox_policy: CodexSandboxPolicy::DangerFullAccess,
            },
        }
    }
}

pub fn parse_jsonrpc_line(line: &str) -> Result<JsonRpcMessage, serde_json::Error> {
    serde_json::from_str(line)
}

pub async fn read_jsonrpc_message<R>(reader: &mut R) -> Result<Option<JsonRpcMessage>, String>
where
    R: AsyncBufRead + Unpin,
{
    let mut line = String::new();
    loop {
        line.clear();
        let bytes = reader
            .read_line(&mut line)
            .await
            .map_err(|e| format!("Failed to read Codex app-server message: {e}"))?;
        if bytes == 0 {
            return Ok(None);
        }
        let trimmed = line.trim();
        if trimmed.is_empty() {
            continue;
        }
        return parse_jsonrpc_line(trimmed)
            .map(Some)
            .map_err(|e| format!("Failed to parse Codex app-server JSON-RPC line: {e}"));
    }
}

pub async fn write_jsonrpc_message<W, M>(writer: &mut W, message: &M) -> Result<(), String>
where
    W: AsyncWrite + Unpin,
    M: Serialize + ?Sized,
{
    let payload = serde_json::to_string(message)
        .map_err(|e| format!("Failed to encode Codex app-server JSON-RPC message: {e}"))?;
    writer
        .write_all(payload.as_bytes())
        .await
        .map_err(|e| format!("Failed to write Codex app-server JSON-RPC message: {e}"))?;
    writer
        .write_all(b"\n")
        .await
        .map_err(|e| format!("Failed to write Codex app-server JSON-RPC newline: {e}"))?;
    writer
        .flush()
        .await
        .map_err(|e| format!("Failed to flush Codex app-server JSON-RPC message: {e}"))?;
    Ok(())
}

pub const CODEX_COMMAND_APPROVAL_TOOL: &str = "CodexCommandApproval";
pub const CODEX_FILE_CHANGE_APPROVAL_TOOL: &str = "CodexFileChangeApproval";
pub const CODEX_PERMISSIONS_APPROVAL_TOOL: &str = "CodexPermissionsApproval";

pub fn is_codex_approval_tool_name(tool_name: &str) -> bool {
    matches!(
        tool_name,
        CODEX_COMMAND_APPROVAL_TOOL
            | CODEX_FILE_CHANGE_APPROVAL_TOOL
            | CODEX_PERMISSIONS_APPROVAL_TOOL
    )
}

fn codex_server_request_tool(method: &str) -> Option<(&'static str, &'static str)> {
    match method {
        "item/commandExecution/requestApproval" => {
            Some((CODEX_COMMAND_APPROVAL_TOOL, "commandExecution"))
        }
        "item/fileChange/requestApproval" => Some((CODEX_FILE_CHANGE_APPROVAL_TOOL, "fileChange")),
        "item/permissions/requestApproval" => {
            Some((CODEX_PERMISSIONS_APPROVAL_TOOL, "permissions"))
        }
        "item/tool/requestUserInput" => Some(("AskUserQuestion", "userInput")),
        "mcpServer/elicitation/request" => Some(("AskUserQuestion", "mcpElicitation")),
        _ => None,
    }
}

pub fn is_supported_codex_server_request(method: &str) -> bool {
    codex_server_request_tool(method).is_some()
}

fn codex_request_id_as_control_id(id: &JsonRpcId) -> Result<String, String> {
    serde_json::to_string(id).map_err(|e| format!("Failed to encode Codex request id: {e}"))
}

fn codex_tool_use_id(request: &JsonRpcRequest, request_id: &str) -> String {
    request
        .params
        .as_ref()
        .and_then(Value::as_object)
        .and_then(|params| {
            params
                .get("itemId")
                .or_else(|| params.get("approvalId"))
                .or_else(|| params.get("id"))
                .and_then(Value::as_str)
        })
        .map(ToOwned::to_owned)
        .unwrap_or_else(|| format!("codex-approval-{request_id}"))
}

pub fn codex_server_request_to_control_event(
    request: &JsonRpcRequest,
) -> Result<Option<AgentEvent>, String> {
    let Some((tool_name, approval_kind)) = codex_server_request_tool(&request.method) else {
        return Ok(None);
    };
    let request_id = codex_request_id_as_control_id(&request.id)?;
    let mut input = request
        .params
        .as_ref()
        .and_then(Value::as_object)
        .cloned()
        .unwrap_or_default();
    if request.method == "mcpServer/elicitation/request" {
        input = codex_mcp_elicitation_as_question_input(&input);
    }
    input.insert(
        "codexMethod".to_string(),
        Value::String(request.method.clone()),
    );
    input.insert(
        "codexApprovalKind".to_string(),
        Value::String(approval_kind.to_string()),
    );
    let tool_use_id = codex_tool_use_id(request, &request_id);
    Ok(Some(AgentEvent::Stream(StreamEvent::ControlRequest {
        request_id,
        request: ControlRequestInner::CanUseTool {
            tool_name: tool_name.to_string(),
            tool_use_id,
            input: Value::Object(input),
        },
    })))
}

fn codex_mcp_elicitation_as_question_input(
    params: &serde_json::Map<String, Value>,
) -> serde_json::Map<String, Value> {
    let request = params.get("request").and_then(Value::as_object);
    let message = request
        .and_then(|request| request.get("message"))
        .and_then(Value::as_str)
        .filter(|message| !message.trim().is_empty())
        .unwrap_or("Codex is requesting additional input from an MCP server.");
    let header = params
        .get("serverName")
        .and_then(Value::as_str)
        .filter(|server_name| !server_name.trim().is_empty())
        .map(|server_name| format!("MCP server: {server_name}"))
        .unwrap_or_else(|| "MCP server request".to_string());
    let mut input = params.clone();
    input.insert(
        "questions".to_string(),
        json!([{
            "id": "action",
            "header": header,
            "question": message,
            "options": [
                { "label": "Accept", "description": "Allow Codex to continue with this MCP request." },
                { "label": "Decline", "description": "Deny this MCP request." },
                { "label": "Cancel", "description": "Cancel the pending MCP request." }
            ]
        }]),
    );
    input
}

pub fn build_codex_server_response_from_control_response(
    request_id: &str,
    response: Value,
) -> Result<JsonRpcServerResponse, String> {
    let id = serde_json::from_str::<JsonRpcId>(request_id)
        .map_err(|e| format!("Invalid Codex server request id `{request_id}`: {e}"))?;
    let method = response
        .get("codexMethod")
        .and_then(Value::as_str)
        .ok_or_else(|| "Codex control response is missing codexMethod".to_string())?;
    if !is_supported_codex_server_request(method) {
        return Err(format!("Unsupported Codex server request `{method}`"));
    }
    let payload = response
        .get("response")
        .cloned()
        .ok_or_else(|| "Codex control response is missing response payload".to_string())?;
    Ok(JsonRpcServerResponse {
        id,
        result: payload,
    })
}

pub fn build_codex_user_input_response_payload(
    original_input: &Value,
    answers: &std::collections::HashMap<String, String>,
) -> Result<Value, String> {
    let method = original_input
        .get("codexMethod")
        .and_then(Value::as_str)
        .ok_or_else(|| "Codex user-input payload is missing codexMethod".to_string())?;

    match method {
        "item/tool/requestUserInput" => {
            let mut by_id = serde_json::Map::new();
            if let Some(questions) = original_input.get("questions").and_then(Value::as_array) {
                for question in questions {
                    let Some(id) = question.get("id").and_then(Value::as_str) else {
                        continue;
                    };
                    let Some(question_text) = question.get("question").and_then(Value::as_str)
                    else {
                        continue;
                    };
                    if let Some(answer) = answers.get(question_text) {
                        let values = answer
                            .split(',')
                            .map(str::trim)
                            .filter(|value| !value.is_empty())
                            .map(|value| Value::String(value.to_string()))
                            .collect::<Vec<_>>();
                        by_id.insert(id.to_string(), json!({ "answers": values }));
                    }
                }
            }
            Ok(json!({
                "codexMethod": method,
                "response": {
                    "answers": by_id,
                },
            }))
        }
        "mcpServer/elicitation/request" => {
            let action = answers
                .values()
                .next()
                .map(|answer| answer.trim().to_ascii_lowercase())
                .filter(|answer| answer == "accept" || answer == "cancel")
                .unwrap_or_else(|| "decline".to_string());
            Ok(json!({
                "codexMethod": method,
                "response": {
                    "action": action,
                    "content": null,
                    "_meta": null,
                },
            }))
        }
        _ => Err(format!(
            "Pending Codex request `{method}` is not a user-input request"
        )),
    }
}

pub fn build_codex_approval_response_payload(
    tool_name: &str,
    original_input: &Value,
    approved: bool,
) -> Result<Value, String> {
    let method = original_input
        .get("codexMethod")
        .and_then(Value::as_str)
        .ok_or_else(|| "Codex approval input is missing codexMethod".to_string())?;
    match tool_name {
        CODEX_COMMAND_APPROVAL_TOOL | CODEX_FILE_CHANGE_APPROVAL_TOOL => Ok(json!({
            "codexMethod": method,
            "response": {
                "decision": if approved { "accept" } else { "decline" },
            },
        })),
        CODEX_PERMISSIONS_APPROVAL_TOOL => {
            let permissions = if approved {
                original_input
                    .get("permissions")
                    .cloned()
                    .unwrap_or_else(|| json!({}))
            } else {
                json!({})
            };
            Ok(json!({
                "codexMethod": method,
                "response": {
                    "permissions": permissions,
                    "scope": "turn",
                },
            }))
        }
        _ => Err(format!(
            "Pending tool `{tool_name}` is not a Codex approval"
        )),
    }
}

pub fn build_codex_server_request_response(
    request: &JsonRpcRequest,
) -> Option<JsonRpcServerResponse> {
    let (tool_name, _) = codex_server_request_tool(&request.method)?;
    let mut original_input = request
        .params
        .as_ref()
        .and_then(Value::as_object)
        .cloned()
        .unwrap_or_default();
    original_input.insert(
        "codexMethod".to_string(),
        Value::String(request.method.clone()),
    );
    let response = if is_codex_approval_tool_name(tool_name) {
        build_codex_approval_response_payload(tool_name, &Value::Object(original_input), false)
            .ok()?
    } else {
        build_codex_user_input_response_payload(
            &Value::Object(original_input),
            &std::collections::HashMap::new(),
        )
        .ok()?
    };
    let request_id = codex_request_id_as_control_id(&request.id).ok()?;
    build_codex_server_response_from_control_response(&request_id, response).ok()
}

pub fn codex_app_server_args() -> [&'static str; 3] {
    ["app-server", "--listen", "stdio://"]
}

pub fn build_initialize_request(id: i64, client_version: &str) -> JsonRpcRequest {
    JsonRpcRequest {
        id: JsonRpcId::Integer(id),
        method: "initialize".to_string(),
        params: Some(json!({
            "clientInfo": {
                "name": "claudette",
                "title": "Claudette",
                "version": client_version,
            },
            "capabilities": {
                "experimentalApi": true,
            },
        })),
    }
}

pub fn build_initialized_notification() -> JsonRpcNotification {
    JsonRpcNotification {
        method: "initialized".to_string(),
        params: None,
    }
}

#[derive(Debug, Clone, Copy)]
pub struct CodexThreadRequestParams<'a> {
    pub model: Option<&'a str>,
    pub cwd: &'a Path,
    pub permission_level: CodexPermissionLevel,
    pub fast_mode: bool,
    pub reasoning_effort: Option<&'a str>,
    pub custom_instructions: Option<&'a str>,
    pub mcp_config: Option<&'a str>,
}

pub fn build_thread_start_request(id: i64, params: CodexThreadRequestParams<'_>) -> JsonRpcRequest {
    let mapping = params.permission_level.mapping();
    let service_tier = params.fast_mode.then_some("priority");
    let config = codex_thread_config(params.reasoning_effort, params.mcp_config);
    JsonRpcRequest {
        id: JsonRpcId::Integer(id),
        method: "thread/start".to_string(),
        params: Some(json!({
            "model": params.model,
            "modelProvider": "openai",
            "serviceTier": service_tier,
            "cwd": params.cwd,
            "approvalPolicy": mapping.approval_policy,
            "approvalsReviewer": "user",
            "sandbox": mapping.thread_sandbox,
            "threadSource": "user",
            "config": config,
            "developerInstructions": params.custom_instructions,
        })),
    }
}

pub fn build_thread_resume_request(
    id: i64,
    thread_id: &str,
    params: CodexThreadRequestParams<'_>,
) -> JsonRpcRequest {
    let mapping = params.permission_level.mapping();
    let service_tier = params.fast_mode.then_some("priority");
    let config = codex_thread_config(params.reasoning_effort, params.mcp_config);
    JsonRpcRequest {
        id: JsonRpcId::Integer(id),
        method: "thread/resume".to_string(),
        params: Some(json!({
            "threadId": thread_id,
            "model": params.model,
            "modelProvider": "openai",
            "serviceTier": service_tier,
            "cwd": params.cwd,
            "approvalPolicy": mapping.approval_policy,
            "approvalsReviewer": "user",
            "sandbox": mapping.thread_sandbox,
            "config": config,
            "developerInstructions": params.custom_instructions,
        })),
    }
}

fn codex_thread_config(reasoning_effort: Option<&str>, mcp_config: Option<&str>) -> Option<Value> {
    let mut config = serde_json::Map::new();
    if let Some(effort) = reasoning_effort {
        config.insert("model_reasoning_effort".to_string(), json!(effort));
    }
    if let Some(servers) = mcp_config.and_then(codex_mcp_servers_from_claude_config) {
        config.insert("mcp_servers".to_string(), servers);
    }
    (!config.is_empty()).then_some(Value::Object(config))
}

fn codex_mcp_servers_from_claude_config(mcp_config: &str) -> Option<Value> {
    let parsed: Value = serde_json::from_str(mcp_config).ok()?;
    let servers = parsed
        .get("mcpServers")
        .or_else(|| parsed.get("mcp_servers"))?
        .as_object()?;
    let mut codex_servers = serde_json::Map::new();
    for (name, server) in servers {
        let mut server = server.clone();
        if let Value::Object(ref mut object) = server {
            object.remove("type");
        }
        codex_servers.insert(name.clone(), server);
    }
    Some(Value::Object(codex_servers))
}

pub struct CodexTurnStartRequest<'a> {
    pub id: i64,
    pub thread_id: &'a str,
    pub prompt: &'a str,
    pub cwd: &'a Path,
    pub model: Option<&'a str>,
    pub permission_level: CodexPermissionLevel,
    pub plan_mode: bool,
    pub fast_mode: bool,
    pub reasoning_effort: Option<&'a str>,
    pub attachments: &'a [FileAttachment],
}

pub fn build_turn_start_request(params: CodexTurnStartRequest<'_>) -> JsonRpcRequest {
    let mapping = params.permission_level.mapping();
    let service_tier = params.fast_mode.then_some("priority");
    let collaboration_mode =
        codex_collaboration_mode(params.plan_mode, params.model, params.reasoning_effort);
    JsonRpcRequest {
        id: JsonRpcId::Integer(params.id),
        method: "turn/start".to_string(),
        params: Some(json!({
            "threadId": params.thread_id,
            "input": build_codex_user_input(params.prompt, params.attachments),
            "cwd": params.cwd,
            "approvalPolicy": mapping.approval_policy,
            "approvalsReviewer": "user",
            "sandboxPolicy": mapping.turn_sandbox_policy,
            "model": params.model,
            "serviceTier": service_tier,
            "effort": params.reasoning_effort,
            "collaborationMode": collaboration_mode,
        })),
    }
}

fn codex_collaboration_mode(
    plan_mode: bool,
    model: Option<&str>,
    reasoning_effort: Option<&str>,
) -> Option<Value> {
    let model = model?;
    Some(json!({
        "mode": if plan_mode { "plan" } else { "default" },
        "settings": {
            "model": model,
            "reasoning_effort": reasoning_effort,
            "developer_instructions": Value::Null,
        },
    }))
}

pub fn normalize_codex_reasoning_effort(effort: Option<&str>) -> Option<String> {
    match effort.map(str::trim).filter(|effort| !effort.is_empty()) {
        Some(value @ ("low" | "medium" | "high" | "xhigh")) => Some(value.to_string()),
        Some("auto" | "default" | "none" | "minimal" | "max") | None => Some("high".to_string()),
        Some(_) => None,
    }
}

pub fn build_turn_steer_request(
    id: i64,
    thread_id: &str,
    expected_turn_id: &str,
    prompt: &str,
    attachments: &[FileAttachment],
) -> JsonRpcRequest {
    JsonRpcRequest {
        id: JsonRpcId::Integer(id),
        method: "turn/steer".to_string(),
        params: Some(json!({
            "threadId": thread_id,
            "expectedTurnId": expected_turn_id,
            "input": build_codex_user_input(prompt, attachments),
        })),
    }
}

fn validate_codex_attachments(attachments: &[FileAttachment]) -> Result<(), String> {
    for attachment in attachments {
        if attachment.text_content.is_some() || attachment.media_type.starts_with("image/") {
            continue;
        }
        let label = attachment.filename.as_deref().unwrap_or("attachment");
        return Err(format!(
            "Codex app-server only supports image and text attachments; `{label}` is {}",
            attachment.media_type
        ));
    }
    Ok(())
}

fn build_codex_user_input(prompt: &str, attachments: &[FileAttachment]) -> Vec<Value> {
    let mut input = Vec::with_capacity(attachments.len() + 1);

    input.push(json!({
        "type": "text",
        "text": prompt,
        "textElements": [],
    }));

    for attachment in attachments {
        if let Some(text) = attachment.text_content.as_deref() {
            let label = attachment.filename.as_deref().unwrap_or("file");
            input.push(json!({
                "type": "text",
                "text": format!("Content of `{label}`:\n```\n{text}\n```"),
                "textElements": [],
            }));
        } else if attachment.media_type.starts_with("image/") {
            input.push(json!({
                "type": "image",
                "url": format!(
                    "data:{};base64,{}",
                    attachment.media_type, attachment.data_base64
                ),
            }));
        }
    }

    input
}

pub fn build_turn_interrupt_request(id: i64, thread_id: &str, turn_id: &str) -> JsonRpcRequest {
    JsonRpcRequest {
        id: JsonRpcId::Integer(id),
        method: "turn/interrupt".to_string(),
        params: Some(json!({
            "threadId": thread_id,
            "turnId": turn_id,
        })),
    }
}

pub fn build_thread_compact_start_request(id: i64, thread_id: &str) -> JsonRpcRequest {
    JsonRpcRequest {
        id: JsonRpcId::Integer(id),
        method: "thread/compact/start".to_string(),
        params: Some(json!({
            "threadId": thread_id,
        })),
    }
}

pub fn build_account_read_request(id: i64, refresh_token: bool) -> JsonRpcRequest {
    JsonRpcRequest {
        id: JsonRpcId::Integer(id),
        method: "account/read".to_string(),
        params: Some(json!({
            "refreshToken": refresh_token,
        })),
    }
}

pub fn build_model_list_request(id: i64, cursor: Option<&str>) -> JsonRpcRequest {
    JsonRpcRequest {
        id: JsonRpcId::Integer(id),
        method: "model/list".to_string(),
        params: Some(json!({
            "cursor": cursor,
            "limit": 100,
            "includeHidden": false,
        })),
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CodexAppServerAccountStatus {
    pub authenticated: bool,
    pub requires_openai_auth: bool,
    pub account_type: Option<String>,
    pub email: Option<String>,
    pub plan_type: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CodexAppServerModel {
    pub id: String,
    pub label: String,
    pub hidden: bool,
    pub is_default: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CodexAppServerModelPage {
    pub models: Vec<CodexAppServerModel>,
    pub next_cursor: Option<String>,
}

// -- v2 rate-limit snapshot -------------------------------------------------
//
// Mirror of Codex's `account/rateLimits/read` response (v2 protocol —
// inspect via `codex app-server generate-json-schema --out <dir>`).
// We strip the schema down to the fields the usage meter actually
// renders: per-window utilization + reset time, optional credits
// balance, plan label. `rate_limits_by_limit_id` is kept as a HashMap
// so a future multi-meter UI (e.g. "Codex" + "openai" buckets side by
// side) can grow without further protocol work.

/// Per-window utilization snapshot. `used_percent` is on the 0-100
/// scale Codex returns. `resets_at` is unix millis; absent means
/// "no scheduled reset" (e.g. a credit bucket).
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CodexRateLimitWindow {
    pub used_percent: i32,
    #[serde(default)]
    pub resets_at: Option<i64>,
    /// Window length in minutes. 300 → "Session (5h)", 10080 → "Weekly",
    /// 43200 → "Monthly". Absent for windowless credit-only buckets.
    #[serde(default)]
    pub window_duration_mins: Option<i64>,
}

#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CodexCreditsSnapshot {
    /// Stringified dollar balance (e.g. `"0.103"`). Codex returns this
    /// as a string to avoid float precision loss for tiny balances.
    #[serde(default)]
    pub balance: Option<String>,
    pub has_credits: bool,
    /// `true` for plan tiers with no spend cap (e.g. enterprise).
    pub unlimited: bool,
}

#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CodexRateLimitSnapshot {
    #[serde(default)]
    pub limit_id: Option<String>,
    #[serde(default)]
    pub limit_name: Option<String>,
    #[serde(default)]
    pub plan_type: Option<String>,
    #[serde(default)]
    pub primary: Option<CodexRateLimitWindow>,
    #[serde(default)]
    pub secondary: Option<CodexRateLimitWindow>,
    #[serde(default)]
    pub credits: Option<CodexCreditsSnapshot>,
    /// Set when Codex has flagged the user as over a limit; consumers
    /// can surface this as an "exhausted" indicator state.
    #[serde(default)]
    pub rate_limit_reached_type: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct GetAccountRateLimitsResponse {
    /// Backward-compatible single-bucket view that mirrors the
    /// historical payload. Present even when `rate_limits_by_limit_id`
    /// is populated.
    pub rate_limits: CodexRateLimitSnapshot,
    /// Multi-bucket view keyed by metered `limit_id` (typically
    /// `"codex"`). `None` on older app-server builds that only emit
    /// the single-bucket view.
    #[serde(default)]
    pub rate_limits_by_limit_id: Option<std::collections::HashMap<String, CodexRateLimitSnapshot>>,
}

pub fn build_account_rate_limits_read_request(id: i64) -> JsonRpcRequest {
    JsonRpcRequest {
        id: JsonRpcId::Integer(id),
        method: "account/rateLimits/read".to_string(),
        params: None,
    }
}

pub fn rate_limits_from_response(
    response: &JsonRpcResponse,
) -> Result<GetAccountRateLimitsResponse, String> {
    serde_json::from_value(response.result.clone())
        .map_err(|e| format!("Failed to parse Codex rateLimits response: {e}"))
}

pub fn account_status_from_response(
    response: &JsonRpcResponse,
) -> Result<CodexAppServerAccountStatus, String> {
    let account = response.result.get("account").unwrap_or(&Value::Null);
    let requires_openai_auth = response
        .result
        .get("requiresOpenaiAuth")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let account_type = account
        .get("type")
        .and_then(Value::as_str)
        .map(str::to_string);
    Ok(CodexAppServerAccountStatus {
        authenticated: account_type.is_some(),
        requires_openai_auth,
        account_type,
        email: account
            .get("email")
            .and_then(Value::as_str)
            .map(str::to_string),
        plan_type: account
            .get("planType")
            .and_then(Value::as_str)
            .map(str::to_string),
    })
}

pub fn model_list_from_response(
    response: &JsonRpcResponse,
) -> Result<CodexAppServerModelPage, String> {
    let data = response
        .result
        .get("data")
        .and_then(Value::as_array)
        .ok_or("Codex app-server model/list response did not include `data`")?;
    let models = data
        .iter()
        .filter_map(|model| {
            let raw_id = model
                .get("model")
                .or_else(|| model.get("id"))
                .and_then(Value::as_str)?;
            let id = raw_id.trim();
            if id.is_empty() {
                return None;
            }
            let label = model
                .get("displayName")
                .and_then(Value::as_str)
                .filter(|label| !label.trim().is_empty())
                .unwrap_or(id);
            Some(CodexAppServerModel {
                id: id.to_string(),
                label: label.to_string(),
                hidden: model
                    .get("hidden")
                    .and_then(Value::as_bool)
                    .unwrap_or(false),
                is_default: model
                    .get("isDefault")
                    .and_then(Value::as_bool)
                    .unwrap_or(false),
            })
        })
        .collect();
    let next_cursor = response
        .result
        .get("nextCursor")
        .and_then(Value::as_str)
        .filter(|cursor| !cursor.trim().is_empty())
        .map(str::to_string);
    Ok(CodexAppServerModelPage {
        models,
        next_cursor,
    })
}

#[derive(Debug, Clone, PartialEq)]
pub struct CodexPlanStep {
    pub step: String,
    pub status: String,
}

#[derive(Debug, Clone, PartialEq)]
pub enum CodexNotificationEvent {
    AgentMessageDelta {
        thread_id: String,
        turn_id: String,
        item_id: String,
        delta: String,
    },
    ReasoningSummaryDelta {
        thread_id: String,
        turn_id: String,
        item_id: String,
        delta: String,
        summary_index: i64,
    },
    ReasoningTextDelta {
        thread_id: String,
        turn_id: String,
        item_id: String,
        delta: String,
        content_index: i64,
    },
    CommandOutputDelta {
        thread_id: String,
        turn_id: String,
        item_id: String,
        delta: String,
    },
    TokenUsageUpdated {
        thread_id: String,
        turn_id: String,
        usage: TokenUsage,
    },
    TurnCompleted {
        thread_id: String,
        turn_id: String,
        duration_ms: Option<i64>,
    },
    TurnFailed {
        thread_id: String,
        turn_id: Option<String>,
        message: String,
    },
    TurnPlanUpdated {
        thread_id: String,
        turn_id: String,
        explanation: Option<String>,
        plan: Vec<CodexPlanStep>,
    },
    ItemStarted {
        thread_id: String,
        turn_id: String,
        item: Value,
    },
    ItemCompleted {
        thread_id: String,
        turn_id: String,
        item: Value,
    },
    TurnDiffUpdated {
        thread_id: String,
        turn_id: String,
        diff: String,
    },
    /// Deprecated upstream — Codex now delivers compaction completion as a
    /// `ContextCompaction` thread item via `item/completed`. Kept as a
    /// transitional fallback so older Codex builds still produce the timeline
    /// divider when a `thread/compact/start` lands.
    ContextCompacted { thread_id: String, turn_id: String },
    /// Pushed by Codex when the account's rate-limit snapshot changes
    /// — typically after a turn lands or a window resets. The host
    /// caches the latest snapshot so the composer's usage meter can
    /// surface live quota data without polling.
    RateLimitsUpdated { rate_limits: CodexRateLimitSnapshot },
    Unknown {
        method: String,
        params: Option<Value>,
    },
}

fn notification_finishes_turn(event: &CodexNotificationEvent) -> bool {
    matches!(
        event,
        CodexNotificationEvent::TurnCompleted { .. } | CodexNotificationEvent::TurnFailed { .. }
    )
}

fn notification_completes_context_compaction(event: &CodexNotificationEvent) -> bool {
    match event {
        CodexNotificationEvent::ItemCompleted { item, .. } => {
            item.get("type").and_then(Value::as_str) == Some("contextCompaction")
        }
        CodexNotificationEvent::ContextCompacted { .. } => true,
        _ => false,
    }
}

fn notification_starts_context_compaction(event: &CodexNotificationEvent) -> bool {
    matches!(
        event,
        CodexNotificationEvent::ItemStarted { item, .. }
            if item.get("type").and_then(Value::as_str) == Some("contextCompaction")
    )
}

/// Accumulate streaming text/thinking into the per-turn buffer, and
/// return any `Assistant` events that should be flushed *before* the
/// caller emits the delta events for this notification.
///
/// We flush on two boundaries:
///   1. **Item-id change inside a delta.** A new `item_id` for the same
///      kind means the previous item is logically complete even if its
///      `item/completed` hasn't arrived yet.
///   2. **`ItemCompleted` for an `agentMessage` / `reasoning` item.**
///      The semantic end-of-item from Codex.
///
/// Each flush emits ONE `ContentBlock` of the matching kind, so the
/// persisted chat row ends up with multiple `ContentBlock::Text` /
/// `ContentBlock::Thinking` entries in chronological order — restoring
/// item boundaries on reload (issue #865).
async fn update_turn_output_buffer(
    buffer: &TurnOutputBuffer,
    notification: &CodexNotificationEvent,
) -> Vec<AgentEvent> {
    let mut buffer = buffer.lock().await;
    let mut flushed = Vec::new();
    match notification {
        CodexNotificationEvent::AgentMessageDelta { item_id, delta, .. } => {
            if buffer
                .text_item_id
                .as_deref()
                .is_some_and(|prev| prev != item_id)
                && let Some(event) = flush_text_buffer(&mut buffer)
            {
                flushed.push(event);
            }
            buffer.text_item_id = Some(item_id.clone());
            buffer.text.push_str(delta);
        }
        CodexNotificationEvent::ReasoningSummaryDelta { item_id, delta, .. }
        | CodexNotificationEvent::ReasoningTextDelta { item_id, delta, .. } => {
            if buffer
                .thinking_item_id
                .as_deref()
                .is_some_and(|prev| prev != item_id)
                && let Some(event) = flush_thinking_buffer(&mut buffer)
            {
                flushed.push(event);
            }
            buffer.thinking_item_id = Some(item_id.clone());
            buffer.thinking.push_str(delta);
        }
        CodexNotificationEvent::ItemCompleted { item, .. } => {
            let item_type = item.get("type").and_then(Value::as_str).unwrap_or_default();
            let item_id = string_field(item, "id");
            match item_type {
                "agentMessage" if buffer.text_item_id.as_deref() == Some(item_id.as_str()) => {
                    if let Some(event) = flush_text_buffer(&mut buffer) {
                        flushed.push(event);
                    }
                }
                "reasoning" if buffer.thinking_item_id.as_deref() == Some(item_id.as_str()) => {
                    if let Some(event) = flush_thinking_buffer(&mut buffer) {
                        flushed.push(event);
                    }
                }
                _ => {}
            }
        }
        _ => {}
    }
    flushed
}

/// Drain whatever's left at turn end as a safety net. Per-item flushes
/// above usually leave both buffers empty by the time TurnCompleted
/// fires, but if Codex ever omits the trailing `item/completed` we
/// don't want to lose the content.
async fn drain_turn_output_buffer(buffer: &TurnOutputBuffer) -> Vec<AgentEvent> {
    let mut buffer = buffer.lock().await;
    let mut events = Vec::new();
    if let Some(event) = flush_thinking_buffer(&mut buffer) {
        events.push(event);
    }
    if let Some(event) = flush_text_buffer(&mut buffer) {
        events.push(event);
    }
    buffer.command_outputs.clear();
    events
}

/// Emit the buffered assistant text as its own `Assistant` event with a
/// single `ContentBlock::Text`, then reset the text buffer + item_id.
/// Returns `None` if the buffer is whitespace-only (no point persisting
/// a blank message; we still clear the buffer so the next item starts
/// fresh).
fn flush_text_buffer(buffer: &mut CodexTurnOutput) -> Option<AgentEvent> {
    if buffer.text.trim().is_empty() {
        buffer.text.clear();
        buffer.text_item_id = None;
        return None;
    }
    let text = std::mem::take(&mut buffer.text);
    buffer.text_item_id = None;
    Some(AgentEvent::Stream(StreamEvent::Assistant {
        message: AssistantMessage {
            content: vec![ContentBlock::Text { text }],
        },
    }))
}

/// Same shape as `flush_text_buffer` but for reasoning — emits a single
/// `ContentBlock::Thinking` so the UI's `Thinking…` collapsible gets a
/// real per-item record instead of one giant late-arriving block.
fn flush_thinking_buffer(buffer: &mut CodexTurnOutput) -> Option<AgentEvent> {
    if buffer.thinking.trim().is_empty() {
        buffer.thinking.clear();
        buffer.thinking_item_id = None;
        return None;
    }
    let thinking = std::mem::take(&mut buffer.thinking);
    buffer.thinking_item_id = None;
    Some(AgentEvent::Stream(StreamEvent::Assistant {
        message: AssistantMessage {
            content: vec![ContentBlock::Thinking { thinking }],
        },
    }))
}

async fn map_notification_to_agent_events_for_route(
    event: CodexNotificationEvent,
    buffer: Option<&TurnOutputBuffer>,
    manual_compaction_in_flight: Option<&Arc<AtomicBool>>,
) -> Vec<AgentEvent> {
    if let (CodexNotificationEvent::CommandOutputDelta { item_id, delta, .. }, Some(buffer)) =
        (&event, buffer)
        && !delta.is_empty()
    {
        let mut buffer = buffer.lock().await;
        let output = buffer.command_outputs.entry(item_id.clone()).or_default();
        output.push_str(delta);
        return vec![AgentEvent::Stream(StreamEvent::User {
            message: super::UserEventMessage {
                content: super::UserMessageContent::Blocks(vec![
                    super::UserContentBlock::ToolResult {
                        tool_use_id: item_id.clone(),
                        content: Value::String(output.clone()),
                    },
                ]),
            },
            uuid: None,
            is_replay: false,
            is_synthetic: true,
        })];
    }

    map_notification_to_agent_events_with_compact_state(event, manual_compaction_in_flight)
}

pub fn decode_notification(notification: JsonRpcNotification) -> CodexNotificationEvent {
    let params = notification.params;
    match (notification.method.as_str(), params.as_ref()) {
        ("item/agentMessage/delta", Some(params)) => CodexNotificationEvent::AgentMessageDelta {
            thread_id: string_field(params, "threadId"),
            turn_id: string_field(params, "turnId"),
            item_id: string_field(params, "itemId"),
            delta: string_field(params, "delta"),
        },
        ("item/reasoning/summaryTextDelta", Some(params)) => {
            CodexNotificationEvent::ReasoningSummaryDelta {
                thread_id: string_field(params, "threadId"),
                turn_id: string_field(params, "turnId"),
                item_id: string_field(params, "itemId"),
                delta: string_field(params, "delta"),
                summary_index: params
                    .get("summaryIndex")
                    .and_then(Value::as_i64)
                    .unwrap_or_default(),
            }
        }
        ("item/reasoning/textDelta", Some(params)) => CodexNotificationEvent::ReasoningTextDelta {
            thread_id: string_field(params, "threadId"),
            turn_id: string_field(params, "turnId"),
            item_id: string_field(params, "itemId"),
            delta: string_field(params, "delta"),
            content_index: params
                .get("contentIndex")
                .and_then(Value::as_i64)
                .unwrap_or_default(),
        },
        ("item/commandExecution/outputDelta", Some(params)) => {
            CodexNotificationEvent::CommandOutputDelta {
                thread_id: string_field(params, "threadId"),
                turn_id: string_field(params, "turnId"),
                item_id: string_field(params, "itemId"),
                delta: string_field(params, "delta"),
            }
        }
        ("thread/tokenUsage/updated", Some(params)) => CodexNotificationEvent::TokenUsageUpdated {
            thread_id: string_field(params, "threadId"),
            turn_id: string_field(params, "turnId"),
            usage: token_usage_from_params(params),
        },
        ("account/rateLimits/updated", Some(params)) => {
            // Notification body wraps the snapshot under a single
            // `rateLimits` field — extract and parse. On a malformed
            // payload we fall through to `Unknown` so the rest of the
            // stream keeps working; the cache just stays stale until
            // the next valid update.
            let rate_limits = params
                .get("rateLimits")
                .cloned()
                .and_then(|v| serde_json::from_value::<CodexRateLimitSnapshot>(v).ok());
            match rate_limits {
                Some(snapshot) => CodexNotificationEvent::RateLimitsUpdated {
                    rate_limits: snapshot,
                },
                None => CodexNotificationEvent::Unknown {
                    method: "account/rateLimits/updated".to_string(),
                    params: Some(params.clone()),
                },
            }
        }
        ("turn/completed", Some(params)) => {
            let turn = params.get("turn").unwrap_or(&Value::Null);
            CodexNotificationEvent::TurnCompleted {
                thread_id: string_field(params, "threadId"),
                turn_id: string_field(turn, "id"),
                duration_ms: turn.get("durationMs").and_then(Value::as_i64),
            }
        }
        ("turn/failed", Some(params)) => {
            let turn = params.get("turn");
            let error = turn
                .and_then(|turn| turn.get("error"))
                .or_else(|| params.get("error"))
                .unwrap_or(&Value::Null);
            let message = string_field(error, "message");
            CodexNotificationEvent::TurnFailed {
                thread_id: string_field(params, "threadId"),
                turn_id: turn.map(|turn| string_field(turn, "id")),
                message: if message.is_empty() {
                    "Codex turn failed".to_string()
                } else {
                    message
                },
            }
        }
        ("turn/plan/updated", Some(params)) => CodexNotificationEvent::TurnPlanUpdated {
            thread_id: string_field(params, "threadId"),
            turn_id: string_field(params, "turnId"),
            explanation: params
                .get("explanation")
                .and_then(Value::as_str)
                .map(str::to_string),
            plan: params
                .get("plan")
                .and_then(Value::as_array)
                .map(|steps| {
                    steps
                        .iter()
                        .filter_map(|step| {
                            let label = string_field(step, "step");
                            if label.is_empty() {
                                return None;
                            }
                            let raw_status = step
                                .get("status")
                                .and_then(Value::as_str)
                                .unwrap_or_default();
                            Some(CodexPlanStep {
                                step: label,
                                status: normalize_codex_plan_status(raw_status),
                            })
                        })
                        .collect()
                })
                .unwrap_or_default(),
        },
        ("item/started", Some(params)) => CodexNotificationEvent::ItemStarted {
            thread_id: string_field(params, "threadId"),
            turn_id: string_field(params, "turnId"),
            item: params.get("item").cloned().unwrap_or(Value::Null),
        },
        ("item/completed", Some(params)) => CodexNotificationEvent::ItemCompleted {
            thread_id: string_field(params, "threadId"),
            turn_id: string_field(params, "turnId"),
            item: params.get("item").cloned().unwrap_or(Value::Null),
        },
        ("turn/diff/updated", Some(params)) => CodexNotificationEvent::TurnDiffUpdated {
            thread_id: string_field(params, "threadId"),
            turn_id: string_field(params, "turnId"),
            diff: string_field(params, "diff"),
        },
        ("thread/compacted", Some(params)) => CodexNotificationEvent::ContextCompacted {
            thread_id: string_field(params, "threadId"),
            turn_id: string_field(params, "turnId"),
        },
        (method, params) => CodexNotificationEvent::Unknown {
            method: method.to_string(),
            params: params.cloned(),
        },
    }
}

pub fn map_notification_to_agent_events(event: CodexNotificationEvent) -> Vec<AgentEvent> {
    map_notification_to_agent_events_with_context(event, false, true)
}

fn map_notification_to_agent_events_with_compact_state(
    event: CodexNotificationEvent,
    manual_compaction_in_flight: Option<&Arc<AtomicBool>>,
) -> Vec<AgentEvent> {
    let finish_compaction_turn = notification_completes_context_compaction(&event)
        && manual_compaction_in_flight.is_some_and(|flag| flag.swap(false, Ordering::SeqCst));
    let emit_context_compacting_status = notification_starts_context_compaction(&event)
        && !manual_compaction_in_flight.is_some_and(|flag| flag.load(Ordering::SeqCst));

    map_notification_to_agent_events_with_context(
        event,
        finish_compaction_turn,
        emit_context_compacting_status,
    )
}

fn map_notification_to_agent_events_with_context(
    event: CodexNotificationEvent,
    finish_compaction_turn: bool,
    emit_context_compacting_status: bool,
) -> Vec<AgentEvent> {
    match event {
        CodexNotificationEvent::AgentMessageDelta { delta, .. } if !delta.is_empty() => {
            vec![AgentEvent::Stream(StreamEvent::Stream {
                event: InnerStreamEvent::ContentBlockDelta {
                    index: 0,
                    delta: Delta::Text { text: delta },
                },
            })]
        }
        CodexNotificationEvent::ReasoningSummaryDelta { delta, .. }
        | CodexNotificationEvent::ReasoningTextDelta { delta, .. }
            if !delta.is_empty() =>
        {
            vec![AgentEvent::Stream(StreamEvent::Stream {
                event: InnerStreamEvent::ContentBlockDelta {
                    index: 1,
                    delta: Delta::Thinking { thinking: delta },
                },
            })]
        }
        CodexNotificationEvent::CommandOutputDelta { item_id, delta, .. } if !delta.is_empty() => {
            vec![AgentEvent::Stream(StreamEvent::User {
                message: super::UserEventMessage {
                    content: super::UserMessageContent::Blocks(vec![
                        super::UserContentBlock::ToolResult {
                            tool_use_id: item_id,
                            content: Value::String(delta),
                        },
                    ]),
                },
                uuid: None,
                is_replay: false,
                is_synthetic: true,
            })]
        }
        CodexNotificationEvent::TokenUsageUpdated { usage, .. } => {
            vec![AgentEvent::Stream(StreamEvent::Stream {
                event: InnerStreamEvent::MessageDelta { usage: Some(usage) },
            })]
        }
        CodexNotificationEvent::TurnCompleted { duration_ms, .. } => {
            vec![AgentEvent::Stream(StreamEvent::Result {
                subtype: "success".to_string(),
                result: None,
                total_cost_usd: None,
                duration_ms,
                usage: None,
            })]
        }
        CodexNotificationEvent::TurnFailed { message, .. } => {
            vec![AgentEvent::Stream(StreamEvent::Result {
                subtype: "error".to_string(),
                result: Some(message),
                total_cost_usd: None,
                duration_ms: None,
                usage: None,
            })]
        }
        CodexNotificationEvent::TurnPlanUpdated {
            explanation, plan, ..
        } => map_codex_plan_updated_to_agent_events(explanation, plan),
        CodexNotificationEvent::ItemStarted { item, .. } => {
            map_codex_item_started_to_agent_events(&item, emit_context_compacting_status)
        }
        CodexNotificationEvent::ItemCompleted { item, .. } => {
            map_codex_item_completed_to_agent_events(&item, finish_compaction_turn)
        }
        // Transitional fallback: older Codex builds emit `thread/compacted`
        // as a notification rather than a `ContextCompaction` thread item.
        // Route it through the same manual-vs-auto decision as the item path.
        CodexNotificationEvent::ContextCompacted { .. } => {
            let mut events = vec![codex_context_compaction_event()];
            if finish_compaction_turn {
                events.push(codex_compaction_finish_event());
            }
            events
        }
        CodexNotificationEvent::AgentMessageDelta { .. }
        | CodexNotificationEvent::ReasoningSummaryDelta { .. }
        | CodexNotificationEvent::ReasoningTextDelta { .. }
        | CodexNotificationEvent::CommandOutputDelta { .. }
        | CodexNotificationEvent::TurnDiffUpdated { .. }
        // Rate-limit updates flow through `rate_limits_tx` (a dedicated
        // side-channel consumed by the AppState usage cache) — they
        // don't translate to AgentEvents in the chat stream.
        | CodexNotificationEvent::RateLimitsUpdated { .. }
            | CodexNotificationEvent::Unknown { .. } => Vec::new(),
    }
}

fn normalize_codex_plan_status(status: &str) -> String {
    match status {
        "pending" | "Pending" => "pending".to_string(),
        "completed" | "Completed" => "completed".to_string(),
        "in_progress" | "inProgress" | "InProgress" => "in_progress".to_string(),
        other => {
            tracing::warn!(
                target: "claudette::codex",
                status = %other,
                "unknown Codex plan step status"
            );
            other.to_string()
        }
    }
}

fn map_codex_plan_updated_to_agent_events(
    explanation: Option<String>,
    plan: Vec<CodexPlanStep>,
) -> Vec<AgentEvent> {
    let tool_use_id = format!("update_plan-{}", uuid::Uuid::new_v4());
    let input = json!({
        "explanation": explanation,
        "plan": plan
            .iter()
            .map(|step| json!({
                "step": step.step,
                "status": step.status,
            }))
            .collect::<Vec<_>>(),
    });
    let index = codex_item_index(&tool_use_id);

    vec![
        AgentEvent::Stream(StreamEvent::Stream {
            event: InnerStreamEvent::ContentBlockStart {
                index,
                content_block: Some(StartContentBlock::ToolUse {
                    id: tool_use_id.clone(),
                    name: "update_plan".to_string(),
                    input: None,
                }),
            },
        }),
        AgentEvent::Stream(StreamEvent::Stream {
            event: InnerStreamEvent::ContentBlockDelta {
                index,
                delta: Delta::ToolUse {
                    partial_json: Some(input.to_string()),
                },
            },
        }),
        AgentEvent::Stream(StreamEvent::Stream {
            event: InnerStreamEvent::ContentBlockStop { index },
        }),
        AgentEvent::Stream(StreamEvent::User {
            message: super::UserEventMessage {
                content: super::UserMessageContent::Blocks(vec![
                    super::UserContentBlock::ToolResult {
                        tool_use_id,
                        content: Value::String("Plan updated".to_string()),
                    },
                ]),
            },
            uuid: None,
            is_replay: false,
            is_synthetic: true,
        }),
    ]
}

fn map_codex_item_started_to_agent_events(
    item: &Value,
    emit_context_compacting_status: bool,
) -> Vec<AgentEvent> {
    if item.get("type").and_then(Value::as_str) == Some("contextCompaction") {
        return if emit_context_compacting_status {
            vec![codex_compacting_status_event()]
        } else {
            Vec::new()
        };
    }
    let Some((item_id, tool_name, input)) = codex_item_tool_use(item) else {
        return Vec::new();
    };
    let index = codex_item_index(&item_id);
    let item_type = item.get("type").and_then(Value::as_str);
    let mut events = vec![AgentEvent::Stream(StreamEvent::Stream {
        event: InnerStreamEvent::ContentBlockStart {
            index,
            content_block: Some(StartContentBlock::ToolUse {
                id: item_id,
                name: tool_name,
                input: None,
            }),
        },
    })];
    if item_type != Some("fileChange") {
        events.push(AgentEvent::Stream(StreamEvent::Stream {
            event: InnerStreamEvent::ContentBlockDelta {
                index,
                delta: Delta::ToolUse {
                    partial_json: Some(input.to_string()),
                },
            },
        }));
    }
    if item_type == Some("commandExecution") {
        events.push(codex_command_execution_status_event(
            item,
            item.get("status")
                .and_then(Value::as_str)
                .unwrap_or("running"),
        ));
    }
    events
}

fn map_codex_item_completed_to_agent_events(
    item: &Value,
    finish_compaction_turn: bool,
) -> Vec<AgentEvent> {
    // ContextCompaction items don't map to a tool use — Codex emits them as
    // the canonical signal that compaction landed. Translate into the existing
    // `compact_boundary` system event shape so `send.rs` persists the
    // `COMPACTION:...` sentinel via `build_compaction_sentinel`, and the
    // timeline divider renders. Only append the synthetic `Result` for an
    // explicit `/compact` turn; Codex's auto-compaction happens mid-turn and
    // must not terminate the live event pump.
    if item.get("type").and_then(Value::as_str) == Some("contextCompaction") {
        let mut events = vec![codex_context_compaction_event()];
        if finish_compaction_turn {
            events.push(codex_compaction_finish_event());
        }
        return events;
    }
    if item.get("type").and_then(Value::as_str) == Some("plan") {
        return map_codex_plan_item_to_agent_events(item);
    }
    let Some((item_id, _tool_name, input)) = codex_item_tool_use(item) else {
        return Vec::new();
    };
    let index = codex_item_index(&item_id);
    let mut events = Vec::new();
    if item.get("type").and_then(Value::as_str) == Some("commandExecution") {
        events.push(codex_command_execution_status_event(
            item,
            item.get("status")
                .and_then(Value::as_str)
                .unwrap_or("completed"),
        ));
    }
    if item.get("type").and_then(Value::as_str) == Some("fileChange") {
        events.push(AgentEvent::Stream(StreamEvent::Stream {
            event: InnerStreamEvent::ContentBlockDelta {
                index,
                delta: Delta::ToolUse {
                    partial_json: Some(input.to_string()),
                },
            },
        }));
    }
    events.extend([
        AgentEvent::Stream(StreamEvent::Stream {
            event: InnerStreamEvent::ContentBlockStop { index },
        }),
        AgentEvent::Stream(StreamEvent::User {
            message: super::UserEventMessage {
                content: super::UserMessageContent::Blocks(vec![
                    super::UserContentBlock::ToolResult {
                        tool_use_id: item_id,
                        content: codex_item_result_content(item),
                    },
                ]),
            },
            uuid: None,
            is_replay: false,
            is_synthetic: true,
        }),
    ]);
    events
}

fn map_codex_plan_item_to_agent_events(item: &Value) -> Vec<AgentEvent> {
    let plan_text = string_field(item, "text");
    if plan_text.trim().is_empty() {
        return Vec::new();
    }
    let item_id = string_field(item, "id");
    let tool_use_id = if item_id.is_empty() {
        format!("codex-plan-{}", uuid::Uuid::new_v4())
    } else {
        format!("codex-plan-{item_id}")
    };
    vec![
        AgentEvent::Stream(StreamEvent::Assistant {
            message: AssistantMessage {
                content: vec![ContentBlock::Text {
                    text: plan_text.clone(),
                }],
            },
        }),
        AgentEvent::Stream(StreamEvent::ControlRequest {
            request_id: format!("synthetic-codex-plan-{tool_use_id}"),
            request: ControlRequestInner::CanUseTool {
                tool_name: "ExitPlanMode".to_string(),
                tool_use_id,
                input: json!({
                    "codexSyntheticPlan": true,
                    "codexPlanContent": plan_text,
                    "allowedPrompts": [],
                }),
            },
        }),
    ]
}

/// Build the in-flight indicator for a Codex compaction. The frontend
/// `useAgentStream` listener flips `workspace.agent_status` to `"Compacting"`
/// when it sees this event, mirroring the Claude CLI's native
/// `subtype: "status", status: "compacting"` emission.
fn codex_compacting_status_event() -> AgentEvent {
    AgentEvent::Stream(StreamEvent::System {
        subtype: "status".to_string(),
        session_id: None,
        state: None,
        detail: None,
        task_id: None,
        tool_use_id: None,
        output_file: None,
        summary: None,
        description: None,
        last_tool_name: None,
        usage: None,
        status: Some("compacting".to_string()),
        compact_result: None,
        compact_metadata: None,
        command_line: None,
    })
}

/// Build the `compact_boundary` system event for a Codex `contextCompaction`
/// item. Codex's item carries only an `id`, not pre/post token counts or a
/// duration, so we synthesize a metadata stub with `trigger="codex"` and
/// zeros. The `CompactionDivider` recognizes the `codex` trigger and renders
/// without the misleading `0 → 0` token arrow.
///
/// `compact_result` stays `None` here: the established `StreamEvent::System`
/// contract reserves that field for an end-of-compaction *status* event
/// (Claude CLI's emission pattern). The `subtype: "compact_boundary"` event
/// carries its payload in `compact_metadata` only.
fn codex_context_compaction_event() -> AgentEvent {
    AgentEvent::Stream(StreamEvent::System {
        subtype: "compact_boundary".to_string(),
        session_id: None,
        state: None,
        detail: None,
        task_id: None,
        tool_use_id: None,
        output_file: None,
        summary: None,
        description: None,
        last_tool_name: None,
        usage: None,
        status: None,
        compact_result: None,
        compact_metadata: Some(crate::agent::types::CompactMetadata {
            trigger: "codex".to_string(),
            pre_tokens: 0,
            post_tokens: 0,
            duration_ms: 0,
        }),
        command_line: None,
    })
}

/// Synthetic terminal event emitted after `codex_context_compaction_event()`
/// so the per-turn pump that `start_compact()` set up (subscribed via
/// `event_tx`, consumed by `send_chat_message`'s stream loop) terminates and
/// flips `session.agent_status` back to `"Idle"`. Mirrors the
/// `subtype: "success"` Result that Claude's CLI emits at the end of its
/// `/compact` turn — without it, the UI would stay stuck in the Compacting
/// state indefinitely.
fn codex_compaction_finish_event() -> AgentEvent {
    AgentEvent::Stream(StreamEvent::Result {
        subtype: "success".to_string(),
        result: None,
        total_cost_usd: None,
        duration_ms: None,
        usage: None,
    })
}

fn codex_command_execution_status_event(item: &Value, status: &str) -> AgentEvent {
    let item_id = string_field(item, "id");
    let terminal_status = match status {
        "inProgress" | "starting" | "running" => "running",
        "failed" => "failed",
        "cancelled" | "canceled" => "cancelled",
        "completed" => "completed",
        other => other,
    };
    let command = item
        .get("command")
        .and_then(Value::as_str)
        .filter(|command| !command.trim().is_empty());
    let exit_code = item.get("exitCode").and_then(Value::as_i64);
    let summary = match (terminal_status, exit_code, command) {
        ("completed", Some(0), _) => Some("Command completed".to_string()),
        ("completed", Some(code), _) => Some(format!("Command failed ({code})")),
        ("failed", Some(code), _) => Some(format!("Command failed ({code})")),
        ("failed", None, _) => Some("Command failed".to_string()),
        (_, _, Some(command)) => Some(command.to_string()),
        _ => None,
    };

    AgentEvent::Stream(StreamEvent::System {
        subtype: "task_notification".to_string(),
        session_id: None,
        state: None,
        detail: None,
        task_id: Some(item_id.clone()),
        tool_use_id: Some(item_id),
        output_file: None,
        summary,
        description: None,
        last_tool_name: None,
        usage: None,
        status: Some(terminal_status.to_string()),
        compact_result: None,
        compact_metadata: None,
        command_line: None,
    })
}

fn codex_item_tool_use(item: &Value) -> Option<(String, String, Value)> {
    let item_type = item.get("type").and_then(Value::as_str)?;
    let item_id = string_field(item, "id");
    if item_id.is_empty() {
        return None;
    }
    match item_type {
        "commandExecution" => Some((
            item_id,
            "Bash".to_string(),
            json!({
                "command": item.get("command").and_then(Value::as_str).unwrap_or_default(),
                "cwd": item.get("cwd").and_then(Value::as_str),
            }),
        )),
        "fileChange" => Some((
            item_id,
            "Edit".to_string(),
            json!({
                "changes": item.get("changes").cloned().unwrap_or_else(|| json!([])),
            }),
        )),
        "mcpToolCall" => {
            let server = string_field(item, "server");
            let tool = string_field(item, "tool");
            Some((
                item_id,
                format!(
                    "mcp__{}__{}",
                    sanitize_tool_segment(&server),
                    sanitize_tool_segment(&tool)
                ),
                item.get("arguments").cloned().unwrap_or(Value::Null),
            ))
        }
        _ => None,
    }
}

fn codex_item_result_content(item: &Value) -> Value {
    match item.get("type").and_then(Value::as_str) {
        Some("commandExecution") => item
            .get("aggregatedOutput")
            .filter(|value| !value.is_null())
            .cloned()
            .unwrap_or_else(|| {
                json!({
                    "status": item.get("status").cloned().unwrap_or(Value::Null),
                    "exitCode": item.get("exitCode").cloned().unwrap_or(Value::Null),
                })
            }),
        Some("fileChange") => json!({
            "status": item.get("status").cloned().unwrap_or(Value::Null),
            "changes": item.get("changes").cloned().unwrap_or_else(|| json!([])),
        }),
        Some("mcpToolCall") => item
            .get("result")
            .filter(|value| !value.is_null())
            .cloned()
            .or_else(|| item.get("error").filter(|value| !value.is_null()).cloned())
            .unwrap_or_else(
                || json!({ "status": item.get("status").cloned().unwrap_or(Value::Null) }),
            ),
        _ => Value::Null,
    }
}

fn codex_item_index(item_id: &str) -> usize {
    let hash = item_id.bytes().fold(0_usize, |acc, byte| {
        acc.wrapping_mul(31).wrapping_add(byte as usize)
    });
    2 + (hash % 10_000)
}

fn sanitize_tool_segment(segment: &str) -> String {
    let sanitized = segment
        .chars()
        .map(|ch| {
            if ch.is_ascii_alphanumeric() || ch == '_' || ch == '-' {
                ch
            } else {
                '_'
            }
        })
        .collect::<String>();
    if sanitized.is_empty() {
        "unknown".to_string()
    } else {
        sanitized
    }
}

pub fn codex_invocation_line() -> String {
    format!("codex {}", codex_app_server_args().join(" "))
}

pub fn codex_command_line_event() -> AgentEvent {
    AgentEvent::Stream(StreamEvent::system_command_line(codex_invocation_line()))
}

pub fn codex_turn_start_events(thread_id: &str) -> Vec<AgentEvent> {
    vec![
        AgentEvent::Stream(StreamEvent::System {
            subtype: "init".to_string(),
            session_id: Some(thread_id.to_string()),
            state: None,
            detail: None,
            task_id: None,
            tool_use_id: None,
            output_file: None,
            summary: None,
            description: None,
            last_tool_name: None,
            usage: None,
            status: None,
            compact_result: None,
            compact_metadata: None,
            command_line: None,
        }),
        AgentEvent::Stream(StreamEvent::Stream {
            event: InnerStreamEvent::MessageStart {},
        }),
        AgentEvent::Stream(StreamEvent::Stream {
            event: InnerStreamEvent::ContentBlockStart {
                index: 0,
                content_block: Some(StartContentBlock::Text {}),
            },
        }),
        AgentEvent::Stream(StreamEvent::Stream {
            event: InnerStreamEvent::ContentBlockStart {
                index: 1,
                content_block: Some(StartContentBlock::Thinking {}),
            },
        }),
    ]
}

fn string_field(value: &Value, key: &str) -> String {
    value
        .get(key)
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_string()
}

fn thread_id_from_response(response: &JsonRpcResponse) -> Result<String, String> {
    response_id_from_response(response, "thread", "thread/start")
}

fn turn_id_from_response(response: &JsonRpcResponse) -> Result<String, String> {
    response_id_from_response(response, "turn", "turn/start")
}

fn response_id_from_response(
    response: &JsonRpcResponse,
    object_key: &str,
    method: &str,
) -> Result<String, String> {
    let id = response
        .result
        .get(object_key)
        .and_then(|value| value.get("id"))
        .and_then(Value::as_str)
        .unwrap_or_default()
        .trim()
        .to_string();
    if id.is_empty() {
        Err(format!(
            "Codex app-server `{method}` response did not include `{object_key}.id`"
        ))
    } else {
        Ok(id)
    }
}

fn token_usage_from_params(params: &Value) -> TokenUsage {
    let token_usage = params.get("tokenUsage").unwrap_or(&Value::Null);
    let total = token_usage.get("total").unwrap_or(&Value::Null);
    let last = token_usage.get("last").unwrap_or(total);
    let input_tokens = last
        .get("inputTokens")
        .and_then(Value::as_u64)
        .unwrap_or_default();
    let cached_input_tokens = last
        .get("cachedInputTokens")
        .and_then(Value::as_u64)
        .unwrap_or_default();
    let output_tokens = last
        .get("outputTokens")
        .and_then(Value::as_u64)
        .unwrap_or_default();
    let reasoning_output_tokens = last
        .get("reasoningOutputTokens")
        .and_then(Value::as_u64)
        .unwrap_or_default();
    let total_tokens = last.get("totalTokens").and_then(Value::as_u64);
    let model_context_window = token_usage
        .get("modelContextWindow")
        .and_then(Value::as_u64);
    TokenUsage {
        total_tokens,
        input_tokens: input_tokens.saturating_sub(cached_input_tokens),
        output_tokens: output_tokens + reasoning_output_tokens,
        cache_creation_input_tokens: None,
        cache_read_input_tokens: Some(cached_input_tokens),
        model_context_window,
        iterations: None,
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn tool_result_text(event: AgentEvent) -> String {
        let AgentEvent::Stream(StreamEvent::User { message, .. }) = event else {
            panic!("expected user tool result event");
        };
        let crate::agent::UserMessageContent::Blocks(blocks) = message.content else {
            panic!("expected structured user blocks");
        };
        let Some(crate::agent::UserContentBlock::ToolResult { content, .. }) =
            blocks.into_iter().next()
        else {
            panic!("expected tool result block");
        };
        content.as_str().expect("string content").to_string()
    }

    #[test]
    fn initialize_request_matches_codex_app_server_shape() {
        let request = build_initialize_request(1, "0.24.0");
        let value = serde_json::to_value(&request).expect("request serializes");

        assert_eq!(value["id"], 1);
        assert_eq!(value["method"], "initialize");
        assert_eq!(value["params"]["clientInfo"]["name"], "claudette");
        assert_eq!(value["params"]["capabilities"]["experimentalApi"], true);
        assert!(value.get("jsonrpc").is_none());
    }

    #[test]
    fn codex_app_server_invocation_uses_stdio_listener() {
        assert_eq!(
            codex_app_server_args(),
            ["app-server", "--listen", "stdio://"]
        );
        assert_eq!(
            codex_invocation_line(),
            "codex app-server --listen stdio://"
        );
    }

    #[tokio::test]
    async fn writes_jsonrpc_message_as_single_line() {
        let mut out = Vec::new();
        write_jsonrpc_message(
            &mut out,
            &JsonRpcMessage::Request(build_initialize_request(1, "0.24.0")),
        )
        .await
        .expect("message writes");

        let text = String::from_utf8(out).expect("utf8");
        assert!(text.ends_with('\n'));
        assert_eq!(text.lines().count(), 1);
        let parsed = parse_jsonrpc_line(text.trim()).expect("round trip parses");
        assert!(matches!(
            parsed,
            JsonRpcMessage::Request(JsonRpcRequest { method, .. }) if method == "initialize"
        ));
    }

    #[tokio::test]
    async fn writes_codex_server_response_as_jsonrpc_result() {
        let request = JsonRpcRequest {
            id: JsonRpcId::Integer(8),
            method: "item/commandExecution/requestApproval".to_string(),
            params: None,
        };
        let response =
            build_codex_server_request_response(&request).expect("approval request is handled");
        let mut out = Vec::new();

        write_jsonrpc_message(&mut out, &response)
            .await
            .expect("response writes");

        let value: Value = serde_json::from_slice(&out).expect("json response");
        assert_eq!(
            value,
            json!({
                "id": 8,
                "result": {
                    "decision": "decline"
                }
            })
        );
    }

    #[test]
    fn builds_codex_control_response_for_approved_command() {
        let response = build_codex_approval_response_payload(
            CODEX_COMMAND_APPROVAL_TOOL,
            &json!({
                "codexMethod": "item/commandExecution/requestApproval",
                "command": "cargo test",
            }),
            true,
        )
        .expect("response payload");
        let server_response =
            build_codex_server_response_from_control_response("8", response).expect("response");

        assert_eq!(server_response.id, JsonRpcId::Integer(8));
        assert_eq!(server_response.result, json!({ "decision": "accept" }));
        let value = serde_json::to_value(&server_response).unwrap();
        assert_eq!(
            value,
            json!({ "id": 8, "result": { "decision": "accept" } })
        );
    }

    #[test]
    fn builds_codex_control_response_for_denied_permission_grant() {
        let response = build_codex_approval_response_payload(
            CODEX_PERMISSIONS_APPROVAL_TOOL,
            &json!({
                "codexMethod": "item/permissions/requestApproval",
                "permissions": { "sandbox": "workspace-write" },
            }),
            false,
        )
        .expect("response payload");
        let server_response =
            build_codex_server_response_from_control_response("\"perm-1\"", response)
                .expect("response");

        assert_eq!(server_response.id, JsonRpcId::String("perm-1".to_string()));
        assert_eq!(
            server_response.result,
            json!({
                "permissions": {},
                "scope": "turn"
            })
        );
    }

    #[test]
    fn builds_codex_user_input_response_by_question_id() {
        let mut answers = std::collections::HashMap::new();
        answers.insert("Pick checks".to_string(), "Lint, Typecheck".to_string());
        answers.insert("Ship it?".to_string(), "Yes".to_string());

        let response = build_codex_user_input_response_payload(
            &json!({
                "codexMethod": "item/tool/requestUserInput",
                "questions": [
                    { "id": "checks", "question": "Pick checks" },
                    { "id": "ship", "question": "Ship it?" }
                ],
            }),
            &answers,
        )
        .expect("response payload");
        let server_response =
            build_codex_server_response_from_control_response("\"ask-1\"", response)
                .expect("server response");

        assert_eq!(
            server_response.result,
            json!({
                "answers": {
                    "checks": { "answers": ["Lint", "Typecheck"] },
                    "ship": { "answers": ["Yes"] }
                }
            })
        );
    }

    #[test]
    fn routes_codex_user_input_to_existing_question_tool() {
        let event = codex_server_request_to_control_event(&JsonRpcRequest {
            id: JsonRpcId::String("ask-1".to_string()),
            method: "item/tool/requestUserInput".to_string(),
            params: Some(json!({
                "itemId": "item-1",
                "questions": [{
                    "id": "q1",
                    "header": "Decision",
                    "question": "Proceed?",
                    "options": [{ "label": "Yes", "description": "Continue" }]
                }]
            })),
        })
        .expect("route")
        .expect("event");

        let AgentEvent::Stream(StreamEvent::ControlRequest {
            request:
                ControlRequestInner::CanUseTool {
                    tool_name,
                    tool_use_id,
                    input,
                },
            ..
        }) = event
        else {
            panic!("expected control request");
        };
        assert_eq!(tool_name, "AskUserQuestion");
        assert_eq!(tool_use_id, "item-1");
        assert_eq!(input["codexMethod"], "item/tool/requestUserInput");
        assert_eq!(input["questions"][0]["id"], "q1");
    }

    #[test]
    fn routes_mcp_elicitation_to_question_and_decline_response() {
        let request = JsonRpcRequest {
            id: JsonRpcId::String("mcp-1".to_string()),
            method: "mcpServer/elicitation/request".to_string(),
            params: Some(json!({
                "serverName": "linear",
                "request": {
                    "message": "Open auth URL?"
                }
            })),
        };
        let event = codex_server_request_to_control_event(&request)
            .expect("route")
            .expect("event");
        let AgentEvent::Stream(StreamEvent::ControlRequest {
            request:
                ControlRequestInner::CanUseTool {
                    tool_name, input, ..
                },
            ..
        }) = event
        else {
            panic!("expected control request");
        };
        assert_eq!(tool_name, "AskUserQuestion");
        assert_eq!(input["questions"][0]["header"], "MCP server: linear");
        assert_eq!(input["questions"][0]["question"], "Open auth URL?");

        let response = build_codex_server_request_response(&request).expect("auto response");
        assert_eq!(
            response.result,
            json!({
                "action": "decline",
                "content": null,
                "_meta": null
            })
        );
    }

    #[tokio::test]
    async fn reads_jsonrpc_messages_from_newline_stream() {
        let input = br#"
{"method":"initialized"}
{"id":7,"result":{"ok":true}}
"#;
        let mut reader = tokio::io::BufReader::new(&input[..]);

        let first = read_jsonrpc_message(&mut reader)
            .await
            .expect("read succeeds")
            .expect("first message");
        assert!(matches!(
            first,
            JsonRpcMessage::Notification(JsonRpcNotification { method, .. }) if method == "initialized"
        ));

        let second = read_jsonrpc_message(&mut reader)
            .await
            .expect("read succeeds")
            .expect("second message");
        assert!(matches!(
            second,
            JsonRpcMessage::Response(JsonRpcResponse {
                id: Some(JsonRpcId::Integer(7)),
                ..
            })
        ));

        assert!(
            read_jsonrpc_message(&mut reader)
                .await
                .expect("eof succeeds")
                .is_none()
        );
    }

    #[test]
    fn router_correlates_responses_to_tracked_requests() {
        let mut router = CodexResponseRouter::default();
        let request = build_turn_start_request(CodexTurnStartRequest {
            id: 9,
            thread_id: "thread-1",
            prompt: "hello",
            cwd: Path::new("/tmp/work"),
            model: None,
            permission_level: CodexPermissionLevel::Readonly,
            plan_mode: false,
            fast_mode: false,
            reasoning_effort: None,
            attachments: &[],
        });
        router.track_request(&request);
        assert_eq!(router.pending_len(), 1);

        let routed = router.route(JsonRpcMessage::Response(JsonRpcResponse {
            id: Some(JsonRpcId::Integer(9)),
            result: json!({"turn":{"id":"turn-1"}}),
        }));

        let CodexRoutedMessage::Response { method, response } = routed else {
            panic!("expected correlated response");
        };
        assert_eq!(method, "turn/start");
        assert_eq!(response.result["turn"]["id"], "turn-1");
        assert_eq!(router.pending_len(), 0);
    }

    #[test]
    fn extracts_thread_and_turn_ids_from_app_server_responses() {
        let thread = JsonRpcResponse {
            id: Some(JsonRpcId::Integer(1)),
            result: json!({"thread":{"id":"thread-1"}}),
        };
        let turn = JsonRpcResponse {
            id: Some(JsonRpcId::Integer(2)),
            result: json!({"turn":{"id":"turn-1"}}),
        };

        assert_eq!(thread_id_from_response(&thread).unwrap(), "thread-1");
        assert_eq!(turn_id_from_response(&turn).unwrap(), "turn-1");
    }

    #[test]
    fn missing_turn_id_is_a_structured_error() {
        let response = JsonRpcResponse {
            id: Some(JsonRpcId::Integer(2)),
            result: json!({"turn":{}}),
        };

        assert_eq!(
            turn_id_from_response(&response).unwrap_err(),
            "Codex app-server `turn/start` response did not include `turn.id`"
        );
    }

    #[test]
    fn router_preserves_server_requests_for_approval_layer() {
        let mut router = CodexResponseRouter::default();
        let routed = router.route(JsonRpcMessage::Request(JsonRpcRequest {
            id: JsonRpcId::String("approval-1".to_string()),
            method: "item/commandExecution/requestApproval".to_string(),
            params: Some(json!({"threadId":"thread-1"})),
        }));

        let CodexRoutedMessage::ServerRequest(request) = routed else {
            panic!("expected server request");
        };
        assert_eq!(request.method, "item/commandExecution/requestApproval");
        assert_eq!(request.params.unwrap()["threadId"], "thread-1");
    }

    #[test]
    fn router_decodes_notifications() {
        let mut router = CodexResponseRouter::default();
        let routed = router.route(JsonRpcMessage::Notification(JsonRpcNotification {
            method: "item/agentMessage/delta".to_string(),
            params: Some(json!({
                "threadId": "thread-1",
                "turnId": "turn-1",
                "itemId": "message-1",
                "delta": "hello"
            })),
        }));

        let CodexRoutedMessage::Notification(CodexNotificationEvent::AgentMessageDelta {
            delta,
            ..
        }) = routed
        else {
            panic!("expected notification");
        };
        assert_eq!(delta, "hello");
    }

    #[test]
    fn router_surfaces_orphan_responses() {
        let mut router = CodexResponseRouter::default();
        let routed = router.route(JsonRpcMessage::Response(JsonRpcResponse {
            id: Some(JsonRpcId::Integer(404)),
            result: json!({"late": true}),
        }));

        assert!(matches!(
            routed,
            CodexRoutedMessage::OrphanResponse(JsonRpcResponse {
                id: Some(JsonRpcId::Integer(404)),
                ..
            })
        ));
    }

    #[test]
    fn parses_null_id_error_as_orphan_error() {
        let message =
            parse_jsonrpc_line(r#"{"id":null,"error":{"code":-32700,"message":"Parse error"}}"#)
                .expect("null-id error parses");
        let mut router = CodexResponseRouter::default();
        let routed = router.route(message);

        let CodexRoutedMessage::OrphanError(JsonRpcError { id: None, error }) = routed else {
            panic!("expected null-id orphan error");
        };
        assert_eq!(error.code, -32700);
        assert_eq!(error.message, "Parse error");
    }

    #[test]
    fn initialized_notification_omits_empty_params() {
        let value = serde_json::to_value(build_initialized_notification()).unwrap();
        assert_eq!(value, json!({"method": "initialized"}));
    }

    #[test]
    fn permission_levels_map_to_codex_policy_and_sandbox() {
        assert_eq!(
            CodexPermissionLevel::from_claudette_level("readonly").mapping(),
            CodexPermissionMapping {
                approval_policy: CodexApprovalPolicy::UnlessTrusted,
                thread_sandbox: CodexSandboxMode::ReadOnly,
                turn_sandbox_policy: CodexSandboxPolicy::ReadOnly {
                    network_access: false
                },
            }
        );
        assert_eq!(
            CodexPermissionLevel::from_claudette_level("standard").mapping(),
            CodexPermissionMapping {
                approval_policy: CodexApprovalPolicy::OnRequest,
                thread_sandbox: CodexSandboxMode::WorkspaceWrite,
                turn_sandbox_policy: CodexSandboxPolicy::WorkspaceWrite {
                    network_access: false
                },
            }
        );
        assert_eq!(
            CodexPermissionLevel::from_claudette_level("full").mapping(),
            CodexPermissionMapping {
                approval_policy: CodexApprovalPolicy::Never,
                thread_sandbox: CodexSandboxMode::DangerFullAccess,
                turn_sandbox_policy: CodexSandboxPolicy::DangerFullAccess,
            }
        );
    }

    #[test]
    fn turn_start_request_carries_text_model_and_permission_overrides() {
        let request = build_turn_start_request(CodexTurnStartRequest {
            id: 7,
            thread_id: "thread-1",
            prompt: "hello",
            cwd: Path::new("/tmp/work"),
            model: Some("gpt-5.1-codex"),
            permission_level: CodexPermissionLevel::Standard,
            plan_mode: true,
            fast_mode: true,
            reasoning_effort: Some("xhigh"),
            attachments: &[],
        });
        let value = serde_json::to_value(request).unwrap();

        assert_eq!(value["method"], "turn/start");
        assert_eq!(value["params"]["threadId"], "thread-1");
        assert_eq!(value["params"]["input"][0]["type"], "text");
        assert_eq!(value["params"]["input"][0]["text"], "hello");
        assert_eq!(value["params"]["model"], "gpt-5.1-codex");
        assert_eq!(value["params"]["serviceTier"], "priority");
        assert_eq!(value["params"]["effort"], "xhigh");
        assert_eq!(value["params"]["collaborationMode"]["mode"], "plan");
        assert_eq!(
            value["params"]["collaborationMode"]["settings"]["model"],
            "gpt-5.1-codex"
        );
        assert_eq!(
            value["params"]["collaborationMode"]["settings"]["reasoning_effort"],
            "xhigh"
        );
        assert_eq!(
            value["params"]["collaborationMode"]["settings"]["developer_instructions"],
            serde_json::Value::Null
        );
        assert_eq!(value["params"]["approvalPolicy"], "on-request");
        assert_eq!(value["params"]["sandboxPolicy"]["type"], "workspaceWrite");
        assert_eq!(value["params"]["sandboxPolicy"]["networkAccess"], false);
    }

    #[test]
    fn turn_start_request_carries_image_and_text_attachments() {
        let attachments = vec![
            FileAttachment {
                media_type: "image/png".to_string(),
                data_base64: "iVBORw0KGgo=".to_string(),
                text_content: None,
                filename: Some("screenshot.png".to_string()),
            },
            FileAttachment {
                media_type: "text/plain".to_string(),
                data_base64: "aGVsbG8=".to_string(),
                text_content: Some("hello from a file".to_string()),
                filename: Some("note.txt".to_string()),
            },
        ];
        let request = build_turn_start_request(CodexTurnStartRequest {
            id: 7,
            thread_id: "thread-1",
            prompt: "inspect this",
            cwd: Path::new("/tmp/work"),
            model: None,
            permission_level: CodexPermissionLevel::Readonly,
            plan_mode: false,
            fast_mode: false,
            reasoning_effort: None,
            attachments: &attachments,
        });
        let value = serde_json::to_value(request).unwrap();

        assert_eq!(value["params"]["input"][0]["type"], "text");
        assert_eq!(value["params"]["input"][0]["text"], "inspect this");
        assert_eq!(value["params"]["input"][1]["type"], "image");
        assert_eq!(
            value["params"]["input"][1]["url"],
            "data:image/png;base64,iVBORw0KGgo="
        );
        assert_eq!(value["params"]["input"][2]["type"], "text");
        assert_eq!(
            value["params"]["input"][2]["text"],
            "Content of `note.txt`:\n```\nhello from a file\n```"
        );
    }

    #[test]
    fn turn_steer_request_reuses_attachment_input_shape() {
        let attachments = vec![FileAttachment {
            media_type: "image/jpeg".to_string(),
            data_base64: "/9j/4AAQ".to_string(),
            text_content: None,
            filename: Some("photo.jpg".to_string()),
        }];
        let request =
            build_turn_steer_request(9, "thread-1", "turn-1", "also inspect this", &attachments);
        let value = serde_json::to_value(request).unwrap();

        assert_eq!(value["method"], "turn/steer");
        assert_eq!(value["params"]["threadId"], "thread-1");
        assert_eq!(value["params"]["expectedTurnId"], "turn-1");
        assert_eq!(value["params"]["input"][0]["text"], "also inspect this");
        assert_eq!(value["params"]["input"][1]["type"], "image");
        assert_eq!(
            value["params"]["input"][1]["url"],
            "data:image/jpeg;base64,/9j/4AAQ"
        );
    }

    #[test]
    fn codex_attachment_validation_rejects_unsupported_binary_files() {
        let attachments = vec![FileAttachment {
            media_type: "application/pdf".to_string(),
            data_base64: "JVBERi0=".to_string(),
            text_content: None,
            filename: Some("brief.pdf".to_string()),
        }];

        assert_eq!(
            validate_codex_attachments(&attachments).unwrap_err(),
            "Codex app-server only supports image and text attachments; `brief.pdf` is application/pdf"
        );
    }

    #[test]
    fn turn_start_request_serializes_full_sandbox_as_tagged_policy() {
        let request = build_turn_start_request(CodexTurnStartRequest {
            id: 8,
            thread_id: "thread-1",
            prompt: "ship it",
            cwd: Path::new("/tmp/work"),
            model: Some("gpt-5.1-codex"),
            permission_level: CodexPermissionLevel::Full,
            plan_mode: false,
            fast_mode: false,
            reasoning_effort: None,
            attachments: &[],
        });
        let value = serde_json::to_value(request).unwrap();

        assert_eq!(value["params"]["approvalPolicy"], "never");
        assert_eq!(
            value["params"]["sandboxPolicy"],
            json!({"type": "dangerFullAccess"})
        );
        assert_eq!(value["params"]["serviceTier"], serde_json::Value::Null);
        assert_eq!(value["params"]["effort"], serde_json::Value::Null);
        assert_eq!(value["params"]["collaborationMode"]["mode"], "default");
    }

    #[test]
    fn thread_start_request_carries_reasoning_effort_config() {
        let request = build_thread_start_request(
            8,
            CodexThreadRequestParams {
                model: Some("gpt-5.4"),
                cwd: Path::new("/tmp/work"),
                permission_level: CodexPermissionLevel::Readonly,
                fast_mode: false,
                reasoning_effort: Some("high"),
                custom_instructions: Some("Follow repo instructions."),
                mcp_config: None,
            },
        );
        let value = serde_json::to_value(request).unwrap();

        assert_eq!(value["method"], "thread/start");
        assert_eq!(value["params"]["config"]["model_reasoning_effort"], "high");
        assert_eq!(
            value["params"]["developerInstructions"],
            "Follow repo instructions."
        );
    }

    #[test]
    fn thread_start_request_translates_claude_mcp_config_for_codex() {
        let request = build_thread_start_request(
            8,
            CodexThreadRequestParams {
                model: Some("gpt-5.4"),
                cwd: Path::new("/tmp/work"),
                permission_level: CodexPermissionLevel::Readonly,
                fast_mode: false,
                reasoning_effort: Some("high"),
                custom_instructions: None,
                mcp_config: Some(
                    r#"{"mcpServers":{"claudette":{"type":"stdio","command":"/app","args":["--agent-mcp"],"env":{"CLAUDETTE_AGENT_MCP_TOKEN":"secret"}}}}"#,
                ),
            },
        );
        let value = serde_json::to_value(request).unwrap();

        let server = &value["params"]["config"]["mcp_servers"]["claudette"];
        assert_eq!(server["command"], "/app");
        assert_eq!(server["args"][0], "--agent-mcp");
        assert_eq!(server["env"]["CLAUDETTE_AGENT_MCP_TOKEN"], "secret");
        assert!(server.get("type").is_none());
        assert_eq!(value["params"]["config"]["model_reasoning_effort"], "high");
    }

    #[test]
    fn thread_resume_request_carries_saved_thread_and_context_overrides() {
        let request = build_thread_resume_request(
            9,
            "thread-existing",
            CodexThreadRequestParams {
                model: Some("gpt-5.4"),
                cwd: Path::new("/tmp/work"),
                permission_level: CodexPermissionLevel::Standard,
                fast_mode: true,
                reasoning_effort: Some("xhigh"),
                custom_instructions: Some("Use the repo conventions."),
                mcp_config: None,
            },
        );
        let value = serde_json::to_value(request).unwrap();

        assert_eq!(value["method"], "thread/resume");
        assert_eq!(value["params"]["threadId"], "thread-existing");
        assert_eq!(
            value["params"]["developerInstructions"],
            "Use the repo conventions."
        );
        assert_eq!(value["params"]["config"]["model_reasoning_effort"], "xhigh");
        assert_eq!(value["params"]["serviceTier"], "priority");
        assert_eq!(value["params"]["approvalPolicy"], "on-request");
    }

    #[test]
    fn codex_reasoning_effort_normalization_matches_app_server_enum() {
        assert_eq!(
            normalize_codex_reasoning_effort(Some(" xhigh ")).as_deref(),
            Some("xhigh")
        );
        assert_eq!(
            normalize_codex_reasoning_effort(Some("max")).as_deref(),
            Some("high")
        );
        assert_eq!(
            normalize_codex_reasoning_effort(Some("auto")).as_deref(),
            Some("high")
        );
        assert_eq!(
            normalize_codex_reasoning_effort(Some("minimal")).as_deref(),
            Some("high")
        );
        assert_eq!(
            normalize_codex_reasoning_effort(None).as_deref(),
            Some("high")
        );
    }

    #[test]
    fn turn_interrupt_request_carries_thread_and_turn_ids() {
        let request = build_turn_interrupt_request(10, "thread-1", "turn-1");
        let value = serde_json::to_value(request).unwrap();

        assert_eq!(value["method"], "turn/interrupt");
        assert_eq!(value["params"]["threadId"], "thread-1");
        assert_eq!(value["params"]["turnId"], "turn-1");
    }

    #[test]
    fn thread_compact_start_request_carries_thread_id() {
        let request = build_thread_compact_start_request(11, "thread-9");
        let value = serde_json::to_value(request).unwrap();

        assert_eq!(value["method"], "thread/compact/start");
        assert_eq!(value["params"]["threadId"], "thread-9");
        assert!(value["params"].get("turnId").is_none());
    }

    #[test]
    fn auto_context_compaction_item_completed_emits_boundary_without_finish() {
        // Regression for issue #881: Codex delivers compaction completion as
        // an `item/completed` with `type: "contextCompaction"`. Make sure we
        // translate it into a `compact_boundary` system event. For automatic
        // mid-turn compaction, issue #909 requires that we do NOT append a
        // synthetic `Result`, because that would terminate the live turn pump.
        let item = json!({
            "id": "item-1",
            "type": "contextCompaction",
        });
        let events = map_codex_item_completed_to_agent_events(&item, false);
        assert_eq!(events.len(), 1);
        match &events[0] {
            AgentEvent::Stream(StreamEvent::System {
                subtype,
                compact_metadata,
                compact_result,
                ..
            }) => {
                assert_eq!(subtype, "compact_boundary");
                // compact_result is reserved for the end-of-compaction
                // status event in the StreamEvent::System contract — the
                // boundary event carries its payload in compact_metadata.
                assert!(compact_result.is_none());
                let meta = compact_metadata.as_ref().expect("compact_metadata set");
                // Codex's ContextCompaction item carries only an id, so we
                // synthesize zeros + a `codex` trigger; the CompactionDivider
                // recognizes that and renders without the misleading 0→0
                // token arrow.
                assert_eq!(meta.trigger, "codex");
                assert_eq!(meta.pre_tokens, 0);
                assert_eq!(meta.post_tokens, 0);
                assert_eq!(meta.duration_ms, 0);
            }
            other => panic!("expected compact_boundary System event, got {other:?}"),
        }
    }

    #[test]
    fn manual_context_compaction_item_completed_emits_boundary_and_finish() {
        // Manual `/compact` still runs as a standalone compaction turn. The
        // synthetic Result is correct there: it lets the start_compact pump
        // return the UI to Idle after the divider lands.
        let item = json!({
            "id": "item-1",
            "type": "contextCompaction",
        });
        let events = map_codex_item_completed_to_agent_events(&item, true);
        assert_eq!(events.len(), 2);
        assert!(matches!(
            &events[0],
            AgentEvent::Stream(StreamEvent::System { subtype, .. }) if subtype == "compact_boundary"
        ));
        match &events[1] {
            AgentEvent::Stream(StreamEvent::Result { subtype, .. }) => {
                assert_eq!(subtype, "success");
            }
            other => panic!("expected synthetic Result event after boundary, got {other:?}"),
        }
    }

    #[test]
    fn codex_compacting_status_event_carries_in_flight_marker() {
        // The frontend `useAgentStream` listener flips `workspace.agent_status`
        // to "Compacting" when it sees this exact shape — `subtype: "status"`
        // + `status: Some("compacting")`. Pin both so a future rename doesn't
        // silently break the UI affordance.
        match codex_compacting_status_event() {
            AgentEvent::Stream(StreamEvent::System {
                subtype, status, ..
            }) => {
                assert_eq!(subtype, "status");
                assert_eq!(status.as_deref(), Some("compacting"));
            }
            other => panic!("expected status System event, got {other:?}"),
        }
    }

    #[test]
    fn context_compaction_item_started_emits_compacting_status() {
        let item = json!({
            "id": "item-1",
            "type": "contextCompaction",
        });

        let events = map_codex_item_started_to_agent_events(&item, true);
        assert_eq!(events.len(), 1);
        match &events[0] {
            AgentEvent::Stream(StreamEvent::System {
                subtype, status, ..
            }) => {
                assert_eq!(subtype, "status");
                assert_eq!(status.as_deref(), Some("compacting"));
            }
            other => panic!("expected compacting status System event, got {other:?}"),
        }
    }

    #[test]
    fn deprecated_thread_compacted_notification_emits_auto_boundary_without_finish() {
        // Older Codex builds emit `thread/compacted` as a top-level
        // notification rather than embedding the boundary in a thread item.
        // Keep the transitional decode path until we can rely on the new
        // item shape across all supported Codex versions — and make sure
        // Auto-compaction must not synthesize a Result, even on this fallback
        // path, because Codex continues the original turn after compacting.
        let notification = JsonRpcNotification {
            method: "thread/compacted".to_string(),
            params: Some(json!({
                "threadId": "thread-x",
                "turnId": "turn-y",
            })),
        };
        let decoded = decode_notification(notification);
        assert!(matches!(
            decoded,
            CodexNotificationEvent::ContextCompacted { .. }
        ));
        let events = map_notification_to_agent_events(decoded);
        assert_eq!(events.len(), 1);
        assert!(matches!(
            &events[0],
            AgentEvent::Stream(StreamEvent::System { subtype, .. }) if subtype == "compact_boundary"
        ));
    }

    #[tokio::test]
    async fn normal_turn_flush_stays_as_assistant_block() {
        // Regression guard: a regular turn emits Assistant blocks (Codex
        // never streams text during compact — see start_compact rustdoc).
        let buffer = Arc::new(tokio::sync::Mutex::new(CodexTurnOutput::default()));
        let delta = CodexNotificationEvent::AgentMessageDelta {
            thread_id: "t".to_string(),
            turn_id: "u".to_string(),
            item_id: "item-1".to_string(),
            delta: "Hi there.".to_string(),
        };
        let _ = update_turn_output_buffer(&buffer, &delta).await;
        let completed = CodexNotificationEvent::ItemCompleted {
            thread_id: "t".to_string(),
            turn_id: "u".to_string(),
            item: json!({"id": "item-1", "type": "agentMessage"}),
        };
        let flushed = update_turn_output_buffer(&buffer, &completed).await;
        assert_eq!(flushed.len(), 1);
        assert!(matches!(
            &flushed[0],
            AgentEvent::Stream(StreamEvent::Assistant { .. })
        ));
    }

    #[test]
    fn account_read_request_can_request_token_refresh() {
        let request = build_account_read_request(12, true);
        let value = serde_json::to_value(request).unwrap();

        assert_eq!(value["method"], "account/read");
        assert_eq!(value["params"]["refreshToken"], true);
    }

    #[test]
    fn model_list_request_uses_default_picker_shape() {
        let request = build_model_list_request(13, Some("100"));
        let value = serde_json::to_value(request).unwrap();

        assert_eq!(value["method"], "model/list");
        assert_eq!(value["params"]["cursor"], "100");
        assert_eq!(value["params"]["limit"], 100);
        assert_eq!(value["params"]["includeHidden"], false);
    }

    #[test]
    fn parses_account_read_response() {
        let response = JsonRpcResponse {
            id: Some(JsonRpcId::Integer(12)),
            result: json!({
                "account": {
                    "type": "chatgpt",
                    "email": "dev@example.com",
                    "planType": "plus"
                },
                "requiresOpenaiAuth": false
            }),
        };

        assert_eq!(
            account_status_from_response(&response).expect("account parses"),
            CodexAppServerAccountStatus {
                authenticated: true,
                requires_openai_auth: false,
                account_type: Some("chatgpt".to_string()),
                email: Some("dev@example.com".to_string()),
                plan_type: Some("plus".to_string()),
            }
        );
    }

    #[test]
    fn parses_openai_required_chatgpt_account_as_authenticated() {
        let response = JsonRpcResponse {
            id: Some(JsonRpcId::Integer(12)),
            result: json!({
                "account": {
                    "type": "chatgpt",
                    "email": "dev@example.com",
                    "planType": "pro"
                },
                "requiresOpenaiAuth": true
            }),
        };

        let status = account_status_from_response(&response).expect("account parses");

        assert!(status.authenticated);
        assert!(status.requires_openai_auth);
        assert_eq!(status.account_type.as_deref(), Some("chatgpt"));
        assert_eq!(status.email.as_deref(), Some("dev@example.com"));
        assert_eq!(status.plan_type.as_deref(), Some("pro"));
    }

    #[test]
    fn parses_unauthenticated_account_read_response() {
        let response = JsonRpcResponse {
            id: Some(JsonRpcId::Integer(12)),
            result: json!({
                "account": null,
                "requiresOpenaiAuth": true
            }),
        };

        assert_eq!(
            account_status_from_response(&response).expect("account parses"),
            CodexAppServerAccountStatus {
                authenticated: false,
                requires_openai_auth: true,
                account_type: None,
                email: None,
                plan_type: None,
            }
        );
    }

    #[test]
    fn parses_model_list_response() {
        let response = JsonRpcResponse {
            id: Some(JsonRpcId::Integer(13)),
            result: json!({
                "data": [
                    {
                        "id": "model-id",
                        "model": "gpt-5.4",
                        "displayName": "GPT-5.4",
                        "hidden": false,
                        "isDefault": true
                    },
                    {
                        "id": "fallback-id",
                        "displayName": "",
                        "hidden": false,
                        "isDefault": false
                    }
                ],
                "nextCursor": "200"
            }),
        };

        let page = model_list_from_response(&response).expect("models parse");

        assert_eq!(
            page,
            CodexAppServerModelPage {
                models: vec![
                    CodexAppServerModel {
                        id: "gpt-5.4".to_string(),
                        label: "GPT-5.4".to_string(),
                        hidden: false,
                        is_default: true,
                    },
                    CodexAppServerModel {
                        id: "fallback-id".to_string(),
                        label: "fallback-id".to_string(),
                        hidden: false,
                        is_default: false,
                    },
                ],
                next_cursor: Some("200".to_string()),
            }
        );
    }

    #[test]
    fn parses_agent_message_delta_notification() {
        let message = parse_jsonrpc_line(
            r#"{"method":"item/agentMessage/delta","params":{"threadId":"t","turnId":"u","itemId":"i","delta":"hi"}}"#,
        )
        .expect("jsonrpc parses");
        let JsonRpcMessage::Notification(notification) = message else {
            panic!("expected notification");
        };

        assert_eq!(
            decode_notification(notification),
            CodexNotificationEvent::AgentMessageDelta {
                thread_id: "t".to_string(),
                turn_id: "u".to_string(),
                item_id: "i".to_string(),
                delta: "hi".to_string(),
            }
        );
    }

    #[test]
    fn parses_turn_plan_updated_notification() {
        let event = decode_notification(JsonRpcNotification {
            method: "turn/plan/updated".to_string(),
            params: Some(json!({
                "threadId": "thread-1",
                "turnId": "turn-1",
                "explanation": "Working in phases.",
                "plan": [
                    {"step": "Read the issue", "status": "completed"},
                    {"step": "Patch the bridge", "status": "inProgress"},
                    {"step": "Validate", "status": "pending"},
                    {"step": "Handle drift", "status": "cancelled"}
                ]
            })),
        });

        assert_eq!(
            event,
            CodexNotificationEvent::TurnPlanUpdated {
                thread_id: "thread-1".to_string(),
                turn_id: "turn-1".to_string(),
                explanation: Some("Working in phases.".to_string()),
                plan: vec![
                    CodexPlanStep {
                        step: "Read the issue".to_string(),
                        status: "completed".to_string(),
                    },
                    CodexPlanStep {
                        step: "Patch the bridge".to_string(),
                        status: "in_progress".to_string(),
                    },
                    CodexPlanStep {
                        step: "Validate".to_string(),
                        status: "pending".to_string(),
                    },
                    CodexPlanStep {
                        step: "Handle drift".to_string(),
                        status: "cancelled".to_string(),
                    },
                ],
            }
        );
    }

    #[test]
    fn maps_token_usage_notification_to_claudette_usage() {
        let event = decode_notification(JsonRpcNotification {
            method: "thread/tokenUsage/updated".to_string(),
            params: Some(json!({
                "threadId": "thread-1",
                "turnId": "turn-1",
                "tokenUsage": {
                    "total": {
                        "totalTokens": 200,
                        "inputTokens": 100,
                        "cachedInputTokens": 20,
                        "outputTokens": 7,
                        "reasoningOutputTokens": 3
                    },
                    "last": {
                        "totalTokens": 110,
                        "inputTokens": 100,
                        "cachedInputTokens": 20,
                        "outputTokens": 7,
                        "reasoningOutputTokens": 3
                    },
                    "modelContextWindow": 400000
                }
            })),
        });

        assert_eq!(
            event,
            CodexNotificationEvent::TokenUsageUpdated {
                thread_id: "thread-1".to_string(),
                turn_id: "turn-1".to_string(),
                usage: TokenUsage {
                    total_tokens: Some(110),
                    input_tokens: 80,
                    output_tokens: 10,
                    cache_creation_input_tokens: None,
                    cache_read_input_tokens: Some(20),
                    model_context_window: Some(400_000),
                    iterations: None,
                },
            }
        );
    }

    #[test]
    fn parses_reasoning_summary_delta_notification() {
        let event = decode_notification(JsonRpcNotification {
            method: "item/reasoning/summaryTextDelta".to_string(),
            params: Some(json!({
                "threadId": "thread-1",
                "turnId": "turn-1",
                "itemId": "reasoning-1",
                "delta": "checking",
                "summaryIndex": 2
            })),
        });

        assert_eq!(
            event,
            CodexNotificationEvent::ReasoningSummaryDelta {
                thread_id: "thread-1".to_string(),
                turn_id: "turn-1".to_string(),
                item_id: "reasoning-1".to_string(),
                delta: "checking".to_string(),
                summary_index: 2,
            }
        );
    }

    #[test]
    fn parses_item_lifecycle_notifications() {
        let item = json!({
            "type": "mcpToolCall",
            "id": "mcp-1",
            "server": "github",
            "tool": "search_issues",
            "status": "inProgress",
            "arguments": {"query": "codex"}
        });

        assert_eq!(
            decode_notification(JsonRpcNotification {
                method: "item/started".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "item": item,
                    "startedAtMs": 1
                })),
            }),
            CodexNotificationEvent::ItemStarted {
                thread_id: "thread-1".to_string(),
                turn_id: "turn-1".to_string(),
                item: json!({
                    "type": "mcpToolCall",
                    "id": "mcp-1",
                    "server": "github",
                    "tool": "search_issues",
                    "status": "inProgress",
                    "arguments": {"query": "codex"}
                }),
            }
        );
    }

    #[test]
    fn parses_turn_failed_with_default_message() {
        let event = decode_notification(JsonRpcNotification {
            method: "turn/failed".to_string(),
            params: Some(json!({
                "threadId": "thread-1",
                "turn": {"id": "turn-1", "error": {}}
            })),
        });

        assert_eq!(
            event,
            CodexNotificationEvent::TurnFailed {
                thread_id: "thread-1".to_string(),
                turn_id: Some("turn-1".to_string()),
                message: "Codex turn failed".to_string(),
            }
        );
    }

    #[test]
    fn maps_agent_message_delta_to_claudette_text_delta() {
        let events = map_notification_to_agent_events(CodexNotificationEvent::AgentMessageDelta {
            thread_id: "thread-1".to_string(),
            turn_id: "turn-1".to_string(),
            item_id: "message-1".to_string(),
            delta: "hello".to_string(),
        });

        let [
            AgentEvent::Stream(StreamEvent::Stream {
                event:
                    InnerStreamEvent::ContentBlockDelta {
                        index,
                        delta: Delta::Text { text },
                    },
            }),
        ] = events.as_slice()
        else {
            panic!("expected text delta event");
        };
        assert_eq!(*index, 0);
        assert_eq!(text, "hello");
    }

    #[test]
    fn maps_reasoning_delta_to_claudette_thinking_delta() {
        let events = map_notification_to_agent_events(CodexNotificationEvent::ReasoningTextDelta {
            thread_id: "thread-1".to_string(),
            turn_id: "turn-1".to_string(),
            item_id: "reasoning-1".to_string(),
            delta: "because".to_string(),
            content_index: 0,
        });

        let [
            AgentEvent::Stream(StreamEvent::Stream {
                event:
                    InnerStreamEvent::ContentBlockDelta {
                        index,
                        delta: Delta::Thinking { thinking },
                    },
            }),
        ] = events.as_slice()
        else {
            panic!("expected thinking delta event");
        };
        assert_eq!(*index, 1);
        assert_eq!(thinking, "because");
    }

    #[test]
    fn maps_mcp_item_started_to_tool_use_block() {
        let events = map_notification_to_agent_events(CodexNotificationEvent::ItemStarted {
            thread_id: "thread-1".to_string(),
            turn_id: "turn-1".to_string(),
            item: json!({
                "type": "mcpToolCall",
                "id": "mcp-1",
                "server": "github",
                "tool": "search_issues",
                "status": "inProgress",
                "arguments": {"query": "codex"}
            }),
        });

        assert_eq!(events.len(), 2);
        let AgentEvent::Stream(StreamEvent::Stream {
            event:
                InnerStreamEvent::ContentBlockStart {
                    content_block: Some(StartContentBlock::ToolUse { id, name, .. }),
                    ..
                },
        }) = &events[0]
        else {
            panic!("expected tool-use start");
        };
        assert_eq!(id, "mcp-1");
        assert_eq!(name, "mcp__github__search_issues");
    }

    #[test]
    fn maps_command_execution_started_to_running_task_notification() {
        let events = map_notification_to_agent_events(CodexNotificationEvent::ItemStarted {
            thread_id: "thread-1".to_string(),
            turn_id: "turn-1".to_string(),
            item: json!({
                "type": "commandExecution",
                "id": "cmd-1",
                "status": "inProgress",
                "command": "rg completed src",
                "cwd": "/repo"
            }),
        });

        assert_eq!(events.len(), 3);
        let AgentEvent::Stream(StreamEvent::System {
            subtype,
            task_id,
            tool_use_id,
            status,
            summary,
            ..
        }) = &events[2]
        else {
            panic!("expected task notification");
        };
        assert_eq!(subtype, "task_notification");
        assert_eq!(task_id.as_deref(), Some("cmd-1"));
        assert_eq!(tool_use_id.as_deref(), Some("cmd-1"));
        assert_eq!(status.as_deref(), Some("running"));
        assert_eq!(summary.as_deref(), Some("rg completed src"));
    }

    #[test]
    fn maps_command_execution_completed_to_completed_task_notification() {
        let events = map_notification_to_agent_events(CodexNotificationEvent::ItemCompleted {
            thread_id: "thread-1".to_string(),
            turn_id: "turn-1".to_string(),
            item: json!({
                "type": "commandExecution",
                "id": "cmd-1",
                "status": "completed",
                "exitCode": 0,
                "aggregatedOutput": "done\n"
            }),
        });

        assert_eq!(events.len(), 3);
        let AgentEvent::Stream(StreamEvent::System {
            subtype,
            task_id,
            status,
            summary,
            ..
        }) = &events[0]
        else {
            panic!("expected task notification");
        };
        assert_eq!(subtype, "task_notification");
        assert_eq!(task_id.as_deref(), Some("cmd-1"));
        assert_eq!(status.as_deref(), Some("completed"));
        assert_eq!(summary.as_deref(), Some("Command completed"));
    }

    #[test]
    fn maps_file_change_completed_to_tool_result() {
        let events = map_notification_to_agent_events(CodexNotificationEvent::ItemCompleted {
            thread_id: "thread-1".to_string(),
            turn_id: "turn-1".to_string(),
            item: json!({
                "type": "fileChange",
                "id": "file-1",
                "status": "completed",
                "changes": [{"path": "/tmp/a.rs", "kind": "update", "diff": "@@"}]
            }),
        });

        assert_eq!(events.len(), 3);
        assert!(
            !events.iter().any(|event| matches!(
                event,
                AgentEvent::Stream(StreamEvent::System { subtype, .. })
                    if subtype == "task_notification"
            )),
            "file changes must not drive the terminal task badge"
        );
        let AgentEvent::Stream(StreamEvent::Stream {
            event:
                InnerStreamEvent::ContentBlockDelta {
                    delta: Delta::ToolUse { partial_json },
                    ..
                },
        }) = &events[0]
        else {
            panic!("expected completed file change input delta");
        };
        assert!(
            partial_json
                .as_deref()
                .is_some_and(|input| input.contains("/tmp/a.rs"))
        );
        let AgentEvent::Stream(StreamEvent::User { message, .. }) = &events[2] else {
            panic!("expected tool result");
        };
        let super::super::UserMessageContent::Blocks(blocks) = &message.content else {
            panic!("expected block content");
        };
        let super::super::UserContentBlock::ToolResult {
            tool_use_id,
            content,
        } = &blocks[0]
        else {
            panic!("expected tool result block");
        };
        assert_eq!(tool_use_id, "file-1");
        assert_eq!(content["status"], "completed");
        assert_eq!(content["changes"][0]["path"], "/tmp/a.rs");
    }

    #[test]
    fn maps_turn_completed_to_success_result() {
        let events = map_notification_to_agent_events(CodexNotificationEvent::TurnCompleted {
            thread_id: "thread-1".to_string(),
            turn_id: "turn-1".to_string(),
            duration_ms: Some(42),
        });

        let [
            AgentEvent::Stream(StreamEvent::Result {
                subtype,
                duration_ms,
                ..
            }),
        ] = events.as_slice()
        else {
            panic!("expected result event");
        };
        assert_eq!(subtype, "success");
        assert_eq!(*duration_ms, Some(42));
    }

    #[test]
    fn codex_start_events_prime_message_and_content_blocks() {
        let events = codex_turn_start_events("thread-1");
        assert!(matches!(
            events.first(),
            Some(AgentEvent::Stream(StreamEvent::System {
                subtype,
                session_id: Some(session_id),
                ..
            })) if subtype == "init" && session_id == "thread-1"
        ));
        assert!(matches!(
            events.get(1),
            Some(AgentEvent::Stream(StreamEvent::Stream {
                event: InnerStreamEvent::MessageStart {}
            }))
        ));
        assert_eq!(events.len(), 4);
    }

    #[tokio::test]
    async fn codex_session_publishes_mapped_notifications() {
        let session = CodexAppServerSession::new_for_test(4321);
        let mut rx = session.subscribe();

        session.publish_notification_event(CodexNotificationEvent::AgentMessageDelta {
            thread_id: "thread-1".to_string(),
            turn_id: "turn-1".to_string(),
            item_id: "message-1".to_string(),
            delta: "hi".to_string(),
        });

        let event = rx.recv().await.expect("event published");
        let AgentEvent::Stream(StreamEvent::Stream {
            event:
                InnerStreamEvent::ContentBlockDelta {
                    delta: Delta::Text { text },
                    ..
                },
        }) = event
        else {
            panic!("expected text delta");
        };
        assert_eq!(text, "hi");
        assert_eq!(session.pid(), 4321);
    }

    #[tokio::test]
    async fn codex_interrupt_requires_started_thread() {
        let session = CodexAppServerSession::new_for_test(4321);

        assert_eq!(
            session.interrupt_turn().await.unwrap_err(),
            "Codex app-server has no thread to interrupt"
        );
    }

    #[test]
    fn maps_turn_plan_updated_to_update_plan_tool_activity() {
        let events = map_notification_to_agent_events(CodexNotificationEvent::TurnPlanUpdated {
            thread_id: "thread-1".to_string(),
            turn_id: "turn-1".to_string(),
            explanation: Some("Working in phases.".to_string()),
            plan: vec![
                CodexPlanStep {
                    step: "Read the issue".to_string(),
                    status: "completed".to_string(),
                },
                CodexPlanStep {
                    step: "Patch the bridge".to_string(),
                    status: "in_progress".to_string(),
                },
            ],
        });

        assert_eq!(events.len(), 4);
        let AgentEvent::Stream(StreamEvent::Stream {
            event:
                InnerStreamEvent::ContentBlockStart {
                    content_block: Some(StartContentBlock::ToolUse { id, name, .. }),
                    ..
                },
        }) = &events[0]
        else {
            panic!("expected tool-use start");
        };
        assert!(id.starts_with("update_plan-"));
        assert_eq!(name, "update_plan");

        let AgentEvent::Stream(StreamEvent::Stream {
            event:
                InnerStreamEvent::ContentBlockDelta {
                    delta: Delta::ToolUse { partial_json },
                    ..
                },
        }) = &events[1]
        else {
            panic!("expected tool-use input");
        };
        let input: Value =
            serde_json::from_str(partial_json.as_deref().expect("plan input")).unwrap();
        assert_eq!(input["explanation"], "Working in phases.");
        assert_eq!(input["plan"][0]["step"], "Read the issue");
        assert_eq!(input["plan"][0]["status"], "completed");
        assert_eq!(input["plan"][1]["status"], "in_progress");

        let AgentEvent::Stream(StreamEvent::User { message, .. }) = &events[3] else {
            panic!("expected synthetic tool result");
        };
        let crate::agent::UserMessageContent::Blocks(blocks) = &message.content else {
            panic!("expected tool-result block");
        };
        let crate::agent::UserContentBlock::ToolResult {
            tool_use_id,
            content,
        } = &blocks[0]
        else {
            panic!("expected tool-result block");
        };
        assert_eq!(tool_use_id, id);
        assert_eq!(content, &Value::String("Plan updated".to_string()));
    }

    #[test]
    fn maps_codex_plan_item_to_plan_approval_prompt() {
        let events = map_notification_to_agent_events(CodexNotificationEvent::ItemCompleted {
            thread_id: "thread-1".to_string(),
            turn_id: "turn-1".to_string(),
            item: json!({
                "type": "plan",
                "id": "plan-1",
                "text": "# Plan\n\n- Rename the README title.",
            }),
        });

        assert_eq!(events.len(), 2);
        let AgentEvent::Stream(StreamEvent::Assistant { message }) = &events[0] else {
            panic!("expected assistant plan message");
        };
        let [ContentBlock::Text { text }] = message.content.as_slice() else {
            panic!("expected text-only assistant plan message");
        };
        assert_eq!(text, "# Plan\n\n- Rename the README title.");

        let AgentEvent::Stream(StreamEvent::ControlRequest {
            request_id,
            request:
                ControlRequestInner::CanUseTool {
                    tool_name,
                    tool_use_id,
                    input,
                },
        }) = &events[1]
        else {
            panic!("expected synthetic plan approval request");
        };
        assert!(request_id.starts_with("synthetic-codex-plan-"));
        assert_eq!(tool_name, "ExitPlanMode");
        assert_eq!(tool_use_id, "codex-plan-plan-1");
        assert_eq!(input["codexSyntheticPlan"], true);
        assert_eq!(
            input["codexPlanContent"],
            "# Plan\n\n- Rename the README title."
        );
    }

    #[tokio::test]
    async fn send_turn_failure_does_not_publish_start_events() {
        let session = CodexAppServerSession::new_for_test(4321);
        let mut rx = session.subscribe();

        let err = match session.send_turn("hello", &[]).await {
            Ok(_) => panic!("test session has no stdin"),
            Err(err) => err,
        };

        assert!(err.contains("stdin is not available"));
        assert!(
            tokio::time::timeout(std::time::Duration::from_millis(25), rx.recv())
                .await
                .is_err(),
            "failed turn start should not emit a stray start event"
        );
    }

    #[tokio::test]
    async fn app_server_router_delivers_pending_response() {
        let (tx, rx) = oneshot::channel();
        let pending: PendingRequests = Arc::new(tokio::sync::Mutex::new(BTreeMap::from([(
            JsonRpcId::Integer(11),
            PendingCodexRequest {
                method: "turn/start".to_string(),
                tx,
            },
        )])));
        let (event_tx, _) = broadcast::channel(8);
        let (rate_limits_tx, _) = broadcast::channel(8);

        route_app_server_message(
            1,
            &event_tx,
            &rate_limits_tx,
            &pending,
            None,
            None,
            None,
            None,
            JsonRpcMessage::Response(JsonRpcResponse {
                id: Some(JsonRpcId::Integer(11)),
                result: json!({"turn":{"id":"turn-1"}}),
            }),
        )
        .await;

        let response = rx
            .await
            .expect("response delivered")
            .expect("response is ok");
        assert_eq!(response.result["turn"]["id"], "turn-1");
        assert!(pending.lock().await.is_empty());
    }

    #[tokio::test]
    async fn app_server_router_emits_notification_events() {
        let pending: PendingRequests = Arc::new(tokio::sync::Mutex::new(BTreeMap::new()));
        let (event_tx, mut rx) = broadcast::channel(8);
        let (rate_limits_tx, _) = broadcast::channel::<CodexRateLimitSnapshot>(8);

        route_app_server_message(
            1,
            &event_tx,
            &rate_limits_tx,
            &pending,
            None,
            None,
            None,
            None,
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/agentMessage/delta".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "itemId": "message-1",
                    "delta": "hello"
                })),
            }),
        )
        .await;

        let event = rx.recv().await.expect("event emitted");
        let AgentEvent::Stream(StreamEvent::Stream {
            event:
                InnerStreamEvent::ContentBlockDelta {
                    delta: Delta::Text { text },
                    ..
                },
        }) = event
        else {
            panic!("expected text delta");
        };
        assert_eq!(text, "hello");
    }

    #[tokio::test]
    async fn app_server_router_clears_active_turn_on_terminal_notification() {
        let pending: PendingRequests = Arc::new(tokio::sync::Mutex::new(BTreeMap::new()));
        let active_turn_id = Arc::new(tokio::sync::Mutex::new(Some("turn-1".to_string())));
        let (event_tx, mut rx) = broadcast::channel(8);
        let (rate_limits_tx, _) = broadcast::channel::<CodexRateLimitSnapshot>(8);

        route_app_server_message(
            1,
            &event_tx,
            &rate_limits_tx,
            &pending,
            None,
            Some(&active_turn_id),
            None,
            None,
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "turn/completed".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "durationMs": 42
                })),
            }),
        )
        .await;

        assert_eq!(*active_turn_id.lock().await, None);
        assert!(matches!(
            rx.recv().await.expect("event emitted"),
            AgentEvent::Stream(StreamEvent::Result { subtype, .. }) if subtype == "success"
        ));
    }

    #[tokio::test]
    async fn app_server_router_keeps_auto_compaction_inside_active_turn() {
        let pending: PendingRequests = Arc::new(tokio::sync::Mutex::new(BTreeMap::new()));
        let active_turn_id = Arc::new(tokio::sync::Mutex::new(Some("turn-1".to_string())));
        let manual_compaction_in_flight = Arc::new(AtomicBool::new(false));
        let (event_tx, mut rx) = broadcast::channel(8);
        let (rate_limits_tx, _) = broadcast::channel::<CodexRateLimitSnapshot>(8);

        route_app_server_message(
            1,
            &event_tx,
            &rate_limits_tx,
            &pending,
            None,
            Some(&active_turn_id),
            Some(&manual_compaction_in_flight),
            None,
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/started".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "item": {"id": "compact-1", "type": "contextCompaction"},
                })),
            }),
        )
        .await;
        route_app_server_message(
            1,
            &event_tx,
            &rate_limits_tx,
            &pending,
            None,
            Some(&active_turn_id),
            Some(&manual_compaction_in_flight),
            None,
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/completed".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "item": {"id": "compact-1", "type": "contextCompaction"},
                })),
            }),
        )
        .await;

        assert_eq!(
            active_turn_id.lock().await.as_deref(),
            Some("turn-1"),
            "auto-compaction must not clear the live turn"
        );
        assert!(matches!(
            rx.recv().await.expect("compacting status"),
            AgentEvent::Stream(StreamEvent::System { subtype, status, .. })
                if subtype == "status" && status.as_deref() == Some("compacting")
        ));
        assert!(matches!(
            rx.recv().await.expect("compact boundary"),
            AgentEvent::Stream(StreamEvent::System { subtype, .. }) if subtype == "compact_boundary"
        ));
        assert!(
            tokio::time::timeout(std::time::Duration::from_millis(25), rx.recv())
                .await
                .is_err(),
            "auto-compaction must not synthesize a terminal Result"
        );
    }

    #[tokio::test]
    async fn app_server_router_finishes_manual_compaction_turn_once() {
        let pending: PendingRequests = Arc::new(tokio::sync::Mutex::new(BTreeMap::new()));
        let manual_compaction_in_flight = Arc::new(AtomicBool::new(true));
        let (event_tx, mut rx) = broadcast::channel(8);
        let (rate_limits_tx, _) = broadcast::channel::<CodexRateLimitSnapshot>(8);

        route_app_server_message(
            1,
            &event_tx,
            &rate_limits_tx,
            &pending,
            None,
            None,
            Some(&manual_compaction_in_flight),
            None,
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/completed".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "compact-turn",
                    "item": {"id": "compact-1", "type": "contextCompaction"},
                })),
            }),
        )
        .await;

        assert!(!manual_compaction_in_flight.load(Ordering::SeqCst));
        assert!(matches!(
            rx.recv().await.expect("compact boundary"),
            AgentEvent::Stream(StreamEvent::System { subtype, .. }) if subtype == "compact_boundary"
        ));
        assert!(matches!(
            rx.recv().await.expect("manual finish"),
            AgentEvent::Stream(StreamEvent::Result { subtype, .. }) if subtype == "success"
        ));
    }

    #[tokio::test]
    async fn app_server_router_synthesizes_assistant_message_on_turn_completion() {
        let pending: PendingRequests = Arc::new(tokio::sync::Mutex::new(BTreeMap::new()));
        let turn_output = Arc::new(tokio::sync::Mutex::new(CodexTurnOutput::default()));
        let (event_tx, mut rx) = broadcast::channel(8);
        let (rate_limits_tx, _) = broadcast::channel::<CodexRateLimitSnapshot>(8);

        route_app_server_message(
            1,
            &event_tx,
            &rate_limits_tx,
            &pending,
            None,
            None,
            None,
            Some(&turn_output),
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/reasoning/textDelta".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "itemId": "reasoning-1",
                    "delta": "thinking"
                })),
            }),
        )
        .await;
        route_app_server_message(
            1,
            &event_tx,
            &rate_limits_tx,
            &pending,
            None,
            None,
            None,
            Some(&turn_output),
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/agentMessage/delta".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "itemId": "message-1",
                    "delta": "hello"
                })),
            }),
        )
        .await;
        route_app_server_message(
            1,
            &event_tx,
            &rate_limits_tx,
            &pending,
            None,
            None,
            None,
            Some(&turn_output),
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "turn/completed".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1"
                })),
            }),
        )
        .await;

        let _thinking_delta = rx.recv().await.expect("thinking delta");
        let _text_delta = rx.recv().await.expect("text delta");
        // Neither item ever sent `item/completed`, so the turn-end
        // safety drain fires: one Assistant event per kind in
        // thinking-then-text order, each carrying a single ContentBlock.
        let AgentEvent::Stream(StreamEvent::Assistant { message }) =
            rx.recv().await.expect("thinking assistant event")
        else {
            panic!("expected synthesized thinking assistant event");
        };
        assert!(matches!(
            &message.content[..],
            [ContentBlock::Thinking { thinking }] if thinking == "thinking"
        ));
        let AgentEvent::Stream(StreamEvent::Assistant { message }) =
            rx.recv().await.expect("text assistant event")
        else {
            panic!("expected synthesized text assistant event");
        };
        assert!(matches!(
            &message.content[..],
            [ContentBlock::Text { text }] if text == "hello"
        ));
        assert!(matches!(
            rx.recv().await.expect("result event"),
            AgentEvent::Stream(StreamEvent::Result { subtype, .. }) if subtype == "success"
        ));
    }

    /// Regression for issue #865: two `agentMessage` items in one turn
    /// must produce TWO separate `ContentBlock::Text` blocks, not a
    /// single concatenated wall of prose like "shell.Perfect…".
    #[tokio::test]
    async fn app_server_router_separates_agent_messages_per_item_id() {
        let pending: PendingRequests = Arc::new(tokio::sync::Mutex::new(BTreeMap::new()));
        let turn_output = Arc::new(tokio::sync::Mutex::new(CodexTurnOutput::default()));
        let (event_tx, mut rx) = broadcast::channel(16);
        let (rate_limits_tx, _) = broadcast::channel::<CodexRateLimitSnapshot>(8);

        let agent_message_delta = |item_id: &str, delta: &str| {
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/agentMessage/delta".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "itemId": item_id,
                    "delta": delta,
                })),
            })
        };

        for message in [
            agent_message_delta("message-1", "first sentence."),
            // Item-id change must flush the previous item before any of
            // the new item's deltas land.
            agent_message_delta("message-2", "Second sentence."),
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "turn/completed".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                })),
            }),
        ] {
            route_app_server_message(
                1,
                &event_tx,
                &rate_limits_tx,
                &pending,
                None,
                None,
                None,
                Some(&turn_output),
                message,
            )
            .await;
        }

        // first delta event (mapped from the first AgentMessageDelta)
        let _first_delta = rx.recv().await.expect("first text delta");
        // item-id change flushes message-1 as its own Assistant event
        // BEFORE the second delta event lands, so chronological order
        // is: [delta-1, assistant(message-1), delta-2, assistant(message-2)]
        let AgentEvent::Stream(StreamEvent::Assistant { message }) =
            rx.recv().await.expect("first assistant event")
        else {
            panic!("expected per-item assistant flush after item-id change");
        };
        assert!(matches!(
            &message.content[..],
            [ContentBlock::Text { text }] if text == "first sentence."
        ));
        let _second_delta = rx.recv().await.expect("second text delta");
        let AgentEvent::Stream(StreamEvent::Assistant { message }) =
            rx.recv().await.expect("second assistant event")
        else {
            panic!("expected turn-end flush of message-2");
        };
        assert!(matches!(
            &message.content[..],
            [ContentBlock::Text { text }] if text == "Second sentence."
        ));
        assert!(matches!(
            rx.recv().await.expect("result event"),
            AgentEvent::Stream(StreamEvent::Result { subtype, .. }) if subtype == "success"
        ));
    }

    /// Regression for issue #865: `item/completed` for an `agentMessage`
    /// item flushes its buffered text as its own Assistant event,
    /// without waiting for turn end. This is what lets multi-paragraph
    /// Codex turns render with item boundaries even when re-loaded from
    /// the chat_messages row.
    #[tokio::test]
    async fn app_server_router_flushes_agent_message_on_item_completed() {
        let pending: PendingRequests = Arc::new(tokio::sync::Mutex::new(BTreeMap::new()));
        let turn_output = Arc::new(tokio::sync::Mutex::new(CodexTurnOutput::default()));
        let (event_tx, mut rx) = broadcast::channel(8);
        let (rate_limits_tx, _) = broadcast::channel::<CodexRateLimitSnapshot>(8);

        for message in [
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/agentMessage/delta".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "itemId": "message-1",
                    "delta": "complete utterance",
                })),
            }),
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/completed".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "item": {
                        "type": "agentMessage",
                        "id": "message-1",
                        "text": "complete utterance",
                    },
                })),
            }),
        ] {
            route_app_server_message(
                1,
                &event_tx,
                &rate_limits_tx,
                &pending,
                None,
                None,
                None,
                Some(&turn_output),
                message,
            )
            .await;
        }

        let _delta = rx.recv().await.expect("text delta");
        let AgentEvent::Stream(StreamEvent::Assistant { message }) =
            rx.recv().await.expect("assistant flush on item/completed")
        else {
            panic!("expected per-item assistant flush on item/completed");
        };
        assert!(matches!(
            &message.content[..],
            [ContentBlock::Text { text }] if text == "complete utterance"
        ));
    }

    /// Regression for issue #865: reasoning items must be flushed
    /// progressively during the turn so the `Thinking…` UI streams in
    /// real time rather than getting one giant late-arriving block.
    #[tokio::test]
    async fn app_server_router_streams_thinking_per_reasoning_item() {
        let pending: PendingRequests = Arc::new(tokio::sync::Mutex::new(BTreeMap::new()));
        let turn_output = Arc::new(tokio::sync::Mutex::new(CodexTurnOutput::default()));
        let (event_tx, mut rx) = broadcast::channel(16);
        let (rate_limits_tx, _) = broadcast::channel::<CodexRateLimitSnapshot>(8);

        let reasoning_delta = |item_id: &str, delta: &str| {
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/reasoning/textDelta".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "itemId": item_id,
                    "delta": delta,
                    "contentIndex": 0,
                })),
            })
        };
        let reasoning_completed = |item_id: &str| {
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/completed".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "item": {
                        "type": "reasoning",
                        "id": item_id,
                    },
                })),
            })
        };

        for message in [
            reasoning_delta("reasoning-1", "thinking about A"),
            reasoning_completed("reasoning-1"),
            reasoning_delta("reasoning-2", "now thinking about B"),
            reasoning_completed("reasoning-2"),
        ] {
            route_app_server_message(
                1,
                &event_tx,
                &rate_limits_tx,
                &pending,
                None,
                None,
                None,
                Some(&turn_output),
                message,
            )
            .await;
        }

        let _first_delta = rx.recv().await.expect("first thinking delta");
        let AgentEvent::Stream(StreamEvent::Assistant { message }) =
            rx.recv().await.expect("first thinking flush")
        else {
            panic!("expected per-item thinking flush after first item/completed");
        };
        assert!(matches!(
            &message.content[..],
            [ContentBlock::Thinking { thinking }] if thinking == "thinking about A"
        ));
        let _second_delta = rx.recv().await.expect("second thinking delta");
        let AgentEvent::Stream(StreamEvent::Assistant { message }) =
            rx.recv().await.expect("second thinking flush")
        else {
            panic!("expected per-item thinking flush after second item/completed");
        };
        assert!(matches!(
            &message.content[..],
            [ContentBlock::Thinking { thinking }] if thinking == "now thinking about B"
        ));
    }

    /// Regression for issue #865: when reasoning and agentMessage items
    /// interleave (reason → answer → reason → answer), the assistant
    /// events must be emitted in chronological order so the chat
    /// surface renders them with the same ordering Codex produced.
    #[tokio::test]
    async fn app_server_router_preserves_interleaved_reasoning_and_message_order() {
        let pending: PendingRequests = Arc::new(tokio::sync::Mutex::new(BTreeMap::new()));
        let turn_output = Arc::new(tokio::sync::Mutex::new(CodexTurnOutput::default()));
        let (event_tx, mut rx) = broadcast::channel(32);
        let (rate_limits_tx, _) = broadcast::channel::<CodexRateLimitSnapshot>(8);

        let messages = [
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/reasoning/textDelta".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "itemId": "reasoning-1",
                    "delta": "reason one",
                    "contentIndex": 0,
                })),
            }),
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/completed".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "item": {"type": "reasoning", "id": "reasoning-1"},
                })),
            }),
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/agentMessage/delta".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "itemId": "message-1",
                    "delta": "answer one",
                })),
            }),
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/completed".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "item": {"type": "agentMessage", "id": "message-1"},
                })),
            }),
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/reasoning/textDelta".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "itemId": "reasoning-2",
                    "delta": "reason two",
                    "contentIndex": 0,
                })),
            }),
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/completed".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "item": {"type": "reasoning", "id": "reasoning-2"},
                })),
            }),
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/agentMessage/delta".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "itemId": "message-2",
                    "delta": "answer two",
                })),
            }),
            JsonRpcMessage::Notification(JsonRpcNotification {
                method: "item/completed".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "item": {"type": "agentMessage", "id": "message-2"},
                })),
            }),
        ];

        for message in messages {
            route_app_server_message(
                1,
                &event_tx,
                &rate_limits_tx,
                &pending,
                None,
                None,
                None,
                Some(&turn_output),
                message,
            )
            .await;
        }

        // Drain Assistant events only — ignore the per-delta stream
        // events (which test their own contract elsewhere).
        let mut assistants = Vec::new();
        while assistants.len() < 4 {
            match rx.recv().await.expect("assistant event") {
                AgentEvent::Stream(StreamEvent::Assistant { message }) => {
                    assistants.push(message);
                }
                _ => continue,
            }
        }

        let payloads: Vec<(&'static str, String)> = assistants
            .iter()
            .map(|message| match &message.content[..] {
                [ContentBlock::Thinking { thinking }] => ("thinking", thinking.clone()),
                [ContentBlock::Text { text }] => ("text", text.clone()),
                other => panic!("unexpected assistant content: {other:?}"),
            })
            .collect();

        assert_eq!(
            payloads,
            vec![
                ("thinking", "reason one".to_string()),
                ("text", "answer one".to_string()),
                ("thinking", "reason two".to_string()),
                ("text", "answer two".to_string()),
            ]
        );
    }

    #[tokio::test]
    async fn app_server_router_emits_cumulative_command_output_deltas() {
        let pending: PendingRequests = Arc::new(tokio::sync::Mutex::new(BTreeMap::new()));
        let turn_output = Arc::new(tokio::sync::Mutex::new(CodexTurnOutput::default()));
        let (event_tx, mut rx) = broadcast::channel(8);
        let (rate_limits_tx, _) = broadcast::channel::<CodexRateLimitSnapshot>(8);

        for delta in ["hello ", "world"] {
            route_app_server_message(
                1,
                &event_tx,
                &rate_limits_tx,
                &pending,
                None,
                None,
                None,
                Some(&turn_output),
                JsonRpcMessage::Notification(JsonRpcNotification {
                    method: "item/commandExecution/outputDelta".to_string(),
                    params: Some(json!({
                        "threadId": "thread-1",
                        "turnId": "turn-1",
                        "itemId": "cmd-1",
                        "delta": delta
                    })),
                }),
            )
            .await;
        }

        let first = rx.recv().await.expect("first output");
        let second = rx.recv().await.expect("second output");
        assert_eq!(tool_result_text(first), "hello ");
        assert_eq!(tool_result_text(second), "hello world");
    }

    #[tokio::test]
    async fn app_server_router_surfaces_unknown_server_request() {
        let pending: PendingRequests = Arc::new(tokio::sync::Mutex::new(BTreeMap::new()));
        let (event_tx, mut rx) = broadcast::channel(8);
        let (rate_limits_tx, _) = broadcast::channel::<CodexRateLimitSnapshot>(8);

        route_app_server_message(
            1,
            &event_tx,
            &rate_limits_tx,
            &pending,
            None,
            None,
            None,
            None,
            JsonRpcMessage::Request(JsonRpcRequest {
                id: JsonRpcId::String("req-1".to_string()),
                method: "item/tool/notImplemented".to_string(),
                params: None,
            }),
        )
        .await;

        let AgentEvent::Stderr(line) = rx.recv().await.expect("stderr event") else {
            panic!("expected stderr event");
        };
        assert!(line.contains("request `item/tool/notImplemented`"));
    }

    #[tokio::test]
    async fn app_server_router_routes_command_approval_requests_to_control_prompt() {
        let pending: PendingRequests = Arc::new(tokio::sync::Mutex::new(BTreeMap::new()));
        let (event_tx, mut rx) = broadcast::channel(8);
        let (rate_limits_tx, _) = broadcast::channel::<CodexRateLimitSnapshot>(8);

        route_app_server_message(
            1,
            &event_tx,
            &rate_limits_tx,
            &pending,
            None,
            None,
            None,
            None,
            JsonRpcMessage::Request(JsonRpcRequest {
                id: JsonRpcId::Integer(42),
                method: "item/commandExecution/requestApproval".to_string(),
                params: Some(json!({
                    "threadId": "thread-1",
                    "turnId": "turn-1",
                    "itemId": "cmd-1"
                })),
            }),
        )
        .await;

        let AgentEvent::Stream(StreamEvent::ControlRequest {
            request_id,
            request:
                ControlRequestInner::CanUseTool {
                    tool_name,
                    tool_use_id,
                    input,
                },
        }) = rx.recv().await.expect("control event")
        else {
            panic!("expected control request");
        };
        assert_eq!(request_id, "42");
        assert_eq!(tool_name, CODEX_COMMAND_APPROVAL_TOOL);
        assert_eq!(tool_use_id, "cmd-1");
        assert_eq!(
            input["codexMethod"],
            "item/commandExecution/requestApproval"
        );
        assert_eq!(input["codexApprovalKind"], "commandExecution");
    }

    #[tokio::test]
    async fn fail_pending_requests_resolves_all_waiters() {
        let (tx, rx) = oneshot::channel();
        let pending: PendingRequests = Arc::new(tokio::sync::Mutex::new(BTreeMap::from([(
            JsonRpcId::Integer(99),
            PendingCodexRequest {
                method: "initialize".to_string(),
                tx,
            },
        )])));

        fail_pending_requests(&pending, "gone").await;

        let error = rx
            .await
            .expect("waiter resolved")
            .expect_err("waiter failed");
        assert_eq!(error.id, Some(JsonRpcId::Integer(99)));
        assert_eq!(error.error.message, "gone");
        assert!(pending.lock().await.is_empty());
    }

    #[test]
    fn detects_codex_refresh_token_failure_line() {
        let line = "\u{1b}[2m2026-05-19T18:52:20Z\u{1b}[0m \u{1b}[31mERROR\u{1b}[0m \
                    \u{1b}[2mcodex_login::auth::manager\u{1b}[0m: Failed to refresh token: \
                    Your access token could not be refreshed because your refresh token \
                    was already used. Please log out and sign in again.";
        assert!(codex_stderr_indicates_auth_expiry(line));
    }

    #[test]
    fn detects_responses_websocket_401() {
        let line = "ERROR codex_api::endpoint::responses_websocket: failed to connect to \
                    websocket: HTTP error: 401 Unauthorized, url: \
                    wss://chatgpt.com/backend-api/codex/responses";
        assert!(codex_stderr_indicates_auth_expiry(line));
    }

    #[test]
    fn ignores_unrelated_401_lines() {
        // 401 against a different host (e.g. an MCP server) must not be
        // classified as Codex auth-expiry — it would tear down healthy
        // sessions over an unrelated upstream failure.
        let line = "HTTP error: 401 Unauthorized, url: https://example.com/api";
        assert!(!codex_stderr_indicates_auth_expiry(line));
    }

    #[test]
    fn ignores_routine_stderr_lines() {
        for line in [
            "INFO codex starting up",
            "tool exec started",
            "thread/started: thread-1",
        ] {
            assert!(
                !codex_stderr_indicates_auth_expiry(line),
                "false positive on routine line: {line}"
            );
        }
    }

    #[tokio::test]
    async fn send_turn_short_circuits_when_auth_expired() {
        let session = CodexAppServerSession::new_for_test(7777);
        session.auth_expired.store(true, Ordering::SeqCst);
        match session.send_turn("hello", &[]).await {
            Ok(_) => panic!("auth-expired must fail send_turn"),
            Err(err) => assert!(
                err.contains("Codex authentication expired"),
                "expected user-facing re-auth message, got: {err}"
            ),
        }
    }
}
