use crate::agent::adapter::{self, ActorSessionMap};
use crate::agent::claude_stream;
use crate::agent::codex_appserver::CodexAppServer;
use crate::agent::session_actor::{self, ActorCommand, AttachmentData, RalphCancelResult};
use crate::agent::session_protocol::{CodexSkillRef, SessionProtocol, StartupCtx};
use crate::agent::spawn_locks::SpawnLocks;
use crate::models::ConversationRef;
use crate::models::{BusEvent, RemoteHost, RunMeta, RunStatus, SessionMode, UserSettings};
use crate::process_ext::HideConsole;
use crate::storage;
use crate::web_server::broadcaster::BroadcastEmitter;
use std::sync::Arc;
use tauri::State;
use tokio_util::sync::CancellationToken;

/// Truncate a string to at most `max` bytes, snapping to a char boundary.
fn truncate_str(s: &str, max: usize) -> &str {
    if s.len() <= max {
        return s;
    }
    let mut end = max;
    while end > 0 && !s.is_char_boundary(end) {
        end -= 1;
    }
    &s[..end]
}

/// Helper: get the actor command sender for a run_id.
async fn get_cmd_tx(
    sessions: &ActorSessionMap,
    run_id: &str,
) -> Result<tokio::sync::mpsc::Sender<ActorCommand>, String> {
    let map = sessions.lock().await;
    map.get(run_id)
        .map(|h| h.cmd_tx.clone())
        .ok_or_else(|| format!("Session {} not found", run_id))
}

/// Helper: stop an existing actor for a run_id, await its shutdown.
/// Returns true if an actor was stopped.
pub(super) async fn stop_actor(sessions: &ActorSessionMap, run_id: &str) -> Result<bool, String> {
    let handle = {
        let mut map = sessions.lock().await;
        map.remove(run_id)
    };

    let Some(handle) = handle else {
        return Ok(false);
    };

    log::debug!("[session] stopping actor for run_id={}", run_id);

    // Send Stop command
    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
    if handle
        .cmd_tx
        .send(ActorCommand::Stop { reply: reply_tx })
        .await
        .is_ok()
    {
        // Wait for reply (actor acknowledged stop)
        let _ = reply_rx.await;
    }

    // Wait for actor task to finish (with timeout)
    let _ = tokio::time::timeout(std::time::Duration::from_secs(5), handle.join_handle).await;

    Ok(true)
}

/// Resolve a RemoteHost from RunMeta.
/// Prefers the snapshot (self-contained), falls back to name lookup for old runs.
fn resolve_remote_host(meta: &RunMeta) -> Result<Option<RemoteHost>, String> {
    // Prefer snapshot (new runs have this)
    if let Some(ref snapshot) = meta.remote_host_snapshot {
        log::debug!(
            "[session] resolve_remote_host: using snapshot for '{}'",
            snapshot.name
        );
        return Ok(Some(snapshot.clone()));
    }
    // Fallback: name-based lookup (old runs without snapshot)
    match &meta.remote_host_name {
        Some(name) => {
            let settings = storage::settings::get_user_settings();
            settings
                .remote_hosts
                .iter()
                .find(|h| h.name == *name)
                .cloned()
                .map(Some)
                .ok_or_else(|| format!("Remote host '{}' not found in settings", name))
        }
        None => Ok(None),
    }
}

/// Resolved authentication and environment info for spawning CLI.
struct ResolvedAuth {
    api_key: Option<String>,
    auth_token: Option<String>,
    base_url: Option<String>,
    /// Full models array from credential/preset (tier mapping applied at injection time).
    models: Option<Vec<String>>,
    extra_env: Option<std::collections::HashMap<String, String>>,
}

/// Resolve models array into (env_key, env_value) pairs for CLI injection.
/// 1 model  → all tiers same
/// 2 models → [0]=Opus+Sonnet, [1]=Haiku
/// 3+ models → [0]=Opus, [1]=Sonnet, [2]=Haiku
pub(crate) fn resolve_model_tiers(models: &[String]) -> Vec<(&'static str, String)> {
    if models.is_empty() {
        return vec![];
    }
    let (opus, sonnet, haiku) = match models.len() {
        1 => (&models[0], &models[0], &models[0]),
        2 => (&models[0], &models[0], &models[1]),
        _ => {
            // 3+ elements: Sonnet (index 1) is the anchor.
            // If Sonnet is empty → no injection (user left all meaningful fields blank).
            let sonnet = &models[1];
            if sonnet.is_empty() {
                return vec![];
            }
            let opus = if models[0].is_empty() {
                sonnet
            } else {
                &models[0]
            };
            let haiku = if models[2].is_empty() {
                sonnet
            } else {
                &models[2]
            };
            (opus, sonnet, haiku)
        }
    };
    log::debug!(
        "[session] resolve_model_tiers: opus={}, sonnet={}, haiku={}",
        opus,
        sonnet,
        haiku
    );
    vec![
        ("ANTHROPIC_MODEL", sonnet.clone()),
        ("ANTHROPIC_DEFAULT_OPUS_MODEL", opus.clone()),
        ("ANTHROPIC_DEFAULT_SONNET_MODEL", sonnet.clone()),
        ("ANTHROPIC_DEFAULT_HAIKU_MODEL", haiku.clone()),
    ]
}

/// Resolve API authentication environment variables.
/// Returns ResolvedAuth with (api_key, auth_token, base_url, models, extra_env).
/// - `api_key`: for Anthropic official (`x-api-key` header)
/// - `auth_token`: for third-party platforms (`Authorization: Bearer` header)
/// - `base_url`: custom API endpoint
///
/// `api_key` and `auth_token` are mutually exclusive.
fn resolve_auth_env(remote: &Option<RemoteHost>, settings: &UserSettings) -> ResolvedAuth {
    let base_url = settings
        .anthropic_base_url
        .as_ref()
        .filter(|s| !s.is_empty())
        .cloned();

    // SSH remote: forward_api_key=false → no credentials forwarded
    if let Some(r) = remote.as_ref() {
        if !r.forward_api_key {
            return ResolvedAuth {
                api_key: None,
                auth_token: None,
                base_url: None,
                models: None,
                extra_env: None,
            };
        }
        // forward_api_key=true: fall through to normal resolution
    }

    // Local API Key mode (also used for remote with forward_api_key=true)
    if settings.auth_mode == "api" {
        if let Some(ref key) = settings.anthropic_api_key {
            if !key.is_empty() {
                // Use auth_env_var from platform preset to decide which header to use.
                // "ANTHROPIC_AUTH_TOKEN" → Bearer header (most third-party platforms)
                // "ANTHROPIC_API_KEY" (or unset) → x-api-key header (Anthropic, AiHubMix, Kimi Coding)
                let use_bearer = settings.auth_env_var.as_deref() == Some("ANTHROPIC_AUTH_TOKEN");

                if use_bearer {
                    return ResolvedAuth {
                        api_key: None,
                        auth_token: Some(key.clone()),
                        base_url,
                        models: None,
                        extra_env: None,
                    };
                } else {
                    return ResolvedAuth {
                        api_key: Some(key.clone()),
                        auth_token: None,
                        base_url,
                        models: None,
                        extra_env: None,
                    };
                }
            }
        }
    }

    // CLI mode: never inject a stale base_url — CLI manages its own connection
    ResolvedAuth {
        api_key: None,
        auth_token: None,
        base_url: if settings.auth_mode == "cli" {
            None
        } else {
            base_url
        },
        models: None,
        extra_env: None,
    }
}

/// Build ResolvedAuth with PROXY_MANAGED placeholder token for keyless local proxies.
fn make_placeholder_auth(
    use_bearer: bool,
    base_url: Option<String>,
    models: Option<Vec<String>>,
    extra_env: Option<std::collections::HashMap<String, String>>,
) -> ResolvedAuth {
    if use_bearer {
        ResolvedAuth {
            api_key: None,
            auth_token: Some("PROXY_MANAGED".to_string()),
            base_url,
            models,
            extra_env,
        }
    } else {
        ResolvedAuth {
            api_key: Some("PROXY_MANAGED".to_string()),
            auth_token: None,
            base_url,
            models,
            extra_env,
        }
    }
}

/// Check whether a URL points to a local address (localhost, 127.x.x.x, ::1, 0.0.0.0).
fn is_local_url(url: &str) -> bool {
    let Ok(parsed) = url::Url::parse(url) else {
        return false;
    };
    let Some(host) = parsed.host_str() else {
        return false;
    };
    if host == "localhost" {
        return true;
    }
    // Parse as IP and check loopback/unspecified
    if let Ok(ip) = host.parse::<std::net::IpAddr>() {
        return ip.is_loopback() || ip.is_unspecified();
    }
    // Handle bracketed IPv6 like [::1]
    let trimmed = host.trim_start_matches('[').trim_end_matches(']');
    if let Ok(ip) = trimmed.parse::<std::net::IpAddr>() {
        return ip.is_loopback() || ip.is_unspecified();
    }
    false
}

/// Preflight reachability check for a provider's base_url.
/// Sends HEAD to `{base_url}/v1/models` — any HTTP response (even 401/403/405)
/// means the service is online. Only connection failure/timeout returns Err.
async fn preflight_check_base_url(
    base_url: Option<&str>,
    platform_id: Option<&str>,
) -> Result<(), String> {
    let Some(url) = base_url else {
        log::debug!("[session] preflight: no base_url, skipping");
        return Ok(());
    };

    let is_local = is_local_url(url);
    let timeout = if is_local {
        std::time::Duration::from_secs(1)
    } else {
        std::time::Duration::from_secs(3)
    };

    let check_url = format!("{}/v1/models", url.trim_end_matches('/'));
    log::debug!(
        "[session] preflight: checking {} (local={}, timeout={:?})",
        check_url,
        is_local,
        timeout
    );

    let mut builder = reqwest::Client::builder().timeout(timeout);
    if is_local {
        builder = builder.no_proxy();
    }
    let client = builder
        .build()
        .map_err(|e| format!("HTTP client error: {}", e))?;

    match client.head(&check_url).send().await {
        Ok(resp) => {
            log::debug!(
                "[session] preflight: {} responded with status {}",
                url,
                resp.status()
            );
            Ok(())
        }
        Err(e) => {
            let display_name = platform_id
                .map(super::onboarding::preset_name)
                .unwrap_or_else(|| "Provider".to_string());

            let suggestion = if is_local {
                format!(
                    "Make sure {} is running and listening on {}",
                    display_name, url
                )
            } else {
                format!(
                    "Check your network connection and verify {} is accessible",
                    url
                )
            };

            log::warn!("[session] preflight: {} unreachable: {}", url, e);
            Err(format!(
                "{} is unreachable ({}). {}",
                display_name, url, suggestion
            ))
        }
    }
}

/// Resolve auth env using per-session platform_id.
/// Looks up the credential from `settings.platform_credentials` by platform_id,
/// then returns ResolvedAuth matching the credential's auth_env_var.
/// Falls back to global `resolve_auth_env()` if platform_id is None or credential not found.
///
/// For keyless local proxies (ccswitch, ccr, ollama): uses PROXY_MANAGED placeholder token
/// with known defaults for base_url and auth_env_var.
///
/// For SSH remote sessions:
/// - `forward_api_key=true`: resolve credentials normally (platform-aware) and forward them
/// - `forward_api_key=false`: return empty ResolvedAuth — remote uses its own auth
fn resolve_auth_env_for_platform(
    remote: &Option<RemoteHost>,
    settings: &UserSettings,
    platform_id: Option<&str>,
) -> ResolvedAuth {
    // SSH remote with forward_api_key=false: don't forward any credentials
    if let Some(r) = remote.as_ref() {
        if !r.forward_api_key {
            log::debug!("[session] resolve_auth_env_for_platform: remote forward_api_key=false, no credentials forwarded");
            return ResolvedAuth {
                api_key: None,
                auth_token: None,
                base_url: None,
                models: None,
                extra_env: None,
            };
        }
        // forward_api_key=true: fall through to normal platform-aware resolution
    }

    // If we have a platform_id, try to find a matching credential
    if let Some(pid) = platform_id {
        if let Some(cred) = settings
            .platform_credentials
            .iter()
            .find(|c| c.platform_id == pid)
        {
            let key = cred.api_key.as_ref().filter(|k| !k.is_empty()).cloned();
            let base_url = cred.base_url.as_ref().filter(|s| !s.is_empty()).cloned();
            let use_bearer = cred.auth_env_var.as_deref() == Some("ANTHROPIC_AUTH_TOKEN");
            let models = cred.models.clone().filter(|m| !m.is_empty());
            let extra_env = cred.extra_env.clone();

            if let Some(k) = key {
                log::debug!(
                    "[session] resolve_auth_env_for_platform: platform={}, use_bearer={}, has_base_url={}, models={:?}, extra_env_count={}",
                    pid,
                    use_bearer,
                    base_url.is_some(),
                    models,
                    extra_env.as_ref().map_or(0, |e| e.len())
                );
                return if use_bearer {
                    ResolvedAuth {
                        api_key: None,
                        auth_token: Some(k),
                        base_url,
                        models,
                        extra_env,
                    }
                } else {
                    ResolvedAuth {
                        api_key: Some(k),
                        auth_token: None,
                        base_url,
                        models,
                        extra_env,
                    }
                };
            }
            // Credential found but no API key — check if key_optional platform
            if storage::settings::is_key_optional_platform(pid) {
                let info = storage::settings::get_provider_info(pid);

                // auth_env_var: known defaults take priority over credential (prevents dirty data)
                let effective_auth = info
                    .as_ref()
                    .and_then(|i| i.auth_env_var.clone())
                    .or_else(|| cred.auth_env_var.clone());
                let effective_bearer = effective_auth.as_deref() == Some("ANTHROPIC_AUTH_TOKEN");

                // base_url fallback: credential → known defaults
                let effective_url =
                    base_url.or_else(|| info.as_ref().and_then(|i| i.base_url.clone()));

                // models / extra_env fallback: credential → defaults
                let effective_models = models.or_else(|| {
                    info.as_ref()
                        .and_then(|i| i.models.clone())
                        .filter(|m| !m.is_empty())
                });
                let effective_extra =
                    extra_env.or_else(|| info.as_ref().and_then(|i| i.extra_env.clone()));

                log::info!(
                    "[session] platform '{}': key_optional, credential config with placeholder (base_url={:?})",
                    pid,
                    effective_url
                );
                return make_placeholder_auth(
                    effective_bearer,
                    effective_url,
                    effective_models,
                    effective_extra,
                );
            }
            log::warn!(
                "[session] resolve_auth_env_for_platform: credential for platform '{}' has no api_key, falling back to global",
                pid
            );
        } else {
            // No credential entry — check if key_optional platform with known defaults
            if let Some(info) = storage::settings::get_provider_info(pid) {
                if info.key_optional {
                    let use_bearer = info.auth_env_var.as_deref() == Some("ANTHROPIC_AUTH_TOKEN");
                    let models = info.models.clone().filter(|m| !m.is_empty());
                    log::info!(
                        "[session] platform '{}': no credential, using known defaults (key_optional, base_url={:?})",
                        pid,
                        info.base_url
                    );
                    return make_placeholder_auth(
                        use_bearer,
                        info.base_url,
                        models,
                        info.extra_env,
                    );
                }
            }
            log::warn!(
                "[session] resolve_auth_env_for_platform: no credential found for platform '{}', falling back to global",
                pid
            );
        }
    }

    // Fallback to global auth env
    resolve_auth_env(remote, settings)
}

#[allow(clippy::too_many_arguments)]
pub(crate) async fn start_session_impl(
    emitter: &Arc<BroadcastEmitter>,
    sessions: &ActorSessionMap,
    spawn_locks: &SpawnLocks,
    cancel_token: &CancellationToken,
    run_id: String,
    mode: Option<SessionMode>,
    session_id: Option<String>,
    initial_message: Option<String>,
    attachments: Option<Vec<AttachmentData>>,
    platform_id: Option<String>,
    permission_mode_override: Option<String>,
) -> Result<(), String> {
    let _guard = spawn_locks.acquire(&run_id).await;
    let session_mode = mode.unwrap_or_default();
    let att_list = attachments.unwrap_or_default();
    log::debug!(
        "[session] start_session called, run_id={}, mode={:?}, session_id={:?}, has_message={}, attachments={}",
        run_id,
        session_mode,
        session_id,
        initial_message.is_some(),
        att_list.len()
    );

    // 1. Read run metadata + validate execution path
    let meta =
        storage::runs::get_run(&run_id).ok_or_else(|| format!("Run {} not found", run_id))?;
    let exec_path = meta.resolved_execution_path();
    if exec_path != crate::models::ExecutionPath::SessionActor {
        return Err(format!(
            "start_session requires execution_path=session_actor, got {:?} for run {}",
            exec_path, run_id
        ));
    }
    log::debug!(
        "[session] meta loaded: agent={}, prompt={:?}, cwd={}, exec_path={:?}",
        meta.agent,
        truncate_str(&meta.prompt, 80),
        meta.cwd,
        exec_path
    );

    // 2. Read settings and build unified adapter settings
    let agent_settings = storage::settings::get_agent_settings(&meta.agent);
    let user_settings = storage::settings::get_user_settings();
    let mut adapter_settings =
        adapter::build_adapter_settings(&agent_settings, &user_settings, meta.model.clone());

    // 2a. Apply per-session permission_mode override (e.g. ExitPlanMode → acceptEdits).
    //     Session-scoped: does not touch persisted user settings. Must run BEFORE spawn
    //     so the CLI's --permission-mode arg reflects the override for the first turn.
    if let Some(ref override_mode) = permission_mode_override {
        log::debug!(
            "[session] permission_mode override: {:?} → {:?}",
            adapter_settings.permission_mode,
            override_mode
        );
        adapter_settings.permission_mode = Some(override_mode.clone());
    }

    // 2b. Resolve remote host from RunMeta (audit #2: single truth source)
    let remote = resolve_remote_host(&meta)?;
    // Use per-session platform_id: prefer IPC param, fallback to RunMeta's saved platform_id
    // CLI Auth mode: ignore platform_id — CLI manages its own connection
    let effective_pid = if user_settings.auth_mode == "cli" {
        None
    } else {
        platform_id.as_deref().or(meta.platform_id.as_deref())
    };
    let resolved = resolve_auth_env_for_platform(&remote, &user_settings, effective_pid);
    adapter::clear_model_if_provider_overrides(
        &mut adapter_settings,
        &meta.model,
        &agent_settings.model,
        &resolved.models,
    );
    let resolved = augment_with_shell_auth(
        resolved,
        &user_settings.auth_mode,
        remote.is_some(),
        &meta.cwd,
    );
    if remote.is_some() {
        log::debug!(
            "[session] remote mode: host={:?}, remote_cwd={:?}, has_key={}",
            meta.remote_host_name,
            meta.remote_cwd,
            resolved.api_key.is_some() || resolved.auth_token.is_some()
        );
    }

    // 3. Resolve resume session_id
    let resume_session_id = match &session_mode {
        SessionMode::Resume | SessionMode::Continue => {
            let sid = session_id
                .or_else(|| meta.session_id.clone())
                .ok_or_else(|| {
                    format!(
                        "session_id required for {:?} but not found in params or run metadata",
                        session_mode
                    )
                })?;
            Some(sid)
        }
        SessionMode::Fork => {
            return Err(
                "Fork mode not supported in start_session — use fork_session command instead"
                    .into(),
            );
        }
        SessionMode::New => None,
    };

    // Validate
    adapter::validate_session_params(&adapter_settings, &session_mode)?;

    let is_new = matches!(session_mode, SessionMode::New);

    // Preflight: check base_url reachability
    // Skip for SSH remote — reachability depends on remote host's network
    if remote.is_none() {
        if let Err(e) = preflight_check_base_url(resolved.base_url.as_deref(), effective_pid).await
        {
            // Only mark as Failed for new runs still in Pending — don't overwrite history
            if is_new && meta.status == RunStatus::Pending {
                storage::runs::update_status(&run_id, RunStatus::Failed, None, Some(e.clone()))
                    .ok();
            }
            return Err(e);
        }
    }

    // 4. Emit RunState(spawning) — UserMessage now handled by actor
    let spawning_event = BusEvent::RunState {
        run_id: run_id.clone(),
        state: "spawning".to_string(),
        exit_code: None,
        error: None,
    };
    emitter.persist_and_emit(&run_id, &spawning_event);
    storage::runs::update_status(&run_id, RunStatus::Running, None, None).ok();

    // 5. Stop any existing actor for this run_id
    let had_session = stop_actor(sessions, &run_id).await?;
    if had_session {
        log::debug!(
            "[session] old actor teardown complete for run_id={}",
            run_id
        );
    }

    // 6. Spawn CLI process (no initial stdin write — actor handles it).
    // Codex on the session_actor path uses the bidirectional `codex app-server` transport;
    // everything else (Claude, and Codex would-be-pipe_exec) uses the stream-json child.
    let effective_cwd = meta.remote_cwd.as_deref().unwrap_or(&meta.cwd);
    let is_codex = meta.agent == "codex";
    let (child, stdin, stdout, stderr, codex, codex_startup) = if is_codex {
        if remote.is_some() {
            return Err("Codex app-server transport is not supported on remote hosts yet".into());
        }
        let (c, si, so, se) = spawn_codex_appserver_process(
            effective_cwd,
            &adapter_settings,
            resolved.extra_env.as_ref(),
        )
        .await?;
        let resume_tid = meta.resolved_conversation_ref().and_then(|r| match r {
            ConversationRef::CodexThread(t) => Some(t),
            _ => None,
        });
        let mut driver = CodexAppServer::new();
        let ctx = StartupCtx {
            cwd: effective_cwd.to_string(),
            resume_thread_id: resume_tid,
            model: adapter_settings.model.clone(),
            model_provider: adapter_settings
                .codex_provider
                .as_ref()
                .map(|p| p.id.clone()),
            // Approval policy derived from permission_mode (mirrors the sandbox mapping):
            // bypassPermissions/dontAsk → "never" (no prompts, matches danger-full-access),
            // everything else → "on-request" (the interactive policy Codex's own TUI uses;
            // surfaces an approval card when a command needs to escape the sandbox).
            approval_policy: Some(codex_approval_for(
                adapter_settings.permission_mode.as_deref(),
            )),
            sandbox: Some(codex_sandbox_for(
                adapter_settings.permission_mode.as_deref(),
            )),
            effort: adapter_settings.effort.clone().filter(|e| !e.is_empty()),
            add_dirs: adapter_settings.add_dirs.clone(),
        };
        let startup = driver.startup_messages(&ctx);
        (c, si, so, se, Some(driver), startup)
    } else {
        let (c, si, so, se) = spawn_cli_process(
            effective_cwd,
            &meta.prompt,
            &adapter_settings,
            &session_mode,
            resume_session_id.as_deref(),
            is_new,
            &att_list,
            remote.as_ref(),
            meta.remote_cwd.as_deref(),
            resolved.api_key.as_deref(),
            resolved.auth_token.as_deref(),
            resolved.base_url.as_deref(),
            &run_id,
            resolved.models.as_deref(),
            resolved.extra_env.as_ref(),
        )
        .await?;
        (c, si, so, se, None, vec![])
    };

    // 7. Compute turn baselines — 1-based: next_turn_index = N means next message gets turnIndex=N.
    // New session: first message gets turnIndex=1. Resume: first new message gets total+1.
    let (initial_turn_index, initial_auto_ctx_id) = if is_new {
        (1_u32, 1_u32)
    } else {
        let (total, normal) = crate::storage::events::count_user_messages(&run_id);
        (total + 1, normal + 1)
    };
    log::debug!(
        "[session] turn baselines: initial_turn_index={}, initial_auto_ctx_id={}",
        initial_turn_index,
        initial_auto_ctx_id
    );

    // 8. Spawn actor
    let actor_handle = session_actor::spawn_actor(
        Arc::clone(emitter),
        sessions.clone(),
        run_id.clone(),
        child,
        stdin,
        stdout,
        stderr,
        !is_new,
        cancel_token.clone(),
        initial_turn_index,
        initial_auto_ctx_id,
        codex,
        codex_startup,
    );
    let cmd_tx = actor_handle.cmd_tx.clone();
    sessions.lock().await.insert(run_id.clone(), actor_handle);

    // 9. Send initial message through actor (unified entry point for Turn Engine).
    // Prefer an explicitly-provided follow-up message; fall back to the stored prompt only
    // for a brand-new run's first turn. A stopped session re-spawned with a new message
    // passes initial_message (mode defaults to New, but the Codex thread resumes via
    // conversation_ref) — it must send that message, NOT re-run the original prompt.
    let initial_text = initial_message.clone().or_else(|| {
        if is_new {
            Some(meta.prompt.clone())
        } else {
            None
        }
    });
    if let Some(text) = initial_text {
        let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
        cmd_tx
            .send(ActorCommand::SendMessage {
                text,
                attachments: att_list,
                skills: Vec::new(),
                reply: reply_tx,
            })
            .await
            .map_err(|_| "Actor dead before initial message".to_string())?;
        reply_rx
            .await
            .map_err(|_| "Actor dropped initial message reply".to_string())??;
        log::debug!(
            "[session] initial message sent through actor for run_id={}",
            run_id
        );
    } else {
        // Resume/continue without message: emit synthetic idle so frontend shows input box
        let idle_event = BusEvent::RunState {
            run_id: run_id.clone(),
            state: "idle".to_string(),
            exit_code: None,
            error: None,
        };
        emitter.persist_and_emit(&run_id, &idle_event);
        // Persist idle status (allows Pending→Idle, not just Running→Idle)
        let should_update = storage::runs::get_run(&run_id)
            .map(|m| m.status != RunStatus::Idle)
            .unwrap_or(false);
        if should_update {
            if let Err(e) = storage::runs::update_status(&run_id, RunStatus::Idle, None, None) {
                log::warn!("[session] synthetic idle meta update failed: {}", e);
            } else {
                emitter.emit_realtime(
                    "ocv:status-changed",
                    &serde_json::json!({"run_id": run_id.as_str(), "status": "idle"}),
                    Some(&run_id),
                );
            }
        }
        log::debug!(
            "[session] resume/continue: emitted synthetic RunState(idle) for run_id={}",
            run_id
        );
    }

    log::debug!("[session] actor spawned successfully for run_id={}", run_id);
    Ok(())
}

#[allow(clippy::too_many_arguments)]
#[tauri::command]
pub async fn start_session(
    emitter: State<'_, Arc<BroadcastEmitter>>,
    sessions: State<'_, ActorSessionMap>,
    spawn_locks: State<'_, SpawnLocks>,
    cancel_token: State<'_, CancellationToken>,
    run_id: String,
    mode: Option<SessionMode>,
    session_id: Option<String>,
    initial_message: Option<String>,
    attachments: Option<Vec<AttachmentData>>,
    platform_id: Option<String>,
    permission_mode_override: Option<String>,
) -> Result<(), String> {
    start_session_impl(
        emitter.inner(),
        sessions.inner(),
        spawn_locks.inner(),
        cancel_token.inner(),
        run_id,
        mode,
        session_id,
        initial_message,
        attachments,
        platform_id,
        permission_mode_override,
    )
    .await
}

#[tauri::command]
pub async fn send_session_message(
    sessions: State<'_, ActorSessionMap>,
    run_id: String,
    message: String,
    attachments: Option<Vec<AttachmentData>>,
    skills: Option<Vec<CodexSkillRef>>,
) -> Result<(), String> {
    // No SpawnLock — data operation, routed through actor channel
    let att_count = attachments.as_ref().map_or(0, |v| v.len());
    let skill_count = skills.as_ref().map_or(0, |v| v.len());
    log::debug!(
        "[session] send_session_message: run_id={}, msg_len={}, attachments={}, skills={}",
        run_id,
        message.len(),
        att_count,
        skill_count
    );

    // Get channel sender
    let cmd_tx = get_cmd_tx(&sessions, &run_id).await?;

    // Send message through actor channel
    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
    cmd_tx
        .send(ActorCommand::SendMessage {
            text: message.clone(),
            attachments: attachments.unwrap_or_default(),
            skills: skills.unwrap_or_default(),
            reply: reply_tx,
        })
        .await
        .map_err(|_| "Actor dead".to_string())?;
    reply_rx
        .await
        .map_err(|_| "Actor dropped reply".to_string())??;

    // Turn Transaction Engine: actor's start_user_turn now handles
    // UserMessage + RunState(running) emission. No post-emit needed here.

    log::debug!(
        "[session] send_session_message: delivered to actor, run_id={}",
        run_id
    );
    Ok(())
}

pub(crate) async fn stop_session_impl(
    emitter: &Arc<BroadcastEmitter>,
    sessions: &ActorSessionMap,
    spawn_locks: &SpawnLocks,
    run_id: String,
) -> Result<(), String> {
    let _guard = spawn_locks.acquire(&run_id).await;

    let was_active = stop_actor(sessions, &run_id).await?;
    if was_active {
        // Actor was active — emit stopped
        let event = BusEvent::RunState {
            run_id: run_id.clone(),
            state: "stopped".to_string(),
            exit_code: None,
            error: None,
        };
        emitter.persist_and_emit(&run_id, &event);
        storage::runs::update_status(&run_id, RunStatus::Stopped, None, None).ok();
    }

    Ok(())
}

#[tauri::command]
pub async fn stop_session(
    emitter: State<'_, Arc<BroadcastEmitter>>,
    sessions: State<'_, ActorSessionMap>,
    spawn_locks: State<'_, SpawnLocks>,
    run_id: String,
) -> Result<(), String> {
    stop_session_impl(
        emitter.inner(),
        sessions.inner(),
        spawn_locks.inner(),
        run_id,
    )
    .await
}

#[tauri::command]
pub async fn send_session_control(
    sessions: State<'_, ActorSessionMap>,
    run_id: String,
    subtype: String,
    params: Option<serde_json::Value>,
) -> Result<serde_json::Value, String> {
    // No SpawnLock — data operation through actor channel
    log::debug!(
        "[session] send_session_control: run_id={}, subtype={}",
        run_id,
        subtype
    );

    let cmd_tx = get_cmd_tx(&sessions, &run_id).await?;

    // Build control request
    let mut request = serde_json::json!({ "subtype": subtype });
    if let Some(p) = params {
        if let Some(obj) = p.as_object() {
            for (k, v) in obj {
                request[k] = v.clone();
            }
        }
    }

    // Phase 1: send control request, get response receiver
    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
    cmd_tx
        .send(ActorCommand::SendControl {
            request,
            reply: reply_tx,
        })
        .await
        .map_err(|_| "Actor dead".to_string())?;

    let (request_id, response_rx) = reply_rx
        .await
        .map_err(|_| "Actor dropped reply".to_string())??;

    // Phase 2: await response outside actor (no lock held)
    match tokio::time::timeout(std::time::Duration::from_secs(10), response_rx).await {
        Ok(Ok(response)) => {
            log::debug!(
                "[session] control response received for req_id={}",
                request_id
            );
            Ok(response)
        }
        Ok(Err(_)) => {
            log::warn!(
                "[session] control response channel closed for req_id={}",
                request_id
            );
            Err("Control response channel closed (session may have ended)".to_string())
        }
        Err(_) => {
            log::warn!(
                "[session] control request timed out for req_id={}",
                request_id
            );
            Err("Timeout waiting for control response".to_string())
        }
    }
}

/// Broadcast mcp_toggle to ALL active sessions (fire-and-forget, best-effort).
#[tauri::command]
pub async fn broadcast_mcp_toggle(
    sessions: State<'_, ActorSessionMap>,
    server_name: String,
    enabled: bool,
) -> Result<u32, String> {
    let senders: Vec<(String, tokio::sync::mpsc::Sender<ActorCommand>)> = {
        let map = sessions.lock().await;
        map.iter()
            .map(|(id, h)| (id.clone(), h.cmd_tx.clone()))
            .collect()
    };
    let request = serde_json::json!({
        "subtype": "mcp_toggle",
        "serverName": server_name,
        "enabled": enabled,
    });
    let mut sent: u32 = 0;
    for (run_id, tx) in &senders {
        let (reply_tx, _reply_rx) = tokio::sync::oneshot::channel();
        if tx
            .send(ActorCommand::SendControl {
                request: request.clone(),
                reply: reply_tx,
            })
            .await
            .is_ok()
        {
            sent += 1;
            log::debug!(
                "[session] broadcast_mcp_toggle: sent to run_id={}, server={}, enabled={}",
                run_id,
                server_name,
                enabled,
            );
        }
    }
    log::debug!(
        "[session] broadcast_mcp_toggle: sent to {}/{} sessions",
        sent,
        senders.len()
    );
    Ok(sent)
}

#[tauri::command]
pub fn get_bus_events(
    id: String,
    since_seq: Option<u64>,
) -> Result<Vec<serde_json::Value>, String> {
    storage::runs::get_run(&id).ok_or_else(|| format!("Run {} not found", id))?;
    Ok(storage::events::list_bus_events(&id, since_seq))
}

pub(crate) async fn fork_session_impl(
    emitter: &Arc<BroadcastEmitter>,
    sessions: &ActorSessionMap,
    spawn_locks: &SpawnLocks,
    run_id: String,
) -> Result<String, String> {
    let _guard = spawn_locks.acquire(&run_id).await;
    log::debug!("[session] fork_session: source run_id={}", run_id);

    // 1. Read source run metadata
    let source =
        storage::runs::get_run(&run_id).ok_or_else(|| format!("Run {} not found", run_id))?;

    // Codex forks through the LIVE app-server (thread/fork returns a new thread id) — no oneshot
    // process, no session_id. Delegate to the Codex-specific path.
    if source.agent == "codex" {
        return fork_session_codex(sessions, &run_id, &source).await;
    }

    // Guard: the oneshot fork path below is Claude-only (session-id based).
    if source.agent != "claude" {
        return Err(format!(
            "Fork is not supported for {} sessions",
            source.agent
        ));
    }

    let session_id = source
        .session_id
        .clone()
        .ok_or_else(|| "No session_id available for fork".to_string())?;

    // 2. Stop source actor if alive
    let was_active = stop_actor(sessions, &run_id).await?;
    if was_active {
        log::debug!("[session] fork_session: stopped active source actor");
        let event = BusEvent::RunState {
            run_id: run_id.clone(),
            state: "stopped".to_string(),
            exit_code: None,
            error: None,
        };
        emitter.persist_and_emit(&run_id, &event);
        storage::runs::update_status(&run_id, RunStatus::Stopped, None, None).ok();
    }

    // 3. Create new run (audit #3: inherit remote_host_name + remote_cwd + snapshot + platform_id)
    let new_id = uuid::Uuid::new_v4().to_string();
    let mut meta = storage::runs::create_run(
        &new_id,
        &source.prompt,
        &source.cwd,
        &source.agent,
        RunStatus::Pending,
        source.model.clone(),
        Some(run_id.clone()),
        source.remote_host_name.clone(),
        source.remote_cwd.clone(),
        source.remote_host_snapshot.clone(),
        source.platform_id.clone(),
    )?;
    log::debug!(
        "[session] fork_session: fork {} ← parent {}, remote={:?}",
        new_id,
        run_id,
        source.remote_host_name
    );

    // 4. Copy parent events
    storage::events::copy_bus_events(&run_id, &new_id)?;

    // 5. Set parent session_id on fork run; inherit execution_path, but NOT conversation_ref
    //    (fork creates a new session — conversation_ref is written after fork_oneshot returns new ID)
    meta.session_id = Some(session_id.clone());
    meta.execution_path = Some(source.resolved_execution_path());
    // conversation_ref intentionally None — will be set in step 8 with new session_id
    storage::runs::save_meta(&meta)?;

    // 6. Build adapter settings + resolve remote (audit #3)
    let agent_settings = storage::settings::get_agent_settings(&source.agent);
    let user_settings = storage::settings::get_user_settings();
    let mut adapter = adapter::build_adapter_settings(&agent_settings, &user_settings, None);
    let remote = resolve_remote_host(&source)?;
    // CLI Auth mode: ignore platform_id — CLI manages its own connection
    let effective_pid = if user_settings.auth_mode == "cli" {
        None
    } else {
        source.platform_id.as_deref()
    };
    let resolved = resolve_auth_env_for_platform(&remote, &user_settings, effective_pid);
    adapter::clear_model_if_provider_overrides(
        &mut adapter,
        &None, // fork has no UI model override
        &agent_settings.model,
        &resolved.models,
    );
    let resolved = augment_with_shell_auth(
        resolved,
        &user_settings.auth_mode,
        remote.is_some(),
        &source.cwd,
    );
    let effective_cwd = source.remote_cwd.as_deref().unwrap_or(&source.cwd);

    // 7. One-shot fork: get new session_id
    log::debug!(
        "[session] fork_session: starting fork_oneshot, source_sid={}, remote={:?}",
        session_id,
        remote.as_ref().map(|r| &r.name)
    );
    let new_session_id = match claude_stream::fork_oneshot(
        &session_id,
        effective_cwd,
        &adapter,
        remote.as_ref(),
        resolved.api_key.as_deref(),
        resolved.auth_token.as_deref(),
        resolved.base_url.as_deref(),
        resolved.models.as_deref(),
        resolved.extra_env.as_ref(),
    )
    .await
    {
        Ok(sid) => sid,
        Err(e) => {
            log::error!(
                "[session] fork_oneshot failed, cleaning up run {}: {}",
                new_id,
                e
            );
            storage::runs::update_status(&new_id, RunStatus::Failed, None, Some(e.clone()))?;
            return Err(e);
        }
    };
    log::debug!(
        "[session] fork_session: fork_oneshot returned new_sid={}",
        new_session_id
    );

    // 8. Persist new session_id + conversation_ref (one write)
    meta.session_id = Some(new_session_id.clone());
    meta.conversation_ref = Some(crate::models::ConversationRef::ClaudeSession(
        new_session_id,
    ));
    storage::runs::save_meta(&meta)?;

    log::debug!(
        "[session] fork_session completed: {} → {} (frontend will start_session to connect)",
        run_id,
        new_id
    );
    Ok(new_id)
}

/// Codex fork: unlike Claude (oneshot process + new session_id), Codex forks via the LIVE
/// app-server — `thread/fork` returns a brand-new thread id at `result.thread.id`. We send the
/// `fork` control to the running actor, await the new thread id, then create a new run pointing
/// at that thread (`ConversationRef::CodexThread`) with the source's events copied so the
/// frontend can switch to it (same contract as Claude: returns the new run id).
///
/// The source actor stays ALIVE (no stop) — fork is non-destructive; both threads remain usable.
/// Requires a live session (the thread must exist server-side to be forked).
async fn fork_session_codex(
    sessions: &ActorSessionMap,
    run_id: &str,
    source: &RunMeta,
) -> Result<String, String> {
    // 1. The thread must be live to fork — get the actor command channel.
    let cmd_tx = get_cmd_tx(sessions, run_id)
        .await
        .map_err(|_| "Fork requires a live Codex session (start it first)".to_string())?;

    // 2. Send the `fork` control through the actor; await the JSON-RPC reply.
    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
    cmd_tx
        .send(ActorCommand::SendControl {
            request: serde_json::json!({ "subtype": "fork" }),
            reply: reply_tx,
        })
        .await
        .map_err(|_| "Actor dead".to_string())?;
    let (_, response_rx) = reply_rx
        .await
        .map_err(|_| "Actor dropped fork reply".to_string())??;
    let response = tokio::time::timeout(std::time::Duration::from_secs(10), response_rx)
        .await
        .map_err(|_| "Timeout waiting for thread/fork response".to_string())?
        .map_err(|_| "Fork response channel closed (session may have ended)".to_string())?;

    // 3. Extract the new thread id from `result.thread.id`.
    let new_thread_id = response
        .get("thread")
        .and_then(|t| t.get("id"))
        .and_then(|v| v.as_str())
        .ok_or_else(|| {
            format!(
                "thread/fork response missing thread.id: {}",
                truncate_str(&response.to_string(), 200)
            )
        })?
        .to_string();
    log::debug!(
        "[session] fork_session_codex: source {} → new thread {}",
        run_id,
        new_thread_id
    );

    // 4. Create the new run (inherit prompt/cwd/model/remote/platform; parent = source run).
    let new_id = uuid::Uuid::new_v4().to_string();
    let mut meta = storage::runs::create_run(
        &new_id,
        &source.prompt,
        &source.cwd,
        &source.agent,
        RunStatus::Pending,
        source.model.clone(),
        Some(run_id.to_string()),
        source.remote_host_name.clone(),
        source.remote_cwd.clone(),
        source.remote_host_snapshot.clone(),
        source.platform_id.clone(),
    )?;

    // 5. Copy parent events so the forked timeline shows the shared history.
    storage::events::copy_bus_events(run_id, &new_id)?;

    // 6. Point the new run at the forked thread; inherit execution_path.
    meta.execution_path = Some(source.resolved_execution_path());
    meta.conversation_ref = Some(ConversationRef::CodexThread(new_thread_id));
    storage::runs::save_meta(&meta)?;

    log::debug!(
        "[session] fork_session_codex completed: {} → {} (frontend will start_session to connect)",
        run_id,
        new_id
    );
    Ok(new_id)
}

#[tauri::command]
pub async fn fork_session(
    emitter: State<'_, Arc<BroadcastEmitter>>,
    sessions: State<'_, ActorSessionMap>,
    spawn_locks: State<'_, SpawnLocks>,
    run_id: String,
) -> Result<String, String> {
    fork_session_impl(
        emitter.inner(),
        sessions.inner(),
        spawn_locks.inner(),
        run_id,
    )
    .await
}

pub(crate) async fn approve_session_tool_impl(
    emitter: &Arc<BroadcastEmitter>,
    sessions: &ActorSessionMap,
    spawn_locks: &SpawnLocks,
    cancel_token: &CancellationToken,
    run_id: String,
    tool_name: String,
) -> Result<(), String> {
    let _guard = spawn_locks.acquire(&run_id).await;
    log::debug!(
        "[session] approve_session_tool: run_id={}, tool={}",
        run_id,
        tool_name
    );

    // Tools that must never be permanently allowed — they require per-use approval.
    // ExitPlanMode: plan approval gate; adding it to allowedTools silently bypasses
    // the CLI's requiresUserInteraction check, permanently auto-approving plans.
    const NEVER_ALLOW_TOOLS: &[&str] = &["ExitPlanMode", "EnterPlanMode"];

    if NEVER_ALLOW_TOOLS.contains(&tool_name.as_str()) {
        log::warn!(
            "[session] approve_session_tool: refusing to permanently allow '{}' (requires per-use approval)",
            tool_name
        );
        return Err(format!(
            "'{}' cannot be permanently allowed — it requires approval each time",
            tool_name
        ));
    }

    // 1. Read run metadata
    let meta =
        storage::runs::get_run(&run_id).ok_or_else(|| format!("Run {} not found", run_id))?;

    // 2. Persist tool to agent allowed_tools
    let mut agent_settings = storage::settings::get_agent_settings(&meta.agent);
    if !agent_settings.allowed_tools.contains(&tool_name) {
        agent_settings.allowed_tools.push(tool_name.clone());
        let patch = serde_json::json!({
            "allowed_tools": agent_settings.allowed_tools,
        });
        storage::settings::update_agent_settings(&meta.agent, patch)?;
        log::debug!(
            "[session] added {} to allowed_tools for {}",
            tool_name,
            meta.agent
        );
    }

    // 3. Resolve remote + auth BEFORE stopping actor (preflight failure keeps old actor alive)
    let remote = resolve_remote_host(&meta)?;
    let effective_cwd = meta.remote_cwd.clone().unwrap_or_else(|| meta.cwd.clone());
    let prompt = meta.prompt.clone();
    let session_id = meta
        .session_id
        .clone()
        .ok_or_else(|| "No session_id for continue".to_string())?;

    let refreshed_agent = storage::settings::get_agent_settings(&meta.agent);
    let user = storage::settings::get_user_settings();
    let mut adapter = adapter::build_adapter_settings(&refreshed_agent, &user, None);
    // CLI Auth mode: ignore platform_id — CLI manages its own connection
    let effective_pid = if user.auth_mode == "cli" {
        None
    } else {
        meta.platform_id.as_deref()
    };
    let resolved = resolve_auth_env_for_platform(&remote, &user, effective_pid);
    adapter::clear_model_if_provider_overrides(
        &mut adapter,
        &None,
        &refreshed_agent.model,
        &resolved.models,
    );
    let resolved = augment_with_shell_auth(resolved, &user.auth_mode, remote.is_some(), &meta.cwd);

    // 4. Preflight — before killing old actor so session can recover on failure
    if remote.is_none() {
        preflight_check_base_url(resolved.base_url.as_deref(), effective_pid).await?;
    }

    // 5. Now safe to stop current actor
    stop_actor(sessions, &run_id).await?;

    // 6. Emit spawning
    let spawning_event = BusEvent::RunState {
        run_id: run_id.clone(),
        state: "spawning".to_string(),
        exit_code: None,
        error: None,
    };
    emitter.persist_and_emit(&run_id, &spawning_event);
    storage::runs::update_status(&run_id, RunStatus::Running, None, None).ok();

    // 8. Spawn CLI with Continue mode (audit #3: SSH path if remote)
    let (child, stdin, stdout, stderr) = spawn_cli_process(
        &effective_cwd,
        &prompt,
        &adapter,
        &SessionMode::Continue,
        Some(&session_id),
        false,
        &[], // approve_session_tool: no attachments
        remote.as_ref(),
        Some(&effective_cwd),
        resolved.api_key.as_deref(),
        resolved.auth_token.as_deref(),
        resolved.base_url.as_deref(),
        &run_id,
        resolved.models.as_deref(),
        resolved.extra_env.as_ref(),
    )
    .await?;

    // 8. Compute turn baselines from existing events (1-based: next message gets total+1)
    let (total, normal) = crate::storage::events::count_user_messages(&run_id);
    log::debug!(
        "[session] approve: turn baselines total={}, normal={} → next=({}, {})",
        total,
        normal,
        total + 1,
        normal + 1
    );

    // 8b. Spawn actor
    let actor_handle = session_actor::spawn_actor(
        Arc::clone(emitter),
        sessions.clone(),
        run_id.clone(),
        child,
        stdin,
        stdout,
        stderr,
        true, // is_resume
        cancel_token.clone(),
        total + 1,
        normal + 1,
        None, // Claude transport
        vec![],
    );
    sessions.lock().await.insert(run_id.clone(), actor_handle);

    // 9. Wait briefly for CLI to be ready.
    // TODO: Replace with event-based approach (WaitForReady actor command).
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;

    // 10. Send retry guidance message via actor.
    // Turn Transaction Engine: actor's start_user_turn handles UserMessage + RunState(running).
    let retry_msg = format!(
        "The tool {} is now allowed. Please retry your previous action using this tool.",
        tool_name
    );
    let cmd_tx = get_cmd_tx(sessions, &run_id).await?;
    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
    cmd_tx
        .send(ActorCommand::SendMessage {
            text: retry_msg,
            attachments: Vec::new(),
            skills: Vec::new(),
            reply: reply_tx,
        })
        .await
        .map_err(|_| "Actor dead after approve restart".to_string())?;
    reply_rx
        .await
        .map_err(|_| "Actor dropped reply".to_string())??;

    log::debug!(
        "[session] approve_session_tool completed for run_id={}",
        run_id
    );
    Ok(())
}

#[tauri::command]
pub async fn approve_session_tool(
    emitter: State<'_, Arc<BroadcastEmitter>>,
    sessions: State<'_, ActorSessionMap>,
    spawn_locks: State<'_, SpawnLocks>,
    cancel_token: State<'_, CancellationToken>,
    run_id: String,
    tool_name: String,
) -> Result<(), String> {
    approve_session_tool_impl(
        emitter.inner(),
        sessions.inner(),
        spawn_locks.inner(),
        cancel_token.inner(),
        run_id,
        tool_name,
    )
    .await
}

/// Respond to an inline permission prompt (--permission-prompt-tool stdio).
/// Writes a control_response back to CLI stdin via the actor.
#[tauri::command]
#[allow(clippy::too_many_arguments)]
pub async fn respond_permission(
    sessions: State<'_, ActorSessionMap>,
    run_id: String,
    request_id: String,
    behavior: String,
    updated_permissions: Option<Vec<serde_json::Value>>,
    updated_input: Option<serde_json::Value>,
    deny_message: Option<String>,
    interrupt: Option<bool>,
) -> Result<(), String> {
    log::debug!(
        "[session] respond_permission: run_id={}, req_id={}, behavior={}, updated_perms={}, has_updated_input={}, has_deny_message={}, interrupt={:?}",
        run_id,
        request_id,
        behavior,
        updated_permissions.as_ref().map_or(0, |v| v.len()),
        updated_input.is_some(),
        deny_message.is_some(),
        interrupt,
    );

    let cmd_tx = get_cmd_tx(&sessions, &run_id).await?;

    // Build the response payload for Claude CLI.
    // CLI validates with Zod: allow requires `updatedInput` (record<string,unknown>),
    // deny requires `message` (string). Missing fields cause ZodError.
    let mut response = if behavior == "allow" {
        // updatedInput is REQUIRED by CLI schema — use provided value or empty object
        let input_val = updated_input.unwrap_or_else(|| serde_json::json!({}));
        serde_json::json!({
            "behavior": "allow",
            "updatedInput": input_val,
        })
    } else {
        let msg = deny_message.unwrap_or_else(|| "User denied permission".to_string());
        let mut deny_obj = serde_json::json!({
            "behavior": "deny",
            "message": msg,
        });
        if interrupt == Some(true) {
            deny_obj["interrupt"] = serde_json::json!(true);
        }
        deny_obj
    };
    // Include updatedPermissions when allowing with suggestions (camelCase per CLI schema)
    if let Some(perms) = updated_permissions {
        if behavior == "allow" && !perms.is_empty() {
            response["updatedPermissions"] = serde_json::Value::Array(perms);
        }
    }

    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
    cmd_tx
        .send(ActorCommand::RespondPermission {
            request_id: request_id.clone(),
            response,
            reply: reply_tx,
        })
        .await
        .map_err(|_| "Actor dead".to_string())?;
    reply_rx
        .await
        .map_err(|_| "Actor dropped reply".to_string())??;

    log::debug!(
        "[session] respond_permission: delivered req_id={}",
        request_id
    );
    Ok(())
}

/// Respond to a hook callback control request (PreToolUse hooks only).
/// Writes a control_response back to CLI stdin via the actor.
#[tauri::command]
pub async fn respond_hook_callback(
    sessions: State<'_, ActorSessionMap>,
    run_id: String,
    request_id: String,
    decision: String, // "allow", "deny", or "defer" (defer pauses tool call in headless sessions)
    // PreToolUse hooks can rewrite tool input alongside `decision: "allow"` (CLI v2.1.85+).
    // Only honored when decision == "allow"; CLI ignores it for deny/defer.
    updated_input: Option<serde_json::Value>,
) -> Result<(), String> {
    log::debug!(
        "[session] respond_hook_callback: run_id={}, req_id={}, decision={}, has_updated_input={}",
        run_id,
        request_id,
        decision,
        updated_input.is_some(),
    );

    let cmd_tx = get_cmd_tx(&sessions, &run_id).await?;

    let mut response = serde_json::json!({ "decision": decision });
    if decision == "allow" {
        if let Some(input) = updated_input {
            response["updatedInput"] = input;
        }
    }

    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
    cmd_tx
        .send(ActorCommand::RespondHookCallback {
            request_id: request_id.clone(),
            response,
            reply: reply_tx,
        })
        .await
        .map_err(|_| "Actor dead".to_string())?;
    reply_rx
        .await
        .map_err(|_| "Actor dropped reply".to_string())??;

    log::debug!(
        "[session] respond_hook_callback: delivered req_id={}",
        request_id
    );
    Ok(())
}

/// Cancel a pending control_request (top-level message type, not a control_request subtype).
#[tauri::command]
pub async fn cancel_control_request(
    sessions: State<'_, ActorSessionMap>,
    run_id: String,
    request_id: String,
) -> Result<(), String> {
    log::debug!(
        "[session] cancel_control_request: run_id={}, req_id={}",
        run_id,
        request_id
    );

    let cmd_tx = get_cmd_tx(&sessions, &run_id).await?;

    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
    cmd_tx
        .send(ActorCommand::CancelControlRequest {
            request_id,
            reply: reply_tx,
        })
        .await
        .map_err(|_| "Actor dead".to_string())?;
    reply_rx
        .await
        .map_err(|_| "Actor dropped reply".to_string())??;

    Ok(())
}

/// Respond to an MCP elicitation control request.
/// Answer a Codex `request_user_input` (multiple-choice) prompt. `answers` maps each
/// question id to the selected option label(s): `{ "<qid>": ["<label>", ...] }`. Routed to
/// the app-server as a JSON-RPC response on the pending request (Codex transport only).
#[tauri::command]
pub async fn respond_user_input(
    sessions: State<'_, ActorSessionMap>,
    run_id: String,
    request_id: String,
    answers: serde_json::Value,
) -> Result<(), String> {
    log::debug!(
        "[session] respond_user_input: run_id={}, req_id={}",
        run_id,
        request_id
    );
    if !answers.is_object() {
        return Err("answers must be a JSON object keyed by question id".into());
    }
    let response = serde_json::json!({ "answers": answers });

    let cmd_tx = get_cmd_tx(&sessions, &run_id).await?;
    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
    cmd_tx
        .send(ActorCommand::RespondUserInput {
            request_id: request_id.clone(),
            response,
            reply: reply_tx,
        })
        .await
        .map_err(|_| "Actor dead".to_string())?;
    reply_rx
        .await
        .map_err(|_| "Actor dropped reply".to_string())??;
    log::debug!(
        "[session] respond_user_input: delivered req_id={}",
        request_id
    );
    Ok(())
}

/// Writes a control_response back to CLI stdin via the actor.
#[tauri::command]
pub async fn respond_elicitation(
    sessions: State<'_, ActorSessionMap>,
    run_id: String,
    request_id: String,
    action: String,
    content: Option<serde_json::Value>,
) -> Result<(), String> {
    log::debug!(
        "[session] respond_elicitation: run_id={}, req_id={}, action={}",
        run_id,
        request_id,
        action
    );

    if !matches!(action.as_str(), "accept" | "decline" | "cancel") {
        return Err(format!("Invalid elicitation action: {}", action));
    }

    let response = match action.as_str() {
        "accept" => {
            let c = content.unwrap_or(serde_json::json!({}));
            if !c.is_object() {
                return Err("content must be a JSON object for accept".into());
            }
            serde_json::json!({"action": "accept", "content": c})
        }
        other => serde_json::json!({"action": other}),
    };

    let cmd_tx = get_cmd_tx(&sessions, &run_id).await?;

    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
    cmd_tx
        .send(ActorCommand::RespondElicitation {
            request_id: request_id.clone(),
            response,
            reply: reply_tx,
        })
        .await
        .map_err(|_| "Actor dead".to_string())?;
    reply_rx
        .await
        .map_err(|_| "Actor dropped reply".to_string())??;

    log::debug!(
        "[session] respond_elicitation: delivered req_id={}",
        request_id
    );
    Ok(())
}

// ── Shell config auth injection (CLI mode only) ──

/// Pure decision: should we skip injecting shell auth based on existing process env?
/// If EITHER ANTHROPIC_API_KEY or ANTHROPIC_AUTH_TOKEN is in process env (non-empty),
/// child inherits it — injecting the other would trigger env_remove mutual exclusion
/// (see spawn_cli_process key/token branches).
fn should_skip_env_injection(key_val: Option<&str>, token_val: Option<&str>) -> bool {
    let has_key = key_val.is_some_and(|v| !v.trim().is_empty());
    let has_token = token_val.is_some_and(|v| !v.trim().is_empty());
    has_key || has_token
}

/// Read CLI auth env vars missing from process environment, using shell config as fallback.
/// Only reads ANTHROPIC_API_KEY / ANTHROPIC_AUTH_TOKEN (not BASE_URL — see augment_with_shell_auth doc).
///
/// Shell config parsing limitations (inherited from onboarding.rs:289 `read_env_from_shell_config`):
/// - Supported: `export VAR=value`, `VAR=value`, single/double quoted values
/// - NOT supported: variable expansion ($OTHER_VAR), command substitution $(cmd),
///   multi-line values, conditional blocks (if/fi), sourced sub-files
/// - Returns the FIRST match found across shell config files (.zshrc → .zprofile → .bashrc → .bash_profile → .profile)
fn resolve_shell_auth() -> (Option<String>, Option<String>) {
    use super::onboarding::read_env_from_shell_config;

    let key_val = std::env::var("ANTHROPIC_API_KEY").ok();
    let token_val = std::env::var("ANTHROPIC_AUTH_TOKEN").ok();
    if should_skip_env_injection(key_val.as_deref(), token_val.as_deref()) {
        log::trace!("[session] shell_auth: process env has auth var, skip injection");
        return (None, None);
    }

    // Neither in process env — try shell config (key first, then token)
    if let Some((val, path)) = read_env_from_shell_config("ANTHROPIC_API_KEY") {
        log::debug!("[session] shell_auth: ANTHROPIC_API_KEY from {}", path);
        return (Some(val), None);
    }
    if let Some((val, path)) = read_env_from_shell_config("ANTHROPIC_AUTH_TOKEN") {
        log::debug!("[session] shell_auth: ANTHROPIC_AUTH_TOKEN from {}", path);
        return (None, Some(val));
    }

    (None, None)
}

/// Pure function: check if a JSON config value contains a non-empty auth key.
/// Checks both `apiKey` and `primaryApiKey` (used by Max/Team plans).
/// See SENSITIVE_KEYS in cli_config.rs:78.
fn config_value_has_auth_key(config: &serde_json::Value) -> bool {
    const AUTH_KEYS: &[&str] = &["apiKey", "primaryApiKey"];
    AUTH_KEYS.iter().any(|k| {
        config
            .get(k)
            .and_then(|v| v.as_str())
            .is_some_and(|s| !s.trim().is_empty())
    })
}

/// Check if any CLI config (user-level or project-level) contains an API key.
fn cli_config_has_auth_key(cwd: &str) -> bool {
    let user_config = crate::storage::cli_config::load_cli_config();
    if config_value_has_auth_key(&user_config) {
        log::trace!("[session] shell_auth: user-level CLI config has auth key, skip");
        return true;
    }

    let project_config = crate::storage::cli_config::load_project_cli_config(cwd);
    if config_value_has_auth_key(&project_config) {
        log::trace!("[session] shell_auth: project-level CLI config has auth key, skip");
        return true;
    }

    false
}

/// In CLI auth mode (local only), supplement resolved auth with shell config credentials.
///
/// Guards:
/// - CLI mode only (API mode manages its own credentials)
/// - Local only (remote respects forward_api_key=false — never inject)
/// - No existing credentials (don't override what resolve_auth_env produced)
/// - CLI config has no apiKey/primaryApiKey (user-level + project-level)
///
/// OAuth safety: CLI's own auth priority is OAuth > settings.json > env vars.
/// Even if we inject an env var, CLI will still prefer OAuth — the injected key
/// is only used when CLI has no higher-priority auth source.
///
/// Does NOT inject ANTHROPIC_BASE_URL: injecting a base_url would (a) trigger
/// preflight_check_base_url which blocks session start if unreachable, and
/// (b) affect routing even when CLI has OAuth that doesn't need a custom URL.
/// Users who need key+url together should use API mode in Settings.
fn augment_with_shell_auth(
    resolved: ResolvedAuth,
    auth_mode: &str,
    is_remote: bool,
    cwd: &str,
) -> ResolvedAuth {
    if auth_mode != "cli" {
        return resolved;
    }
    if is_remote {
        return resolved;
    }
    if resolved.api_key.is_some() || resolved.auth_token.is_some() {
        return resolved;
    }
    if cli_config_has_auth_key(cwd) {
        return resolved;
    }

    let (key, token) = resolve_shell_auth();
    if key.is_some() || token.is_some() {
        log::debug!(
            "[session] CLI+local: supplementing auth from shell config (key={}, token={})",
            key.is_some(),
            token.is_some()
        );
        ResolvedAuth {
            api_key: key,
            auth_token: token,
            ..resolved
        }
    } else {
        resolved
    }
}

// ── CLI process spawning (extracted from claude_stream.rs) ──

/// Spawn a Claude CLI process and return (Child, ChildStdin, ChildStdout, ChildStderr).
/// Sends the initial prompt via stdin for new sessions.
/// For remote sessions, wraps the CLI command in SSH.
#[allow(clippy::too_many_arguments)]
/// Map OpenCovibe permission_mode → Codex app-server `sandbox` mode.
pub(crate) fn codex_sandbox_for(perm: Option<&str>) -> String {
    match perm {
        Some("plan") => "read-only".to_string(),
        Some("bypassPermissions") | Some("dontAsk") => "danger-full-access".to_string(),
        _ => "workspace-write".to_string(),
    }
}

/// Map OpenCovibe permission_mode → Codex `AskForApproval` string. Mirrors the sandbox mapping
/// (`codex_sandbox_for`): the relaxed modes get full autonomy ("never" — no approval prompts to
/// match danger-full-access), everything else keeps the interactive "on-request" policy the TUI
/// uses (Codex surfaces an approval card when a command needs to escape the sandbox).
pub(crate) fn codex_approval_for(perm: Option<&str>) -> String {
    match perm {
        Some("bypassPermissions") | Some("dontAsk") => "never".to_string(),
        _ => "on-request".to_string(),
    }
}

/// Auto-fallback probe: does the installed Codex CLI support the app-server transport?
///
/// Codex now defaults to app-server (see `commands/runs.rs`), but an old/incompatible CLI
/// would reject `codex app-server --enable …` and the session would die before the handshake
/// (process spawns, then exits — NOT a spawn error, so it can't be caught at spawn time).
/// To avoid that, we probe once: `codex app-server --help` exits 0 AND lists the `--enable`
/// option only on a CLI new enough to run our interactive transport. If not, the run falls
/// back to the one-shot `exec` path. `--help` returns immediately, so this is a cheap sync
/// check; the result is cached for the process lifetime (re-probed only after an app restart,
/// e.g. following a Codex upgrade).
pub(crate) fn codex_appserver_supported() -> bool {
    use std::sync::OnceLock;
    static CACHE: OnceLock<bool> = OnceLock::new();
    *CACHE.get_or_init(|| {
        let Some(bin) = claude_stream::which_binary("codex") else {
            log::warn!("[codex] app-server probe: codex binary not found → exec fallback");
            return false;
        };
        let out = std::process::Command::new(&bin)
            .arg("app-server")
            .arg("--help")
            .env("PATH", claude_stream::augmented_path())
            .output();
        match out {
            Ok(o) => {
                let txt = format!(
                    "{}{}",
                    String::from_utf8_lossy(&o.stdout),
                    String::from_utf8_lossy(&o.stderr)
                );
                // Need both the subcommand (exit 0) and the `--enable` feature flag we rely on.
                let ok = o.status.success() && txt.contains("--enable");
                if ok {
                    log::debug!("[codex] app-server supported → using interactive transport");
                } else {
                    log::warn!(
                        "[codex] app-server unsupported (help exit={:?}, has --enable={}) → exec fallback",
                        o.status.code(),
                        txt.contains("--enable")
                    );
                }
                ok
            }
            Err(e) => {
                log::warn!("[codex] app-server probe failed: {} → exec fallback", e);
                false
            }
        }
    })
}

/// Spawn `codex app-server` (bidirectional JSON-RPC) for an interactive Codex session.
/// Local only — remote/SSH app-server is out of scope for v1. The `--enable
/// default_mode_request_user_input` flag is REQUIRED for the multiple-choice tool to fire
/// in normal sessions (verified codex 0.136 — otherwise "unavailable in Default mode").
async fn spawn_codex_appserver_process(
    cwd: &str,
    settings: &adapter::AdapterSettings,
    extra_env: Option<&std::collections::HashMap<String, String>>,
) -> Result<
    (
        tokio::process::Child,
        tokio::process::ChildStdin,
        tokio::process::ChildStdout,
        tokio::process::ChildStderr,
    ),
    String,
> {
    use tokio::process::Command;

    let codex_bin = claude_stream::which_binary("codex")
        .ok_or_else(|| "Codex CLI not found in PATH".to_string())?;

    let mut args: Vec<String> = vec![
        "app-server".into(),
        "--enable".into(),
        "default_mode_request_user_input".into(),
        "-c".into(),
        "suppress_unstable_features_warning=true".into(),
    ];

    // Third-party provider overrides (shared with the exec + side-question paths). The provider
    // API key is injected as an env var (env_key=api_key) below, mirroring chat.rs's run_agent.
    if let Some(p) = &settings.codex_provider {
        args.extend(crate::agent::spawn::codex_provider_config_args(p));
    }

    let mut cmd = Command::new(&codex_bin);
    for a in &args {
        cmd.arg(a);
    }
    let path_env = claude_stream::augmented_path();
    cmd.current_dir(cwd)
        .env("PATH", &path_env)
        .env_remove("ANTHROPIC_API_KEY")
        .env_remove("ANTHROPIC_AUTH_TOKEN")
        .env_remove("CLAUDECODE")
        .stdin(std::process::Stdio::piped())
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .hide_console()
        .kill_on_drop(true);
    if let Some(env) = extra_env {
        for (k, v) in env {
            cmd.env(k, v);
        }
    }
    // Provider API key (env_key=api_key) — the exec path sets this in chat.rs's run_agent, but
    // the app-server actor path has no such hook, so inject it here.
    if let Some(p) = &settings.codex_provider {
        if let Some((k, v)) = crate::agent::spawn::codex_provider_env(p) {
            cmd.env(k, v);
        }
    }

    let mut child = cmd
        .spawn()
        .map_err(|e| format!("Failed to spawn codex app-server: {}", e))?;
    let stdin = child.stdin.take().ok_or("no app-server stdin")?;
    let stdout = child.stdout.take().ok_or("no app-server stdout")?;
    let stderr = child.stderr.take().ok_or("no app-server stderr")?;
    Ok((child, stdin, stdout, stderr))
}

#[allow(clippy::too_many_arguments)]
async fn spawn_cli_process(
    cwd: &str,
    prompt: &str,
    settings: &adapter::AdapterSettings,
    session_mode: &SessionMode,
    resume_session_id: Option<&str>,
    _is_new: bool,
    _initial_attachments: &[AttachmentData],
    remote_host: Option<&RemoteHost>,
    remote_cwd: Option<&str>,
    api_key: Option<&str>,
    auth_token: Option<&str>,
    base_url: Option<&str>,
    _run_id: &str,
    models: Option<&[String]>,
    extra_env: Option<&std::collections::HashMap<String, String>>,
) -> Result<
    (
        tokio::process::Child,
        tokio::process::ChildStdin,
        tokio::process::ChildStdout,
        tokio::process::ChildStderr,
    ),
    String,
> {
    // Build CLI args (shared between local and remote)
    let mut claude_args: Vec<String> = vec![
        "--output-format".into(),
        "stream-json".into(),
        "--input-format".into(),
        "stream-json".into(),
        "--verbose".into(),
        "--permission-prompt-tool".into(),
        "stdio".into(),
    ];

    // Session mode args
    match session_mode {
        SessionMode::Resume | SessionMode::Continue => {
            let sid = resume_session_id.ok_or("session_id required for resume/continue")?;
            claude_args.push("--resume".into());
            claude_args.push(sid.into());
        }
        SessionMode::Fork => {
            return Err("Fork mode not supported in spawn_cli_process — use fork_oneshot()".into());
        }
        SessionMode::New => {}
    }

    // Settings flags
    let flag_args = adapter::build_settings_args(settings, false);
    claude_args.extend(flag_args.iter().cloned());
    if settings.include_partial_messages {
        claude_args.push("--include-partial-messages".into());
    }

    log::debug!(
        "[session] session_mode={:?}, resume_id={:?}, flag_args={:?}, remote={:?}",
        session_mode,
        resume_session_id,
        flag_args,
        remote_host.map(|r| &r.name),
    );

    let mut child = if let Some(remote) = remote_host {
        // SSH branch: wrap claude command in ssh
        let effective_remote_cwd = remote_cwd.unwrap_or(cwd);
        let remote_cmd = crate::agent::ssh::build_remote_claude_command(
            remote,
            effective_remote_cwd,
            &claude_args,
            api_key,
            auth_token,
            base_url,
            models,
            extra_env,
        );
        let mut ssh_cmd = crate::agent::ssh::build_ssh_command(remote, &remote_cmd);
        ssh_cmd
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped());

        log::debug!(
            "[session] spawning remote CLI via SSH: {}@{}, cwd={}",
            remote.user,
            remote.host,
            effective_remote_cwd
        );

        ssh_cmd
            .hide_console()
            .kill_on_drop(true)
            .spawn()
            .map_err(|e| {
                log::error!("[session] Failed to spawn ssh: {}", e);
                format!("Failed to spawn ssh: {}", e)
            })?
    } else {
        // Local branch: existing logic
        let claude_bin = claude_stream::resolve_claude_path();
        log::debug!("[session] resolved binary: {}", claude_bin);

        let mut cmd = tokio::process::Command::new(&claude_bin);
        for arg in &claude_args {
            cmd.arg(arg);
        }

        let path_env = claude_stream::augmented_path();
        log::debug!("[session] PATH: {}", path_env);
        log::debug!(
            "[session] cwd: {}, prompt: {:?}",
            cwd,
            truncate_str(prompt, 80)
        );
        cmd.current_dir(cwd)
            .env("PATH", &path_env)
            .env_remove("CLAUDECODE")
            .env("CLAUDE_CODE_ENABLE_SDK_FILE_CHECKPOINTING", "1")
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped());

        // Pass API key to CLI when using API Key authentication mode (x-api-key header).
        // MUST remove AUTH_TOKEN to avoid inherited shell env vars taking priority.
        // Use env_remove (not empty string) — CLI may treat empty as "set but invalid".
        if let Some(key) = api_key {
            log::debug!("[session] setting ANTHROPIC_API_KEY env for local CLI");
            cmd.env("ANTHROPIC_API_KEY", key);
            cmd.env_remove("ANTHROPIC_AUTH_TOKEN");
        }

        // Pass auth token for third-party platforms using Bearer auth.
        // MUST remove API_KEY to avoid inherited shell env vars causing conflicts.
        if let Some(token) = auth_token {
            log::debug!("[session] setting ANTHROPIC_AUTH_TOKEN env for local CLI");
            cmd.env("ANTHROPIC_AUTH_TOKEN", token);
            cmd.env_remove("ANTHROPIC_API_KEY");
        }

        // Pass Base URL for third-party API endpoints
        if let Some(url) = base_url {
            log::debug!("[session] setting ANTHROPIC_BASE_URL={}", url);
            cmd.env("ANTHROPIC_BASE_URL", url);
        }

        // Pass model tier env vars for third-party platforms (low priority — --model flag overrides)
        if let Some(m) = models {
            for (k, v) in resolve_model_tiers(m) {
                cmd.env(k, v);
            }
        }

        // Pass extra env vars for third-party platforms (e.g. API_TIMEOUT_MS for DeepSeek)
        if let Some(extra) = extra_env {
            for (k, v) in extra {
                log::debug!("[session] setting extra env {}={}", k, v);
                cmd.env(k, v);
            }
        }

        cmd.hide_console().kill_on_drop(true).spawn().map_err(|e| {
            log::error!("[session] Failed to spawn claude: {}", e);
            format!("Failed to spawn claude: {}", e)
        })?
    };
    log::debug!("[session] child process spawned, pid={:?}", child.id());

    let stdin = child.stdin.take().ok_or("Failed to capture claude stdin")?;
    let stdout = child
        .stdout
        .take()
        .ok_or("Failed to capture claude stdout")?;
    let stderr = child
        .stderr
        .take()
        .ok_or("Failed to capture claude stderr")?;

    // Initial prompt is now sent via ActorCommand::SendMessage after actor spawn.
    // This ensures ALL user messages go through the Turn Transaction Engine.

    Ok((child, stdin, stdout, stderr))
}

// ── Side question (BTW) ──

/// Spawn a one-shot forked CLI process to answer a side question without
/// polluting the original session. Streams text deltas back via Tauri events.
#[tauri::command]
pub async fn side_question(
    app: tauri::AppHandle,
    run_id: String,
    question: String,
) -> Result<String, String> {
    use serde_json::Value;
    use tokio::io::{AsyncBufReadExt, BufReader};
    use tokio::process::Command;

    let btw_id = uuid::Uuid::new_v4().to_string();
    log::debug!(
        "[btw] side_question: run_id={}, btw_id={}, question={}",
        run_id,
        btw_id,
        truncate_str(&question, 80)
    );

    // 1. Read source run metadata
    let source =
        storage::runs::get_run(&run_id).ok_or_else(|| format!("Run {} not found", run_id))?;

    // Codex: ephemeral side question (branch before session_id — Codex has none)
    if source.agent == "codex" {
        return codex_side_question(app, &source, &question, &btw_id).await;
    }

    let session_id = source
        .session_id
        .clone()
        .ok_or_else(|| "No session_id available for side question".to_string())?;

    // 2. Resolve auth
    let user_settings = storage::settings::get_user_settings();
    let remote = resolve_remote_host(&source)?;
    let effective_pid = if user_settings.auth_mode == "cli" {
        None
    } else {
        source.platform_id.as_deref()
    };
    let resolved = resolve_auth_env_for_platform(&remote, &user_settings, effective_pid);
    let resolved = augment_with_shell_auth(
        resolved,
        &user_settings.auth_mode,
        remote.is_some(),
        &source.cwd,
    );

    // 3. Wrap question in system-reminder (matches CLI's side question prompt)
    let wrapped_question = format!(
        "<system-reminder>\nThe user is asking a side question. Answer it concisely. \
         This answer will NOT be added to the conversation history.\n</system-reminder>\n\n{}",
        question
    );

    // 4. Build CLI args
    let claude_bin = claude_stream::resolve_claude_path();
    let effective_cwd = source.remote_cwd.as_deref().unwrap_or(&source.cwd);

    let mut claude_args: Vec<String> = vec![
        "--resume".into(),
        session_id.clone(),
        "--fork-session".into(),
        "--no-session-persistence".into(),
        "-p".into(),
        wrapped_question,
        "--output-format".into(),
        "stream-json".into(),
        "--verbose".into(),
        "--max-turns".into(),
        "1".into(),
    ];

    // Add adapter flags (model overrides, etc.)
    let agent_settings = storage::settings::get_agent_settings(&source.agent);
    let mut adapter = adapter::build_adapter_settings(&agent_settings, &user_settings, None);
    adapter::clear_model_if_provider_overrides(
        &mut adapter,
        &None,
        &agent_settings.model,
        &resolved.models,
    );
    let flag_args = adapter::build_settings_args(&adapter, false);
    claude_args.extend(flag_args.iter().cloned());

    // 5. Spawn CLI process
    let mut cmd = if let Some(ref remote_host) = remote {
        let remote_cmd = crate::agent::ssh::build_remote_claude_command(
            remote_host,
            effective_cwd,
            &claude_args,
            resolved.api_key.as_deref(),
            resolved.auth_token.as_deref(),
            resolved.base_url.as_deref(),
            resolved.models.as_deref(),
            resolved.extra_env.as_ref(),
        );
        let mut ssh_cmd = crate::agent::ssh::build_ssh_command(remote_host, &remote_cmd);
        ssh_cmd
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())
            .kill_on_drop(true);
        ssh_cmd
    } else {
        let mut local_cmd = Command::new(&claude_bin);
        for arg in &claude_args {
            local_cmd.arg(arg);
        }
        let path_env = claude_stream::augmented_path();
        local_cmd
            .current_dir(effective_cwd)
            .env("PATH", &path_env)
            .env_remove("CLAUDECODE")
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped());
        // Auth env (mutually exclusive)
        if let Some(key) = &resolved.api_key {
            local_cmd.env("ANTHROPIC_API_KEY", key);
            local_cmd.env_remove("ANTHROPIC_AUTH_TOKEN");
        }
        if let Some(token) = &resolved.auth_token {
            local_cmd.env("ANTHROPIC_AUTH_TOKEN", token);
            local_cmd.env_remove("ANTHROPIC_API_KEY");
        }
        if let Some(url) = &resolved.base_url {
            local_cmd.env("ANTHROPIC_BASE_URL", url);
        }
        if let Some(m) = &resolved.models {
            for (k, v) in resolve_model_tiers(m) {
                local_cmd.env(k, v);
            }
        }
        if let Some(extra) = &resolved.extra_env {
            for (k, v) in extra {
                local_cmd.env(k, v);
            }
        }
        local_cmd
    };

    cmd.hide_console().kill_on_drop(true);

    let mut child = cmd
        .spawn()
        .map_err(|e| format!("Failed to spawn side question CLI: {}", e))?;

    let stdout = child
        .stdout
        .take()
        .ok_or("Failed to capture CLI stdout for side question")?;

    let stderr = child.stderr.take();

    // 6. Stream text deltas back via Tauri events
    let btw_id_clone = btw_id.clone();
    let app_clone = app.clone();
    tokio::spawn(async move {
        use tauri::Emitter;

        // Drain stderr in background for debugging
        if let Some(stderr) = stderr {
            let btw_id_err = btw_id_clone.clone();
            tokio::spawn(async move {
                let mut err_reader = BufReader::new(stderr).lines();
                while let Ok(Some(line)) = err_reader.next_line().await {
                    log::debug!("[btw] stderr ({}): {}", btw_id_err, line);
                }
            });
        }

        let mut reader = BufReader::new(stdout).lines();
        let mut got_content = false;
        while let Ok(Some(line)) = reader.next_line().await {
            log::trace!("[btw] stdout line: {}", &line[..line.len().min(200)]);
            if let Ok(obj) = serde_json::from_str::<Value>(&line) {
                // Unwrap stream_event envelope: CLI wraps API events as
                // {"type":"stream_event","event":{"type":"content_block_delta",...}}
                let event = if obj.get("type").and_then(|t| t.as_str()) == Some("stream_event") {
                    obj.get("event").cloned().unwrap_or(obj.clone())
                } else {
                    obj.clone()
                };

                let event_type = event.get("type").and_then(|t| t.as_str()).unwrap_or("");
                match event_type {
                    // Streaming text chunks (standard API streaming)
                    "content_block_delta" => {
                        if let Some(text) = event.pointer("/delta/text").and_then(|v| v.as_str()) {
                            got_content = true;
                            log::debug!("[btw] delta: {} chars", text.len());
                            let _ = app_clone.emit(
                                "btw-delta",
                                serde_json::json!({
                                    "btw_id": btw_id_clone,
                                    "text": text
                                }),
                            );
                        }
                    }
                    // Complete assistant message (CLI may batch text in -p mode)
                    "assistant" => {
                        let message = event.get("message").unwrap_or(&event);
                        if let Some(content) = message.get("content").and_then(|c| c.as_array()) {
                            for block in content {
                                if block.get("type").and_then(|t| t.as_str()) == Some("text") {
                                    if let Some(text) = block.get("text").and_then(|v| v.as_str()) {
                                        if !text.is_empty() {
                                            got_content = true;
                                            log::debug!(
                                                "[btw] assistant text block: {} chars",
                                                text.len()
                                            );
                                            let _ = app_clone.emit(
                                                "btw-delta",
                                                serde_json::json!({
                                                    "btw_id": btw_id_clone,
                                                    "text": text
                                                }),
                                            );
                                        }
                                    }
                                }
                            }
                        }
                    }
                    "result" => {
                        log::debug!(
                            "[btw] received result event, completing btw_id={}",
                            btw_id_clone
                        );
                        break;
                    }
                    "error" => {
                        let msg = event
                            .get("error")
                            .and_then(|e| e.as_str())
                            .unwrap_or("unknown error");
                        log::error!("[btw] CLI error: {}", msg);
                        let _ = app_clone.emit(
                            "btw-error",
                            serde_json::json!({
                                "btw_id": btw_id_clone,
                                "error": msg
                            }),
                        );
                        break;
                    }
                    other => {
                        log::debug!("[btw] event type: {}", other);
                    }
                }
            }
        }

        // If no content was received, the CLI likely failed — check exit status
        if !got_content {
            let status = child.wait().await;
            let code = status.as_ref().ok().and_then(|s| s.code());
            log::error!(
                "[btw] no content received, exit={:?}, btw_id={}",
                code,
                btw_id_clone
            );
            let _ = app_clone.emit(
                "btw-error",
                serde_json::json!({
                    "btw_id": btw_id_clone,
                    "error": format!("Side question failed (exit code: {:?})", code)
                }),
            );
        } else {
            // Emit completion
            let _ = app_clone.emit(
                "btw-complete",
                serde_json::json!({ "btw_id": btw_id_clone }),
            );
        }

        // Clean up child process
        let _ = child.kill().await;
        log::debug!(
            "[btw] side question process finished, btw_id={}",
            btw_id_clone
        );
    });

    log::debug!("[btw] spawned side question stream, btw_id={}", btw_id);
    Ok(btw_id)
}

// ── Codex side question (ephemeral, read-only) ──

/// Safely truncate a String to at most `max_chars` characters (UTF-8 safe).
fn safe_tail(s: &mut String, max_chars: usize) {
    let char_count = s.chars().count();
    if char_count > max_chars {
        let skip = char_count - max_chars;
        if let Some((idx, _)) = s.char_indices().nth(skip) {
            *s = s[idx..].to_string();
        }
    }
}

/// Extract assistant text only from a Codex NDJSON payload.
/// Unlike `extract_codex_delta()` this intentionally skips command_execution
/// output — side questions should only surface the assistant's answer.
fn extract_codex_btw_text(payload: &serde_json::Value) -> Option<String> {
    let type_str = payload.get("type").and_then(|v| v.as_str()).unwrap_or("");

    // Codex v0.98+: item.completed → item.type == "agent_message" → item.text
    if type_str == "item.completed" {
        if let Some(item) = payload.get("item") {
            let item_type = item.get("type").and_then(|v| v.as_str()).unwrap_or("");
            if item_type == "agent_message" {
                return item
                    .get("text")
                    .and_then(|v| v.as_str())
                    .filter(|s| !s.is_empty())
                    .map(|s| s.to_string());
            }
        }
    }

    // Direct delta field (older Codex versions)
    if type_str.contains("delta") {
        if let Some(delta) = payload.get("delta").and_then(|v| v.as_str()) {
            return Some(delta.to_string());
        }
        if let Some(text) = payload.get("text").and_then(|v| v.as_str()) {
            return Some(text.to_string());
        }
    }

    // output_text field (legacy)
    if let Some(text) = payload.get("output_text").and_then(|v| v.as_str()) {
        if !text.is_empty() {
            return Some(text.to_string());
        }
    }

    None
}

/// Decide which btw event to emit based on what was collected during the stream.
#[derive(Debug, PartialEq)]
enum BtwOutcome {
    /// Got content → emit btw-complete (optionally with a non-zero exit warning).
    Complete {
        exit_code: Option<i32>,
        exit_ok: bool,
    },
    /// No content → emit btw-error with this message.
    Error(String),
}

fn decide_btw_outcome(
    got_content: bool,
    error_msg: Option<String>,
    exit_code: Option<i32>,
    exit_ok: bool,
    stderr_tail: &str,
) -> BtwOutcome {
    if got_content {
        BtwOutcome::Complete { exit_code, exit_ok }
    } else if let Some(msg) = error_msg {
        BtwOutcome::Error(msg)
    } else if stderr_tail.is_empty() {
        BtwOutcome::Error(format!("Side question failed (exit code: {:?})", exit_code))
    } else {
        BtwOutcome::Error(format!(
            "Side question failed: {}",
            truncate_str(stderr_tail, 200)
        ))
    }
}

/// Codex ephemeral side question: spawn `codex exec --ephemeral --sandbox read-only --json`
/// with a timeout. Streams btw-delta/btw-complete/btw-error events to the frontend.
async fn codex_side_question(
    app: tauri::AppHandle,
    source: &RunMeta,
    question: &str,
    btw_id: &str,
) -> Result<String, String> {
    use crate::agent::claude_stream;
    use crate::process_ext::HideConsole;
    use serde_json::Value;
    use tokio::io::{AsyncBufReadExt, BufReader};
    use tokio::process::Command;

    let codex_bin = claude_stream::which_binary("codex")
        .ok_or_else(|| "Codex CLI not found in PATH".to_string())?;

    let wrapped_question = format!(
        "The user is asking a side question. Answer it concisely. \
         This answer will NOT be added to the conversation history.\n\n{}",
        question
    );

    let mut codex_args: Vec<String> = vec![
        "exec".into(),
        "--ephemeral".into(),
        "--sandbox".into(),
        "read-only".into(),
        "--json".into(),
        "--skip-git-repo-check".into(),
    ];

    // Resolve the configured Codex provider so the side question routes to the same gateway as
    // the main run (otherwise a custom/gateway provider would be ignored → wrong provider or an
    // auth error). Mirrors the normal run path: `-c model_providers.*` overrides + env_key=api_key.
    let user_settings = storage::settings::get_user_settings();
    let agent_settings = storage::settings::get_agent_settings(&source.agent);
    let adapter = adapter::build_adapter_settings(&agent_settings, &user_settings, None);
    let codex_provider = adapter.codex_provider.clone();
    if let Some(ref p) = codex_provider {
        codex_args.extend(crate::agent::spawn::codex_provider_config_args(p));
    }

    // Model: a provider-pinned model wins; otherwise inherit the source run's model (skipping
    // Claude model names that Codex rejects).
    let provider_model = codex_provider
        .as_ref()
        .map(|p| p.model.clone())
        .filter(|m| !m.is_empty());
    if let Some(m) = provider_model {
        codex_args.push("--model".into());
        codex_args.push(m);
    } else if let Some(ref m) = source.model {
        let lm = m.to_lowercase();
        let is_claude_model = lm.is_empty()
            || lm.contains("claude")
            || lm.contains("opus")
            || lm.contains("sonnet")
            || lm.contains("haiku");
        if !is_claude_model {
            codex_args.push("--model".into());
            codex_args.push(m.clone());
        }
    }

    // Prompt must be last arg
    codex_args.push(wrapped_question);

    let effective_cwd = &source.cwd;

    log::debug!(
        "[btw] codex_side_question: btw_id={}, cwd={}, args_len={}, provider={:?}",
        btw_id,
        effective_cwd,
        codex_args.len(),
        codex_provider.as_ref().map(|p| &p.id)
    );

    let mut cmd = Command::new(&codex_bin);
    for arg in &codex_args {
        cmd.arg(arg);
    }
    let path_env = claude_stream::augmented_path();
    cmd.current_dir(effective_cwd)
        .env("PATH", &path_env)
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .hide_console()
        .kill_on_drop(true);
    // Provider API key (env_key=api_key), same as the main run path.
    if let Some(ref p) = codex_provider {
        if let Some((k, v)) = crate::agent::spawn::codex_provider_env(p) {
            cmd.env(k, v);
        }
    }

    let mut child = cmd
        .spawn()
        .map_err(|e| format!("Failed to spawn Codex side question: {}", e))?;

    let stdout = child
        .stdout
        .take()
        .ok_or("Failed to capture Codex stdout for side question")?;
    let stderr = child.stderr.take();

    // Wrap child in Arc<Mutex> so the timeout branch can still kill explicitly
    let child = std::sync::Arc::new(tokio::sync::Mutex::new(child));

    let btw_id_owned = btw_id.to_string();
    let app_clone = app.clone();

    tokio::spawn(async move {
        use tauri::Emitter;

        // Drain stderr in background to a ring buffer
        let stderr_buf = std::sync::Arc::new(tokio::sync::Mutex::new(String::new()));
        if let Some(stderr) = stderr {
            let buf = stderr_buf.clone();
            let btw_id_err = btw_id_owned.clone();
            tokio::spawn(async move {
                let mut err_reader = BufReader::new(stderr).lines();
                while let Ok(Some(line)) = err_reader.next_line().await {
                    log::debug!("[btw] codex stderr ({}): {}", btw_id_err, line);
                    let mut b = buf.lock().await;
                    if !b.is_empty() {
                        b.push('\n');
                    }
                    b.push_str(&line);
                    safe_tail(&mut b, 500);
                }
            });
        }

        // Only move stdout reader into the timeout — child stays accessible
        let child_inner = child.clone();
        let btw_id_inner = btw_id_owned.clone();
        let app_inner = app_clone.clone();
        let read_result = tokio::time::timeout(std::time::Duration::from_secs(120), async move {
            let mut reader = BufReader::new(stdout).lines();
            let mut got_content = false;
            let mut error_msg: Option<String> = None;

            while let Ok(Some(line)) = reader.next_line().await {
                log::trace!("[btw] codex stdout: {}", truncate_str(&line, 200));
                if let Ok(obj) = serde_json::from_str::<Value>(&line) {
                    let event_type = obj.get("type").and_then(|t| t.as_str()).unwrap_or("");

                    // BTW-specific: only agent_message text, not command_execution
                    if let Some(text) = extract_codex_btw_text(&obj) {
                        got_content = true;
                        let _ = app_inner.emit(
                            "btw-delta",
                            serde_json::json!({
                                "btw_id": &btw_id_inner,
                                "text": text
                            }),
                        );
                    } else if event_type == "error" {
                        let msg = obj
                            .get("message")
                            .or_else(|| obj.get("error"))
                            .and_then(|e| e.as_str())
                            .unwrap_or("unknown error");
                        error_msg = Some(msg.to_string());
                        log::error!("[btw] codex error event: {}", msg);
                    } else {
                        log::debug!("[btw] codex event type: {}", event_type);
                    }
                }
            }

            // stdout EOF → wait for process exit
            let status = child_inner.lock().await.wait().await;
            (got_content, error_msg, status)
        })
        .await;

        match read_result {
            Ok((got_content, error_msg, status)) => {
                let exit_ok = status.as_ref().ok().is_some_and(|s| s.success());
                let exit_code = status.as_ref().ok().and_then(|s| s.code());
                let stderr_tail = stderr_buf.lock().await;

                match decide_btw_outcome(got_content, error_msg, exit_code, exit_ok, &stderr_tail) {
                    BtwOutcome::Complete { exit_code, exit_ok } => {
                        if !exit_ok {
                            log::warn!(
                                "[btw] non-zero exit with content, code={:?}, stderr_tail={}",
                                exit_code,
                                truncate_str(&stderr_tail, 200)
                            );
                        }
                        let _ = app_clone.emit(
                            "btw-complete",
                            serde_json::json!({ "btw_id": &btw_id_owned }),
                        );
                    }
                    BtwOutcome::Error(err) => {
                        log::error!("[btw] no content, exit={:?}, error={}", exit_code, err);
                        let _ = app_clone.emit(
                            "btw-error",
                            serde_json::json!({
                                "btw_id": &btw_id_owned,
                                "error": err
                            }),
                        );
                    }
                }
            }
            Err(_timeout) => {
                log::error!(
                    "[btw] codex side question timed out, btw_id={}",
                    btw_id_owned
                );
                // Explicit kill + wait (child not consumed by timeout future)
                let mut ch = child.lock().await;
                let _ = ch.kill().await;
                let _ = ch.wait().await;
                let _ = app_clone.emit(
                    "btw-error",
                    serde_json::json!({
                        "btw_id": &btw_id_owned,
                        "error": "Side question timed out (120s)"
                    }),
                );
            }
        }

        log::debug!(
            "[btw] codex side question finished, btw_id={}",
            btw_id_owned
        );
    });

    log::debug!("[btw] spawned codex side question, btw_id={}", btw_id);
    Ok(btw_id.to_string())
}

// ── Ralph Loop commands ──

#[tauri::command]
pub async fn start_ralph_loop(
    sessions: State<'_, ActorSessionMap>,
    run_id: String,
    prompt: String,
    max_iterations: u32,
    completion_promise: Option<String>,
) -> Result<(), String> {
    log::debug!(
        "[session] start_ralph_loop: run_id={}, prompt_len={}, max_iterations={}, promise={:?}",
        run_id,
        prompt.len(),
        max_iterations,
        completion_promise
    );

    let cmd_tx = get_cmd_tx(&sessions, &run_id).await?;

    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
    cmd_tx
        .send(ActorCommand::StartRalphLoop {
            prompt,
            max_iterations,
            completion_promise,
            reply: reply_tx,
        })
        .await
        .map_err(|_| "Actor dead".to_string())?;

    reply_rx
        .await
        .map_err(|_| "Actor dropped reply".to_string())?
}

#[tauri::command]
pub async fn cancel_ralph_loop(
    sessions: State<'_, ActorSessionMap>,
    run_id: String,
) -> Result<RalphCancelResult, String> {
    log::debug!("[session] cancel_ralph_loop: run_id={}", run_id);

    let cmd_tx = get_cmd_tx(&sessions, &run_id).await?;

    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
    cmd_tx
        .send(ActorCommand::CancelRalphLoop { reply: reply_tx })
        .await
        .map_err(|_| "Actor dead".to_string())?;

    reply_rx
        .await
        .map_err(|_| "Actor dropped reply".to_string())?
}

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

    fn default_user_settings() -> UserSettings {
        UserSettings {
            auth_mode: "api".to_string(),
            ..Default::default()
        }
    }

    fn make_cred(
        pid: &str,
        key: Option<&str>,
        base_url: Option<&str>,
        auth_env_var: Option<&str>,
    ) -> PlatformCredential {
        PlatformCredential {
            platform_id: pid.to_string(),
            api_key: key.map(|s| s.to_string()),
            base_url: base_url.map(|s| s.to_string()),
            auth_env_var: auth_env_var.map(|s| s.to_string()),
            name: None,
            models: None,
            extra_env: None,
        }
    }

    #[test]
    fn key_optional_no_credential_uses_defaults() {
        let settings = default_user_settings();
        let resolved = resolve_auth_env_for_platform(&None, &settings, Some("ccswitch"));

        assert_eq!(resolved.auth_token.as_deref(), Some("PROXY_MANAGED"));
        assert!(resolved.api_key.is_none());
        assert_eq!(resolved.base_url.as_deref(), Some("http://127.0.0.1:15721"));
    }

    #[test]
    fn key_optional_credential_empty_key_with_base_url() {
        let mut settings = default_user_settings();
        settings.platform_credentials.push(make_cred(
            "ccswitch",
            None,
            Some("http://custom:15721"),
            Some("ANTHROPIC_AUTH_TOKEN"),
        ));

        let resolved = resolve_auth_env_for_platform(&None, &settings, Some("ccswitch"));

        assert_eq!(resolved.auth_token.as_deref(), Some("PROXY_MANAGED"));
        assert_eq!(resolved.base_url.as_deref(), Some("http://custom:15721"));
    }

    #[test]
    fn key_optional_credential_has_key_uses_key() {
        let mut settings = default_user_settings();
        settings.platform_credentials.push(make_cred(
            "ccswitch",
            Some("real-key-123"),
            Some("http://127.0.0.1:15721"),
            Some("ANTHROPIC_AUTH_TOKEN"),
        ));

        let resolved = resolve_auth_env_for_platform(&None, &settings, Some("ccswitch"));

        assert_eq!(resolved.auth_token.as_deref(), Some("real-key-123"));
        assert!(resolved.api_key.is_none());
    }

    #[test]
    fn non_key_optional_empty_key_falls_back_global() {
        let mut settings = default_user_settings();
        settings.anthropic_api_key = Some("global-key".to_string());
        settings.platform_credentials.push(make_cred(
            "deepseek",
            None,
            Some("https://api.deepseek.com/anthropic"),
            None,
        ));

        let resolved = resolve_auth_env_for_platform(&None, &settings, Some("deepseek"));

        assert_eq!(resolved.api_key.as_deref(), Some("global-key"));
        assert!(resolved.auth_token.is_none());
    }

    #[test]
    fn unknown_platform_no_credential_falls_back_global() {
        let mut settings = default_user_settings();
        settings.anthropic_api_key = Some("global-key".to_string());

        let resolved =
            resolve_auth_env_for_platform(&None, &settings, Some("unknown-platform-xyz"));

        assert_eq!(resolved.api_key.as_deref(), Some("global-key"));
    }

    #[test]
    fn key_optional_missing_auth_env_var_uses_defaults() {
        let mut settings = default_user_settings();
        settings.platform_credentials.push(make_cred(
            "ccswitch",
            None,
            Some("http://127.0.0.1:15721"),
            None, // auth_env_var missing
        ));

        let resolved = resolve_auth_env_for_platform(&None, &settings, Some("ccswitch"));

        assert_eq!(resolved.auth_token.as_deref(), Some("PROXY_MANAGED"));
        assert!(resolved.api_key.is_none());
    }

    #[test]
    fn key_optional_wrong_auth_env_var_overridden_by_defaults() {
        let mut settings = default_user_settings();
        settings.platform_credentials.push(make_cred(
            "ccswitch",
            None,
            Some("http://127.0.0.1:15721"),
            Some("ANTHROPIC_API_KEY"), // wrong — defaults should override
        ));

        let resolved = resolve_auth_env_for_platform(&None, &settings, Some("ccswitch"));

        assert_eq!(resolved.auth_token.as_deref(), Some("PROXY_MANAGED"));
        assert!(resolved.api_key.is_none());
    }

    #[test]
    fn ccr_no_credential_includes_default_model() {
        let settings = default_user_settings();
        let resolved = resolve_auth_env_for_platform(&None, &settings, Some("ccr"));

        assert_eq!(resolved.auth_token.as_deref(), Some("PROXY_MANAGED"));
        assert_eq!(resolved.base_url.as_deref(), Some("http://127.0.0.1:3456"));
        assert_eq!(
            resolved.models.as_deref(),
            Some(vec!["claude-sonnet-4-6".to_string()].as_slice())
        );
    }

    // ── is_local_url tests ──

    #[test]
    fn is_local_url_loopback_variants() {
        assert!(is_local_url("http://127.0.0.1:15721"));
        assert!(is_local_url("http://127.0.99.1:8080"));
        assert!(is_local_url("http://localhost:11434"));
        assert!(is_local_url("http://[::1]:8080"));
        assert!(is_local_url("http://0.0.0.0:3000"));
    }

    #[test]
    fn is_local_url_remote_not_matched() {
        assert!(!is_local_url("https://api.deepseek.com"));
        assert!(!is_local_url("https://127.example.com"));
        assert!(!is_local_url("https://example.com/path?host=127.0.0.1"));
        assert!(!is_local_url("not-a-url"));
    }

    // ── preflight_check_base_url tests ──

    #[tokio::test]
    async fn preflight_none_url_skips() {
        let timeout = tokio::time::timeout(std::time::Duration::from_secs(5), async {
            let result = preflight_check_base_url(None, None).await;
            assert!(result.is_ok());
        });
        timeout.await.expect("test timed out");
    }

    #[tokio::test]
    async fn preflight_unreachable_returns_error() {
        let timeout = tokio::time::timeout(std::time::Duration::from_secs(5), async {
            // RFC 5737 TEST-NET — guaranteed non-routable
            let result =
                preflight_check_base_url(Some("http://192.0.2.1:1"), Some("ccswitch")).await;
            assert!(result.is_err());
            let err = result.unwrap_err();
            assert!(err.contains("unreachable"), "error: {}", err);
            assert!(err.contains("CC Switch"), "error: {}", err);
        });
        timeout.await.expect("test timed out");
    }

    #[tokio::test]
    async fn preflight_reachable_200_is_ok() {
        use tokio::io::AsyncWriteExt;
        let timeout = tokio::time::timeout(std::time::Duration::from_secs(5), async {
            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let port = listener.local_addr().unwrap().port();
            tokio::spawn(async move {
                if let Ok((mut stream, _)) = listener.accept().await {
                    let mut buf = [0u8; 1024];
                    let _ = tokio::io::AsyncReadExt::read(&mut stream, &mut buf).await;
                    let resp = "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n[]";
                    let _ = stream.write_all(resp.as_bytes()).await;
                }
            });
            let url = format!("http://127.0.0.1:{}", port);
            let result = preflight_check_base_url(Some(&url), Some("ccswitch")).await;
            assert!(result.is_ok(), "expected Ok, got: {:?}", result);
        });
        timeout.await.expect("test timed out");
    }

    #[tokio::test]
    async fn preflight_reachable_401_is_ok() {
        use tokio::io::AsyncWriteExt;
        let timeout = tokio::time::timeout(std::time::Duration::from_secs(5), async {
            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let port = listener.local_addr().unwrap().port();
            tokio::spawn(async move {
                if let Ok((mut stream, _)) = listener.accept().await {
                    let mut buf = [0u8; 1024];
                    let _ = tokio::io::AsyncReadExt::read(&mut stream, &mut buf).await;
                    let resp = "HTTP/1.1 401 Unauthorized\r\nContent-Length: 0\r\n\r\n";
                    let _ = stream.write_all(resp.as_bytes()).await;
                }
            });
            let url = format!("http://127.0.0.1:{}", port);
            let result = preflight_check_base_url(Some(&url), Some("deepseek")).await;
            assert!(result.is_ok(), "401 should be treated as reachable");
        });
        timeout.await.expect("test timed out");
    }

    #[tokio::test]
    async fn preflight_reachable_405_is_ok() {
        use tokio::io::AsyncWriteExt;
        let timeout = tokio::time::timeout(std::time::Duration::from_secs(5), async {
            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let port = listener.local_addr().unwrap().port();
            tokio::spawn(async move {
                if let Ok((mut stream, _)) = listener.accept().await {
                    let mut buf = [0u8; 1024];
                    let _ = tokio::io::AsyncReadExt::read(&mut stream, &mut buf).await;
                    let resp = "HTTP/1.1 405 Method Not Allowed\r\nContent-Length: 0\r\n\r\n";
                    let _ = stream.write_all(resp.as_bytes()).await;
                }
            });
            let url = format!("http://127.0.0.1:{}", port);
            let result = preflight_check_base_url(Some(&url), Some("ollama")).await;
            assert!(result.is_ok(), "405 should be treated as reachable");
        });
        timeout.await.expect("test timed out");
    }

    // ── augment_with_shell_auth tests ──

    fn empty_resolved() -> ResolvedAuth {
        ResolvedAuth {
            api_key: None,
            auth_token: None,
            base_url: None,
            models: None,
            extra_env: None,
        }
    }

    // ── Guard tests (pure logic, no filesystem dependency) ──

    #[test]
    fn augment_remote_never_injects() {
        let r = augment_with_shell_auth(empty_resolved(), "cli", true, "/tmp");
        assert!(r.api_key.is_none() && r.auth_token.is_none());
    }

    #[test]
    fn augment_api_mode_never_injects() {
        let r = augment_with_shell_auth(empty_resolved(), "api", false, "/tmp");
        assert!(r.api_key.is_none() && r.auth_token.is_none());
    }

    #[test]
    fn augment_preserves_existing_key() {
        let existing = ResolvedAuth {
            api_key: Some("k".into()),
            ..empty_resolved()
        };
        let r = augment_with_shell_auth(existing, "cli", false, "/tmp");
        assert_eq!(r.api_key.as_deref(), Some("k"));
    }

    #[test]
    fn augment_preserves_existing_token() {
        let existing = ResolvedAuth {
            auth_token: Some("t".into()),
            ..empty_resolved()
        };
        let r = augment_with_shell_auth(existing, "cli", false, "/tmp");
        assert_eq!(r.auth_token.as_deref(), Some("t"));
    }

    // ── should_skip_env_injection tests (pure function, zero env dependency) ──

    #[test]
    fn skip_env_injection_when_key_present() {
        assert!(should_skip_env_injection(Some("sk-123"), None));
    }

    #[test]
    fn skip_env_injection_when_token_present() {
        assert!(should_skip_env_injection(None, Some("oauth-token")));
    }

    #[test]
    fn skip_env_injection_when_both_present() {
        assert!(should_skip_env_injection(
            Some("sk-123"),
            Some("oauth-token")
        ));
    }

    #[test]
    fn no_skip_when_both_none() {
        assert!(!should_skip_env_injection(None, None));
    }

    #[test]
    fn no_skip_when_whitespace_only() {
        assert!(!should_skip_env_injection(Some("  "), Some("")));
    }

    // ── config_value_has_auth_key tests (pure function, zero filesystem dependency) ──

    #[test]
    fn config_value_detects_api_key() {
        let config = serde_json::json!({"apiKey": "sk-ant-123"});
        assert!(config_value_has_auth_key(&config));
    }

    #[test]
    fn config_value_detects_primary_api_key() {
        let config = serde_json::json!({"primaryApiKey": "pk-team-456"});
        assert!(config_value_has_auth_key(&config));
    }

    #[test]
    fn config_value_empty_config_returns_false() {
        let config = serde_json::json!({});
        assert!(!config_value_has_auth_key(&config));
    }

    #[test]
    fn config_value_whitespace_only_key_returns_false() {
        let config = serde_json::json!({"apiKey": "  ", "primaryApiKey": ""});
        assert!(!config_value_has_auth_key(&config));
    }

    // ── Integration: project config loading + auth key detection ──
    // Tests load_project_cli_config → config_value_has_auth_key pipeline directly,
    // bypassing cli_config_has_auth_key to avoid false-positive from user-level config
    // on dev machines that have a real ~/.claude/settings.json with apiKey.

    #[test]
    fn project_config_with_api_key_detected() {
        let tmp = tempfile::tempdir().unwrap();
        let claude_dir = tmp.path().join(".claude");
        std::fs::create_dir_all(&claude_dir).unwrap();
        std::fs::write(claude_dir.join("settings.json"), r#"{"apiKey":"proj-key"}"#).unwrap();

        let config =
            crate::storage::cli_config::load_project_cli_config(tmp.path().to_str().unwrap());
        assert!(config_value_has_auth_key(&config));
    }

    #[test]
    fn project_config_without_api_key_not_detected() {
        let tmp = tempfile::tempdir().unwrap();
        let claude_dir = tmp.path().join(".claude");
        std::fs::create_dir_all(&claude_dir).unwrap();
        std::fs::write(claude_dir.join("settings.json"), r#"{"model":"sonnet"}"#).unwrap();

        let config =
            crate::storage::cli_config::load_project_cli_config(tmp.path().to_str().unwrap());
        assert!(!config_value_has_auth_key(&config));
    }

    #[test]
    fn project_config_missing_dir_not_detected() {
        let tmp = tempfile::tempdir().unwrap();
        // No .claude/ dir at all
        let config =
            crate::storage::cli_config::load_project_cli_config(tmp.path().to_str().unwrap());
        assert!(!config_value_has_auth_key(&config));
    }

    // ── resolve_model_tiers tests ──

    fn tier_env(result: &[(&str, String)], key: &str) -> String {
        result
            .iter()
            .find(|(k, _)| *k == key)
            .map(|(_, v)| v.clone())
            .unwrap_or_default()
    }

    #[test]
    fn model_tiers_empty_returns_nothing() {
        let r = resolve_model_tiers(&[]);
        assert!(r.is_empty());
    }

    #[test]
    fn model_tiers_single_all_same() {
        let r = resolve_model_tiers(&["m".into()]);
        assert_eq!(tier_env(&r, "ANTHROPIC_MODEL"), "m");
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_OPUS_MODEL"), "m");
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_HAIKU_MODEL"), "m");
    }

    #[test]
    fn model_tiers_two_main_and_haiku() {
        let r = resolve_model_tiers(&["main".into(), "eco".into()]);
        assert_eq!(tier_env(&r, "ANTHROPIC_MODEL"), "main");
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_OPUS_MODEL"), "main");
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_SONNET_MODEL"), "main");
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_HAIKU_MODEL"), "eco");
    }

    #[test]
    fn model_tiers_three_independent() {
        let r = resolve_model_tiers(&["o".into(), "s".into(), "h".into()]);
        assert_eq!(tier_env(&r, "ANTHROPIC_MODEL"), "s");
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_OPUS_MODEL"), "o");
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_SONNET_MODEL"), "s");
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_HAIKU_MODEL"), "h");
    }

    #[test]
    fn model_tiers_three_only_sonnet() {
        // ["", "s", ""] → all tiers = s
        let r = resolve_model_tiers(&["".into(), "s".into(), "".into()]);
        assert_eq!(tier_env(&r, "ANTHROPIC_MODEL"), "s");
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_OPUS_MODEL"), "s");
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_HAIKU_MODEL"), "s");
    }

    #[test]
    fn model_tiers_three_sonnet_and_haiku() {
        // ["", "s", "h"] → Opus inherits Sonnet
        let r = resolve_model_tiers(&["".into(), "s".into(), "h".into()]);
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_OPUS_MODEL"), "s");
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_SONNET_MODEL"), "s");
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_HAIKU_MODEL"), "h");
    }

    #[test]
    fn model_tiers_three_opus_sonnet_empty_haiku() {
        // ["o", "s", ""] → Haiku inherits Sonnet
        let r = resolve_model_tiers(&["o".into(), "s".into(), "".into()]);
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_OPUS_MODEL"), "o");
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_SONNET_MODEL"), "s");
        assert_eq!(tier_env(&r, "ANTHROPIC_DEFAULT_HAIKU_MODEL"), "s");
    }

    #[test]
    fn model_tiers_three_all_empty_returns_nothing() {
        // ["", "", ""] → Sonnet empty → no injection
        let r = resolve_model_tiers(&["".into(), "".into(), "".into()]);
        assert!(r.is_empty());
    }

    #[test]
    fn model_tiers_three_sonnet_empty_with_others_returns_nothing() {
        // ["o", "", "h"] → Sonnet empty → no injection
        let r = resolve_model_tiers(&["o".into(), "".into(), "h".into()]);
        assert!(r.is_empty());
    }

    #[test]
    fn model_tiers_two_empty_first_returns_nothing() {
        // ["", ""] → first element empty → no injection (existing behavior)
        let r = resolve_model_tiers(&["".into(), "".into()]);
        // 2-element branch uses [0] for opus/sonnet — empty string still produces envs
        // This is existing behavior; the empty-string guard only applies to 3+ elements
        assert_eq!(r.len(), 4);
    }

    // ── Codex BTW helper tests ──

    #[test]
    fn btw_extract_agent_message() {
        let payload = serde_json::json!({
            "type": "item.completed",
            "item": {"type": "agent_message", "text": "Hello from btw"}
        });
        assert_eq!(
            extract_codex_btw_text(&payload),
            Some("Hello from btw".to_string())
        );
    }

    #[test]
    fn btw_extract_skips_command_execution() {
        let payload = serde_json::json!({
            "type": "item.completed",
            "item": {"type": "command_execution", "command": "ls", "aggregated_output": "file.txt"}
        });
        assert_eq!(extract_codex_btw_text(&payload), None);
    }

    #[test]
    fn btw_extract_agent_message_empty_text() {
        let payload = serde_json::json!({
            "type": "item.completed",
            "item": {"type": "agent_message", "text": ""}
        });
        assert_eq!(extract_codex_btw_text(&payload), None);
    }

    #[test]
    fn btw_extract_delta_fallback() {
        let payload = serde_json::json!({
            "type": "response.output_text.delta",
            "delta": "streaming chunk"
        });
        assert_eq!(
            extract_codex_btw_text(&payload),
            Some("streaming chunk".to_string())
        );
    }

    #[test]
    fn btw_extract_unrelated_event() {
        let payload = serde_json::json!({"type": "turn.started"});
        assert_eq!(extract_codex_btw_text(&payload), None);
    }

    #[test]
    fn btw_outcome_content_ok() {
        assert_eq!(
            decide_btw_outcome(true, None, Some(0), true, ""),
            BtwOutcome::Complete {
                exit_code: Some(0),
                exit_ok: true
            }
        );
    }

    #[test]
    fn btw_outcome_content_nonzero_exit() {
        // Got content + non-zero exit → still Complete (warn logged separately)
        assert_eq!(
            decide_btw_outcome(true, None, Some(1), false, "some stderr"),
            BtwOutcome::Complete {
                exit_code: Some(1),
                exit_ok: false
            }
        );
    }

    #[test]
    fn btw_outcome_no_content_error_msg() {
        assert_eq!(
            decide_btw_outcome(false, Some("rate limited".into()), Some(1), false, ""),
            BtwOutcome::Error("rate limited".to_string())
        );
    }

    #[test]
    fn btw_outcome_no_content_no_error_msg_stderr() {
        let outcome = decide_btw_outcome(false, None, Some(1), false, "oops\npanic");
        match outcome {
            BtwOutcome::Error(msg) => assert!(msg.contains("oops")),
            _ => panic!("expected Error"),
        }
    }

    #[test]
    fn btw_outcome_no_content_no_error_no_stderr() {
        let outcome = decide_btw_outcome(false, None, Some(42), false, "");
        match outcome {
            BtwOutcome::Error(msg) => assert!(msg.contains("42")),
            _ => panic!("expected Error"),
        }
    }

    #[test]
    fn safe_tail_short_string_unchanged() {
        let mut s = "hello".to_string();
        safe_tail(&mut s, 10);
        assert_eq!(s, "hello");
    }

    #[test]
    fn safe_tail_truncates_to_char_boundary() {
        // 3 CJK chars = 9 bytes, each is 1 char
        let mut s = "你好世界额外文本".to_string(); // 8 chars
        safe_tail(&mut s, 3);
        assert_eq!(s, "外文本");
    }

    #[test]
    fn safe_tail_multi_byte_emoji() {
        let mut s = "🎉🎊🎈🎁🎂".to_string(); // 5 chars
        safe_tail(&mut s, 2);
        assert_eq!(s, "🎁🎂");
    }
}
