use std::collections::{HashMap, VecDeque};
use std::fs::OpenOptions;
use std::io::{self, BufRead, BufReader, Read, Write};
use std::path::Path;
use std::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use serde_json::{Value, json};

use crate::model::ResolvedCodexPermissions;
use crate::storage::AppResult;

const DEFAULT_CODEX_BIN: &str = "codex";
const CONVERSATION_DEVELOPER_INSTRUCTIONS: &str = "You are Mentat in normal conversation mode. Answer the user's question directly and concisely from the current cwd context. Do not invent task state, run state, or reflection artifacts unless the user explicitly asks about them. Return plain text only.";
const MENTAT_SELF_REFLECTION_DEVELOPER_INSTRUCTIONS: &str = "You are Mentat performing a conservative post-run self reflection. Default to ignore unless the evidence clearly shows the execution goal is incomplete or a reflective review is likely to materially improve the trajectory. Follow the requested output format exactly.";
const MENTAT_REFLECTION_DEVELOPER_INSTRUCTIONS: &str = "You are Mentat, the reflective agent. Review the current goal and visible trajectory carefully, stay grounded in explicit evidence, and surface process, verification, tooling, or goal-framing improvements when warranted. Do not pretend to be the worker and do not invent task-machine state.";
const MENTAT_ADOPTION_DEVELOPER_INSTRUCTIONS: &str = "You are Mentat adopting an approved improvement. Apply the smallest real change that materially implements the improvement in the current cwd and SPEC when relevant. Do not modify Mentat session state files or invent hidden runtime state. Return concise plain text only after the real work is complete.";

#[derive(Debug)]
pub struct Agent {
    child: Child,
    stdin: ChildStdin,
    stdout_messages: Receiver<io::Result<String>>,
    permissions: ResolvedCodexPermissions,
    next_request_id: u64,
    thread_id: String,
    resolved_model: Option<String>,
    stderr_buffer: Arc<Mutex<String>>,
    stdout_handle: Option<JoinHandle<()>>,
    stderr_handle: Option<JoinHandle<()>>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AgentRole {
    Conversation,
    MentatSelfReflection,
    Mentat,
    MentatAdoption,
}

#[derive(Debug)]
pub enum ConversationTurnControlMessage {
    Steer { request_id: u64, text: String },
}

#[derive(Debug)]
pub enum ConversationTurnControlEvent {
    SteerDispatched { request_id: u64 },
    SteerFailed { request_id: u64, message: String },
}

#[derive(Debug)]
pub struct ConversationTurnControlChannels {
    pub receiver: Receiver<ConversationTurnControlMessage>,
    pub event_sender: Sender<ConversationTurnControlEvent>,
}

#[derive(Debug)]
enum IncomingMessage {
    Response { id: String, result: Value },
    ErrorResponse { id: Option<String>, message: String },
    Notification { method: String, params: Value },
    Request { id: String, method: String },
}

impl Agent {
    pub fn for_conversation(
        working_dir: &Path,
        model: Option<&str>,
        permissions: ResolvedCodexPermissions,
        raw_event_stream_path: &Path,
    ) -> AppResult<Self> {
        Self::spawn(
            working_dir,
            model,
            permissions,
            raw_event_stream_path,
            AgentRole::Conversation,
        )
    }

    pub fn for_mentat_reflection(
        working_dir: &Path,
        model: Option<&str>,
        permissions: ResolvedCodexPermissions,
        raw_event_stream_path: &Path,
    ) -> AppResult<Self> {
        Self::spawn(
            working_dir,
            model,
            permissions,
            raw_event_stream_path,
            AgentRole::Mentat,
        )
    }

    pub fn for_mentat_self_reflection(
        working_dir: &Path,
        model: Option<&str>,
        permissions: ResolvedCodexPermissions,
        raw_event_stream_path: &Path,
    ) -> AppResult<Self> {
        Self::spawn(
            working_dir,
            model,
            permissions,
            raw_event_stream_path,
            AgentRole::MentatSelfReflection,
        )
    }

    pub fn for_mentat_adoption(
        working_dir: &Path,
        model: Option<&str>,
        permissions: ResolvedCodexPermissions,
        raw_event_stream_path: &Path,
    ) -> AppResult<Self> {
        Self::spawn(
            working_dir,
            model,
            permissions,
            raw_event_stream_path,
            AgentRole::MentatAdoption,
        )
    }

    fn spawn(
        working_dir: &Path,
        model: Option<&str>,
        permissions: ResolvedCodexPermissions,
        raw_event_stream_path: &Path,
        role: AgentRole,
    ) -> AppResult<Self> {
        let codex_bin =
            std::env::var("MENTAI_CODEX_BIN").unwrap_or_else(|_| DEFAULT_CODEX_BIN.to_string());

        let mut child = Command::new(&codex_bin)
            .arg("app-server")
            .current_dir(working_dir)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
            .map_err(|error| {
                io::Error::new(
                    error.kind(),
                    format!("failed to start Codex app-server with {codex_bin}: {error}"),
                )
            })?;

        let stdin = child
            .stdin
            .take()
            .ok_or_else(|| io::Error::other("failed to open Codex stdin"))?;
        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| io::Error::other("failed to open Codex stdout"))?;
        let stderr = child
            .stderr
            .take()
            .ok_or_else(|| io::Error::other("failed to open Codex stderr"))?;

        let stderr_buffer = Arc::new(Mutex::new(String::new()));
        let (stdout_messages, stdout_handle) = spawn_stdout_reader(stdout, raw_event_stream_path)?;
        let stderr_handle = Some(spawn_stderr_reader(stderr, Arc::clone(&stderr_buffer)));

        let mut agent = Self {
            child,
            stdin,
            stdout_messages,
            permissions,
            next_request_id: 0,
            thread_id: String::new(),
            resolved_model: None,
            stderr_buffer,
            stdout_handle: Some(stdout_handle),
            stderr_handle,
        };
        agent.initialize(working_dir, model, role)?;
        Ok(agent)
    }

    pub fn probe_conversation_model(working_dir: &Path) -> AppResult<Option<String>> {
        let raw_event_stream_path = std::env::temp_dir().join(format!(
            "mentat-codex-model-probe-{}-{}.ndjson",
            std::process::id(),
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap_or_default()
                .as_nanos()
        ));
        let resolved_model = {
            let agent = Self::spawn(
                working_dir,
                None,
                ResolvedCodexPermissions {
                    network_access: false,
                },
                &raw_event_stream_path,
                AgentRole::Conversation,
            )?;
            agent.resolved_model.clone()
        };
        let _ = std::fs::remove_file(&raw_event_stream_path);
        Ok(resolved_model)
    }

    pub fn ask_text(&mut self, working_dir: &Path, prompt: &str) -> AppResult<String> {
        self.ask_text_with_cancel(working_dir, prompt, None)
    }

    pub fn ask_text_with_cancel(
        &mut self,
        working_dir: &Path,
        prompt: &str,
        cancel_requested: Option<Arc<AtomicBool>>,
    ) -> AppResult<String> {
        self.ask_text_with_cancel_and_control(working_dir, prompt, cancel_requested, None)
    }

    pub fn ask_text_with_cancel_and_control(
        &mut self,
        working_dir: &Path,
        prompt: &str,
        cancel_requested: Option<Arc<AtomicBool>>,
        control: Option<ConversationTurnControlChannels>,
    ) -> AppResult<String> {
        self.run_turn(
            working_dir,
            prompt,
            vec![working_dir],
            "conversation",
            cancel_requested.as_ref(),
            control,
        )
    }

    fn run_turn(
        &mut self,
        working_dir: &Path,
        prompt: &str,
        writable_roots: Vec<&Path>,
        turn_label: &str,
        cancel_requested: Option<&Arc<AtomicBool>>,
        control: Option<ConversationTurnControlChannels>,
    ) -> AppResult<String> {
        let request_id = self.next_request_id();
        self.send_json(&build_turn_start_request(
            &request_id,
            &self.thread_id,
            working_dir,
            prompt,
            &self.permissions,
            writable_roots,
        ))?;

        let mut response_seen = false;
        let mut completed_status = None::<String>;
        let mut answer = String::new();
        let mut completed_message_text = None::<String>;
        let mut last_turn_error = None::<String>;
        let mut active_turn_id = None::<String>;
        let mut queued_controls = VecDeque::new();
        let mut steer_requests_in_flight = HashMap::new();

        while !response_seen || completed_status.is_none() {
            match self.read_message_with_timeout_handler(cancel_requested, |agent| {
                enqueue_control_messages(control.as_ref(), &mut queued_controls);
                dispatch_queued_control_messages(
                    agent,
                    control.as_ref(),
                    active_turn_id.as_deref(),
                    &mut queued_controls,
                    &mut steer_requests_in_flight,
                )
            })? {
                IncomingMessage::Response { id, result } if id == request_id => {
                    response_seen = true;
                    active_turn_id = turn_id_from_value(&result).or(active_turn_id);
                    dispatch_queued_control_messages(
                        self,
                        control.as_ref(),
                        active_turn_id.as_deref(),
                        &mut queued_controls,
                        &mut steer_requests_in_flight,
                    )?;
                }
                IncomingMessage::ErrorResponse { id, message }
                    if id.as_deref() == Some(&request_id) =>
                {
                    return Err(io::Error::other(format!(
                        "Codex turn failed for {turn_label}: {}{}",
                        message,
                        self.stderr_suffix()
                    ))
                    .into());
                }
                IncomingMessage::Response { id, .. } => {
                    if steer_requests_in_flight.remove(&id).is_some() {
                        dispatch_queued_control_messages(
                            self,
                            control.as_ref(),
                            active_turn_id.as_deref(),
                            &mut queued_controls,
                            &mut steer_requests_in_flight,
                        )?;
                    }
                }
                IncomingMessage::ErrorResponse { id, message } => {
                    if let Some(control_request_id) = id
                        .as_ref()
                        .and_then(|response_id| steer_requests_in_flight.remove(response_id))
                    {
                        send_control_event(
                            control.as_ref(),
                            ConversationTurnControlEvent::SteerFailed {
                                request_id: control_request_id,
                                message,
                            },
                        );
                        dispatch_queued_control_messages(
                            self,
                            control.as_ref(),
                            active_turn_id.as_deref(),
                            &mut queued_controls,
                            &mut steer_requests_in_flight,
                        )?;
                    }
                }
                IncomingMessage::Notification { method, params } => match method.as_str() {
                    "item/agentMessage/delta" => {
                        if let Some(delta) = params.get("delta").and_then(Value::as_str) {
                            answer.push_str(delta);
                        }
                    }
                    "item/completed" => {
                        if let Some(text) = completed_agent_message_text(&params) {
                            completed_message_text = Some(text);
                        }
                    }
                    "turn/completed" => {
                        completed_status = turn_status(&params).map(|status| status.to_string());
                        active_turn_id = None;
                        fail_queued_control_messages(
                            control.as_ref(),
                            &mut queued_controls,
                            "worker turn completed before steer could be sent",
                        );
                        if completed_status.as_deref() == Some("failed") {
                            let message = turn_error_message(&params)
                                .or(last_turn_error.clone())
                                .unwrap_or_else(|| "turn failed".to_string());
                            return Err(io::Error::other(format!(
                                "Codex turn failed for {turn_label}: {}{}",
                                message,
                                self.stderr_suffix()
                            ))
                            .into());
                        }
                    }
                    "turn/started" => {
                        active_turn_id = turn_id_from_value(&params).or(active_turn_id);
                        dispatch_queued_control_messages(
                            self,
                            control.as_ref(),
                            active_turn_id.as_deref(),
                            &mut queued_controls,
                            &mut steer_requests_in_flight,
                        )?;
                    }
                    "error" => {
                        last_turn_error = error_notification_message(&params);
                    }
                    _ => {}
                },
                IncomingMessage::Request { id, method } => {
                    return Err(io::Error::other(format!(
                        "unsupported Codex server request {method} ({id}){}",
                        self.stderr_suffix()
                    ))
                    .into());
                }
            }
        }

        if let Some(text) = completed_message_text {
            answer = text;
        }

        Ok(answer.trim().to_string())
    }

    fn initialize(
        &mut self,
        working_dir: &Path,
        model: Option<&str>,
        role: AgentRole,
    ) -> AppResult<()> {
        let _ = self.request(
            "initialize",
            json!({
                "clientInfo": {
                    "name": "mentai",
                    "version": env!("CARGO_PKG_VERSION"),
                }
            }),
        )?;
        self.send_json(&json!({ "method": "initialized" }))?;

        let mut params = serde_json::Map::new();
        params.insert("approvalPolicy".to_string(), json!("never"));
        params.insert("cwd".to_string(), json!(working_dir));
        params.insert(
            "developerInstructions".to_string(),
            json!(developer_instructions_for(role)),
        );
        params.insert("sandbox".to_string(), json!("workspace-write"));
        params.insert("personality".to_string(), json!("pragmatic"));

        if let Some(model) = model {
            params.insert("model".to_string(), json!(model));
        }

        let result = self.request("thread/start", Value::Object(params))?;
        let thread_id = result
            .get("thread")
            .and_then(|thread| thread.get("id"))
            .and_then(Value::as_str)
            .ok_or_else(|| io::Error::other("Codex thread/start response missing thread.id"))?;
        self.thread_id = thread_id.to_string();
        self.resolved_model = result
            .get("model")
            .and_then(Value::as_str)
            .map(|value| value.to_string());
        Ok(())
    }

    fn request(&mut self, method: &str, params: Value) -> AppResult<Value> {
        let request_id = self.next_request_id();
        self.send_json(&json!({
            "id": request_id,
            "method": method,
            "params": params,
        }))?;

        loop {
            match self.read_message(None)? {
                IncomingMessage::Response { id, result } if id == request_id => return Ok(result),
                IncomingMessage::ErrorResponse { id, message }
                    if id.as_deref() == Some(&request_id) =>
                {
                    return Err(io::Error::other(format!(
                        "Codex request {method} failed: {}{}",
                        message,
                        self.stderr_suffix()
                    ))
                    .into());
                }
                IncomingMessage::Request { id, method } => {
                    return Err(io::Error::other(format!(
                        "unsupported Codex server request {method} ({id}){}",
                        self.stderr_suffix()
                    ))
                    .into());
                }
                IncomingMessage::Notification { .. }
                | IncomingMessage::Response { .. }
                | IncomingMessage::ErrorResponse { .. } => {}
            }
        }
    }

    fn send_turn_steer(&mut self, turn_id: &str, text: &str) -> AppResult<String> {
        let request_id = self.next_request_id();
        self.send_json(&build_turn_steer_request(
            &request_id,
            &self.thread_id,
            turn_id,
            text,
        ))?;
        Ok(request_id)
    }

    fn send_json(&mut self, value: &Value) -> AppResult<()> {
        let line = serde_json::to_string(value)?;
        self.stdin.write_all(line.as_bytes()).map_err(|error| {
            io::Error::new(
                error.kind(),
                format!(
                    "failed to write to Codex stdin: {error}{}",
                    self.stderr_suffix()
                ),
            )
        })?;
        self.stdin.write_all(b"\n")?;
        self.stdin.flush()?;
        Ok(())
    }

    fn read_message(
        &mut self,
        cancel_requested: Option<&Arc<AtomicBool>>,
    ) -> AppResult<IncomingMessage> {
        self.read_message_with_timeout_handler(cancel_requested, |_| Ok(()))
    }

    fn read_message_with_timeout_handler<F>(
        &mut self,
        cancel_requested: Option<&Arc<AtomicBool>>,
        mut on_timeout: F,
    ) -> AppResult<IncomingMessage>
    where
        F: FnMut(&mut Self) -> AppResult<()>,
    {
        let line = loop {
            match self.stdout_messages.recv_timeout(Duration::from_millis(50)) {
                Ok(Ok(line)) => break line,
                Ok(Err(error)) => {
                    return Err(io::Error::other(format!(
                        "failed to read Codex stdout: {error}{}",
                        self.stderr_suffix()
                    ))
                    .into());
                }
                Err(RecvTimeoutError::Timeout) => {
                    on_timeout(self)?;
                    if cancel_requested.is_some_and(|flag| flag.load(Ordering::Relaxed)) {
                        let _ = self.child.kill();
                        let _ = self.child.wait();
                        return Err(io::Error::new(
                            io::ErrorKind::Interrupted,
                            "Codex turn canceled",
                        )
                        .into());
                    }
                }
                Err(RecvTimeoutError::Disconnected) => {
                    let status = self.child.try_wait()?;
                    let suffix = self.stderr_suffix();
                    return Err(io::Error::other(format!(
                        "Codex app-server closed stdout unexpectedly (status: {:?}){}",
                        status.and_then(|value| value.code()),
                        suffix
                    ))
                    .into());
                }
            }
        };

        let message: Value = serde_json::from_str(line.trim()).map_err(|error| {
            io::Error::other(format!(
                "failed to parse Codex message: {error}{}",
                self.stderr_suffix()
            ))
        })?;

        if message.get("result").is_some() {
            let id = message_id(&message)
                .ok_or_else(|| io::Error::other("Codex response missing id"))?;
            let result = message.get("result").cloned().unwrap_or(Value::Null);
            return Ok(IncomingMessage::Response { id, result });
        }

        if message.get("error").is_some() {
            let id = message_id(&message);
            let error_value = message.get("error").cloned().unwrap_or(Value::Null);
            return Ok(IncomingMessage::ErrorResponse {
                id,
                message: error_message(&error_value),
            });
        }

        let method = message
            .get("method")
            .and_then(Value::as_str)
            .ok_or_else(|| io::Error::other("Codex message missing method"))?
            .to_string();

        if message.get("id").is_some() {
            let id = message_id(&message)
                .ok_or_else(|| io::Error::other("Codex server request missing id"))?;
            return Ok(IncomingMessage::Request { id, method });
        }

        Ok(IncomingMessage::Notification {
            method,
            params: message.get("params").cloned().unwrap_or(Value::Null),
        })
    }

    fn next_request_id(&mut self) -> String {
        self.next_request_id += 1;
        self.next_request_id.to_string()
    }

    fn stderr_suffix(&self) -> String {
        let stderr = self
            .stderr_buffer
            .lock()
            .ok()
            .map(|buffer| buffer.trim().to_string())
            .unwrap_or_default();
        if stderr.is_empty() {
            String::new()
        } else {
            format!("\nstderr:\n{stderr}")
        }
    }
}

impl Drop for Agent {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
        if let Some(handle) = self.stdout_handle.take() {
            let _ = handle.join();
        }
        if let Some(handle) = self.stderr_handle.take() {
            let _ = handle.join();
        }
    }
}

fn build_turn_start_request(
    request_id: &str,
    thread_id: &str,
    working_dir: &Path,
    prompt: &str,
    permissions: &ResolvedCodexPermissions,
    writable_roots: Vec<&Path>,
) -> Value {
    json!({
        "id": request_id,
        "method": "turn/start",
        "params": {
            "threadId": thread_id,
            "cwd": working_dir,
            "approvalPolicy": "never",
            "sandboxPolicy": {
                "type": "workspaceWrite",
                "writableRoots": writable_roots,
                "readOnlyAccess": {
                    "type": "fullAccess"
                },
                "networkAccess": permissions.network_access,
                "excludeTmpdirEnvVar": false,
                "excludeSlashTmp": false
            },
            "input": [
                {
                    "type": "text",
                    "text": prompt,
                }
            ],
        }
    })
}

fn build_turn_steer_request(request_id: &str, thread_id: &str, turn_id: &str, text: &str) -> Value {
    json!({
        "id": request_id,
        "method": "turn/steer",
        "params": {
            "threadId": thread_id,
            "expectedTurnId": turn_id,
            "input": [
                {
                    "type": "text",
                    "text": text,
                }
            ],
        }
    })
}

fn developer_instructions_for(role: AgentRole) -> &'static str {
    match role {
        AgentRole::Conversation => CONVERSATION_DEVELOPER_INSTRUCTIONS,
        AgentRole::MentatSelfReflection => MENTAT_SELF_REFLECTION_DEVELOPER_INSTRUCTIONS,
        AgentRole::Mentat => MENTAT_REFLECTION_DEVELOPER_INSTRUCTIONS,
        AgentRole::MentatAdoption => MENTAT_ADOPTION_DEVELOPER_INSTRUCTIONS,
    }
}

fn spawn_stdout_reader(
    stdout: ChildStdout,
    raw_event_stream_path: &Path,
) -> AppResult<(Receiver<io::Result<String>>, JoinHandle<()>)> {
    if let Some(parent) = raw_event_stream_path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let mut raw_event_writer = OpenOptions::new()
        .create(true)
        .append(true)
        .open(raw_event_stream_path)?;
    let (sender, receiver) = mpsc::channel();
    let handle = thread::spawn(move || {
        let mut reader = BufReader::new(stdout);
        loop {
            let mut line = String::new();
            match reader.read_line(&mut line) {
                Ok(0) => break,
                Ok(_) => {
                    if let Err(error) = raw_event_writer
                        .write_all(line.as_bytes())
                        .and_then(|_| raw_event_writer.flush())
                    {
                        let _ = sender.send(Err(error));
                        break;
                    }
                    if sender.send(Ok(line)).is_err() {
                        break;
                    }
                }
                Err(error) => {
                    let _ = sender.send(Err(error));
                    break;
                }
            }
        }
    });
    Ok((receiver, handle))
}

fn spawn_stderr_reader(stderr: ChildStderr, buffer: Arc<Mutex<String>>) -> JoinHandle<()> {
    thread::spawn(move || {
        let mut reader = BufReader::new(stderr);
        let mut content = String::new();
        let _ = reader.read_to_string(&mut content);
        if !content.is_empty()
            && let Ok(mut current) = buffer.lock()
        {
            current.push_str(&content);
        }
    })
}

fn message_id(message: &Value) -> Option<String> {
    let id = message.get("id")?;
    if let Some(value) = id.as_str() {
        return Some(value.to_string());
    }
    if let Some(value) = id.as_i64() {
        return Some(value.to_string());
    }
    if let Some(value) = id.as_u64() {
        return Some(value.to_string());
    }
    None
}

fn error_message(value: &Value) -> String {
    value
        .get("message")
        .and_then(Value::as_str)
        .map(|message| message.to_string())
        .unwrap_or_else(|| value.to_string())
}

fn completed_agent_message_text(params: &Value) -> Option<String> {
    let item = params.get("item")?;
    if item.get("type").and_then(Value::as_str) != Some("agentMessage") {
        return None;
    }
    item.get("text")
        .and_then(Value::as_str)
        .map(|text| text.to_string())
}

fn error_notification_message(params: &Value) -> Option<String> {
    params
        .get("error")
        .and_then(|error| error.get("message"))
        .and_then(Value::as_str)
        .map(|message| message.to_string())
}

fn turn_error_message(params: &Value) -> Option<String> {
    params
        .get("turn")
        .and_then(|turn| turn.get("error"))
        .and_then(|error| error.get("message"))
        .and_then(Value::as_str)
        .map(|message| message.to_string())
}

fn turn_status(params: &Value) -> Option<&str> {
    params
        .get("turn")
        .and_then(|turn| turn.get("status"))
        .and_then(Value::as_str)
}

fn turn_id_from_value(value: &Value) -> Option<String> {
    value
        .get("turn")
        .and_then(|turn| turn.get("id"))
        .and_then(Value::as_str)
        .map(|turn_id| turn_id.to_string())
}

fn enqueue_control_messages(
    control: Option<&ConversationTurnControlChannels>,
    queued_controls: &mut VecDeque<ConversationTurnControlMessage>,
) {
    let Some(control) = control else {
        return;
    };

    while let Ok(message) = control.receiver.try_recv() {
        queued_controls.push_back(message);
    }
}

fn dispatch_queued_control_messages(
    agent: &mut Agent,
    control: Option<&ConversationTurnControlChannels>,
    active_turn_id: Option<&str>,
    queued_controls: &mut VecDeque<ConversationTurnControlMessage>,
    steer_requests_in_flight: &mut HashMap<String, u64>,
) -> AppResult<()> {
    let Some(active_turn_id) = active_turn_id else {
        return Ok(());
    };

    if !steer_requests_in_flight.is_empty() {
        return Ok(());
    }

    let Some(message) = queued_controls.pop_front() else {
        return Ok(());
    };

    match message {
        ConversationTurnControlMessage::Steer { request_id, text } => {
            match agent.send_turn_steer(active_turn_id, &text) {
                Ok(protocol_request_id) => {
                    send_control_event(
                        control,
                        ConversationTurnControlEvent::SteerDispatched { request_id },
                    );
                    steer_requests_in_flight.insert(protocol_request_id, request_id);
                }
                Err(error) => {
                    send_control_event(
                        control,
                        ConversationTurnControlEvent::SteerFailed {
                            request_id,
                            message: error.to_string(),
                        },
                    );
                }
            }
        }
    }

    Ok(())
}

fn fail_queued_control_messages(
    control: Option<&ConversationTurnControlChannels>,
    queued_controls: &mut VecDeque<ConversationTurnControlMessage>,
    message: &str,
) {
    while let Some(control_message) = queued_controls.pop_front() {
        match control_message {
            ConversationTurnControlMessage::Steer { request_id, .. } => {
                send_control_event(
                    control,
                    ConversationTurnControlEvent::SteerFailed {
                        request_id,
                        message: message.to_string(),
                    },
                );
            }
        }
    }
}

fn send_control_event(
    control: Option<&ConversationTurnControlChannels>,
    event: ConversationTurnControlEvent,
) {
    if let Some(control) = control {
        let _ = control.event_sender.send(event);
    }
}

#[cfg(test)]
mod tests {
    use std::path::Path;

    use super::{build_turn_start_request, build_turn_steer_request};
    use crate::model::ResolvedCodexPermissions;
    use serde_json::json;

    #[test]
    fn turn_start_request_uses_resolved_network_access_setting() {
        let request = build_turn_start_request(
            "1",
            "thread-1",
            Path::new("/tmp/workspace"),
            "do the work",
            &ResolvedCodexPermissions {
                network_access: false,
            },
            vec![Path::new("/tmp/workspace")],
        );

        assert_eq!(request["method"], "turn/start");
        assert_eq!(request["params"]["approvalPolicy"], "never");
        assert_eq!(
            request["params"]["sandboxPolicy"],
            json!({
                "type": "workspaceWrite",
                "writableRoots": ["/tmp/workspace"],
                "readOnlyAccess": {
                    "type": "fullAccess"
                },
                "networkAccess": false,
                "excludeTmpdirEnvVar": false,
                "excludeSlashTmp": false
            })
        );
    }

    #[test]
    fn turn_steer_request_targets_active_turn_with_text_input() {
        let request = build_turn_steer_request("2", "thread-1", "turn-9", "Focus on tests first");

        assert_eq!(
            request,
            json!({
                "id": "2",
                "method": "turn/steer",
                "params": {
                    "threadId": "thread-1",
                    "expectedTurnId": "turn-9",
                    "input": [
                        {
                            "type": "text",
                            "text": "Focus on tests first",
                        }
                    ],
                }
            })
        );
    }
}
