use std::sync::Arc;

use claudette::agent::{
    self, AgentEvent, AgentSettings, ControlRequestInner, InnerStreamEvent, PersistentSession,
    StreamEvent,
};
use claudette::chat::{
    BuildAssistantArgs, CheckpointArgs, build_assistant_chat_message, create_turn_checkpoint,
    extract_assistant_text, extract_event_thinking,
};
use claudette::db::Database;
use claudette::model::{ChatMessage, ChatRole};
use portable_pty::{CommandBuilder, PtySize, native_pty_system};
use serde_json::json;

use crate::ws::{
    AgentSessionState, PendingPermission, PtyHandle, QueuedMessage, ServerState, Writer,
    send_message,
};

use claudette::permissions::tools_for_level;

/// Dispatch a JSON-RPC request and return a JSON-RPC response.
pub async fn handle_request(
    state: &Arc<ServerState>,
    writer: &Arc<Writer>,
    request: &serde_json::Value,
) -> serde_json::Value {
    let id = request
        .get("id")
        .cloned()
        .unwrap_or(serde_json::Value::Null);
    let method = request.get("method").and_then(|m| m.as_str()).unwrap_or("");
    let params = request.get("params").cloned().unwrap_or_default();

    let result = match method {
        "load_initial_data" => handle_load_initial_data(state).await,
        "load_chat_history" => {
            let chat_session_id = param_chat_session_id(&params);
            handle_load_chat_history(state, &chat_session_id).await
        }
        "load_completed_turns" => {
            let chat_session_id = param_chat_session_id(&params);
            handle_load_completed_turns(state, &chat_session_id)
        }
        "get_chat_snapshot" => {
            let chat_session_id = param_chat_session_id(&params);
            let limit = crate::snapshot::clamp_limit(params.get("limit").and_then(|v| v.as_i64()));
            let before = params
                .get("before_message_id")
                .or_else(|| params.get("beforeMessageId"))
                .and_then(|v| v.as_str())
                .map(String::from);
            match crate::snapshot::build(state, &chat_session_id, limit, before.as_deref()).await {
                Ok(snapshot) => serde_json::to_value(snapshot).map_err(|e| e.to_string()),
                Err(e) => Err(e),
            }
        }
        "send_chat_message" => {
            let chat_session_id = param_chat_session_id(&params);
            let content = param_str(&params, "content");
            let permission_level = params
                .get("permission_level")
                .and_then(|v| v.as_str())
                .map(String::from);
            let model = params
                .get("model")
                .and_then(|v| v.as_str())
                .map(String::from);
            let fast_mode = params.get("fast_mode").and_then(|v| v.as_bool());
            let thinking_enabled = params.get("thinking_enabled").and_then(|v| v.as_bool());
            let plan_mode = params.get("plan_mode").and_then(|v| v.as_bool());
            let effort = params
                .get("effort")
                .and_then(|v| v.as_str())
                .map(String::from);
            let chrome_enabled = params.get("chrome_enabled").and_then(|v| v.as_bool());
            let disable_1m_context = params.get("disable_1m_context").and_then(|v| v.as_bool());
            let backend_id = params
                .get("backend_id")
                .and_then(|v| v.as_str())
                .map(String::from);
            let mentioned_files: Option<Vec<String>> = params
                .get("mentioned_files")
                .and_then(|v| serde_json::from_value(v.clone()).ok());
            handle_send_chat_message(
                state,
                writer,
                &chat_session_id,
                &content,
                permission_level.as_deref(),
                model,
                fast_mode,
                thinking_enabled,
                plan_mode,
                effort,
                chrome_enabled,
                disable_1m_context,
                backend_id,
                mentioned_files,
            )
            .await
        }
        "steer_queued_chat_message" => {
            let chat_session_id = param_chat_session_id(&params);
            let content = param_str(&params, "content");
            handle_steer_queued_chat_message(state, &chat_session_id, &content).await
        }
        "submit_agent_answer" => {
            let chat_session_id = param_chat_session_id(&params);
            let tool_use_id = param_str(&params, "tool_use_id");
            let answers: std::collections::HashMap<String, String> = params
                .get("answers")
                .and_then(|v| serde_json::from_value(v.clone()).ok())
                .unwrap_or_default();
            let annotations = params.get("annotations").cloned();
            handle_submit_agent_answer(state, &chat_session_id, &tool_use_id, answers, annotations)
                .await
        }
        "submit_plan_approval" => {
            let chat_session_id = param_chat_session_id(&params);
            let tool_use_id = param_str(&params, "tool_use_id");
            let approved = params
                .get("approved")
                .and_then(|v| v.as_bool())
                .unwrap_or(false);
            let reason = params
                .get("reason")
                .and_then(|v| v.as_str())
                .map(String::from);
            handle_submit_approval(
                state,
                &chat_session_id,
                &tool_use_id,
                approved,
                reason,
                ApprovalKind::PlanModeOnly,
            )
            .await
        }
        "submit_agent_approval" => {
            let chat_session_id = param_chat_session_id(&params);
            let tool_use_id = param_str(&params, "tool_use_id");
            let approved = params
                .get("approved")
                .and_then(|v| v.as_bool())
                .unwrap_or(false);
            let reason = params
                .get("reason")
                .and_then(|v| v.as_str())
                .map(String::from);
            handle_submit_approval(
                state,
                &chat_session_id,
                &tool_use_id,
                approved,
                reason,
                ApprovalKind::AnyApprovalTool,
            )
            .await
        }
        "stop_agent" => {
            let chat_session_id = param_chat_session_id(&params);
            handle_stop_agent(state, &chat_session_id).await
        }
        "reset_agent_session" => {
            let chat_session_id = param_chat_session_id(&params);
            let mut agents = state.agents.write().await;
            agents.remove(&chat_session_id);
            Ok(json!(null))
        }
        "list_repositories" => {
            let db = open_db(state).map_err(|e| e.to_string());
            match db {
                Ok(db) => db
                    .list_repositories()
                    .map(|repos| serde_json::to_value(repos).unwrap_or_default())
                    .map_err(|e| e.to_string()),
                Err(e) => Err(e),
            }
        }
        "list_workspaces" => {
            let db = open_db(state).map_err(|e| e.to_string());
            match db {
                Ok(db) => db
                    .list_workspaces()
                    .map(|ws| serde_json::to_value(ws).unwrap_or_default())
                    .map_err(|e| e.to_string()),
                Err(e) => Err(e),
            }
        }
        "create_workspace" => {
            let repository_id = param_str(&params, "repository_id");
            let name = param_str(&params, "name");
            let preserve_supplied_name = params
                .get("preserve_name")
                .and_then(|v| v.as_bool())
                .unwrap_or(false);
            handle_create_workspace(state, &repository_id, &name, preserve_supplied_name).await
        }
        "archive_workspace" => {
            let workspace_id = param_str(&params, "workspace_id");
            handle_archive_workspace(state, &workspace_id).await
        }
        "restore_workspace" => {
            let workspace_id = param_str(&params, "workspace_id");
            handle_restore_workspace(state, &workspace_id).await
        }
        "load_diff_files" => {
            let workspace_id = param_str(&params, "workspace_id");
            handle_load_diff_files(state, &workspace_id).await
        }
        "load_file_diff" => {
            let worktree_path = param_str(&params, "worktree_path");
            let file_path = param_str(&params, "file_path");
            let merge_base = param_str(&params, "merge_base");
            let diff_layer = params
                .get("diff_layer")
                .and_then(|v| v.as_str())
                .map(String::from);
            handle_load_file_diff(
                &worktree_path,
                &file_path,
                &merge_base,
                diff_layer.as_deref(),
            )
            .await
        }
        "spawn_pty" => {
            let workspace_id = param_str(&params, "workspace_id");
            let cwd = param_str(&params, "cwd");
            let rows = params.get("rows").and_then(|v| v.as_u64()).unwrap_or(24) as u16;
            let cols = params.get("cols").and_then(|v| v.as_u64()).unwrap_or(80) as u16;
            handle_spawn_pty(state, writer, &workspace_id, &cwd, rows, cols).await
        }
        "write_pty" => {
            let pty_id = params.get("pty_id").and_then(|v| v.as_u64()).unwrap_or(0);
            let data: Vec<u8> = params
                .get("data")
                .and_then(|v| v.as_array())
                .map(|arr| {
                    arr.iter()
                        .filter_map(|b| b.as_u64().map(|n| n as u8))
                        .collect()
                })
                .unwrap_or_default();
            handle_write_pty(state, pty_id, &data).await
        }
        "resize_pty" => {
            let pty_id = params.get("pty_id").and_then(|v| v.as_u64()).unwrap_or(0);
            let rows = params.get("rows").and_then(|v| v.as_u64()).unwrap_or(24) as u16;
            let cols = params.get("cols").and_then(|v| v.as_u64()).unwrap_or(80) as u16;
            handle_resize_pty(state, pty_id, rows, cols).await
        }
        "close_pty" => {
            let pty_id = params.get("pty_id").and_then(|v| v.as_u64()).unwrap_or(0);
            handle_close_pty(state, pty_id).await
        }
        "get_app_setting" => {
            let key = param_str(&params, "key");
            handle_get_app_setting(state, &key)
        }
        "set_app_setting" => {
            let key = param_str(&params, "key");
            let value = param_str(&params, "value");
            handle_set_app_setting(state, &key, &value).await
        }
        "list_chat_sessions" => {
            let workspace_id = param_str(&params, "workspace_id");
            let include_archived = params
                .get("include_archived")
                .and_then(|v| v.as_bool())
                .unwrap_or(false);
            handle_list_chat_sessions(state, &workspace_id, include_archived)
        }
        "get_chat_session" => {
            let chat_session_id = param_chat_session_id(&params);
            handle_get_chat_session(state, &chat_session_id)
        }
        "create_chat_session" => {
            let workspace_id = param_str(&params, "workspace_id");
            handle_create_chat_session(state, &workspace_id)
        }
        "rename_chat_session" => {
            let chat_session_id = param_chat_session_id(&params);
            let name = param_str(&params, "name");
            handle_rename_chat_session(state, &chat_session_id, &name)
        }
        "archive_chat_session" => {
            let chat_session_id = param_chat_session_id(&params);
            handle_archive_chat_session(state, &chat_session_id).await
        }
        "refresh_claude_flags" => {
            // Drop the cache so the next `cached_claude_flag_defs` call
            // re-runs `discover_claude_flags()`. Useful when `claude` has
            // been upgraded while the server is running. Returns null —
            // the next turn (or an explicit list call, if the wire ever
            // grows one) re-fetches lazily.
            let mut guard = state.claude_flag_defs.write().await;
            *guard = None;
            Ok(json!(null))
        }
        _ => Err(format!("Unknown method: {method}")),
    };

    match result {
        Ok(value) => json!({"id": id, "result": value}),
        Err(msg) => json!({"id": id, "error": {"code": -1, "message": msg}}),
    }
}

fn param_str(params: &serde_json::Value, key: &str) -> String {
    params
        .get(key)
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_string()
}

/// Read a chat session id from JSON-RPC params. Prefers the canonical
/// `chat_session_id` key (which the multi-session UI sends) and falls back
/// to the legacy `session_id` key so older clients keep working.
fn param_chat_session_id(params: &serde_json::Value) -> String {
    params
        .get("chat_session_id")
        .or_else(|| params.get("session_id"))
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_string()
}

fn open_db(state: &ServerState) -> Result<Database, String> {
    Database::open(&state.db_path).map_err(|e| e.to_string())
}

/// Mirror of `commands::env::load_disabled_providers` from the Tauri side.
/// Reads per-repo env-provider toggles persisted in `app_settings` under
/// keys of the form `repo:{repo_id}:env_provider:{plugin}:enabled`. The
/// dispatcher consults this set to skip the matching plugin without
/// running its detect/export operations.
fn load_disabled_providers(db: &Database, repo_id: &str) -> std::collections::HashSet<String> {
    let prefix = format!("repo:{repo_id}:env_provider:");
    db.list_app_settings_with_prefix(&prefix)
        .unwrap_or_default()
        .into_iter()
        .filter_map(|(key, value)| {
            if value == "false" {
                let rest = key.strip_prefix(&prefix)?;
                rest.strip_suffix(":enabled").map(|n| n.to_string())
            } else {
                None
            }
        })
        .collect()
}

/// Resolve the merged env-provider activation for a workspace turn.
/// When the server's plugin registry is `None` (no bundled or
/// user-installed plugins) this returns a default `ResolvedEnv`, which
/// applies as a no-op on the spawned command. That matches the
/// behaviour of `resolve_with_registry` against an empty registry and
/// keeps deployments without plugins working unchanged.
///
/// `disabled` is the per-repo set of disabled env-provider names
/// (typically from [`load_disabled_providers`]). Computing it
/// synchronously in the caller avoids holding `&Database` (which is
/// not `Sync`) across the `.await` on the plugin-registry lock —
/// `handle_request` is invoked from inside `tokio::spawn`, so its
/// future tree must stay `Send`.
pub async fn resolve_workspace_env(
    state: &ServerState,
    ws: &claudette::model::Workspace,
    repo: Option<&claudette::model::Repository>,
    worktree_path: &str,
    disabled: std::collections::HashSet<String>,
) -> claudette::env_provider::ResolvedEnv {
    let Some(plugins_lock) = state.plugins.as_ref() else {
        return claudette::env_provider::ResolvedEnv::default();
    };

    let ws_info = claudette::plugin_runtime::host_api::WorkspaceInfo {
        id: ws.id.clone(),
        name: ws.name.clone(),
        branch: ws.branch_name.clone(),
        worktree_path: worktree_path.to_string(),
        repo_path: repo.map(|r| r.path.clone()).unwrap_or_default(),
        repo_id: Some(ws.repository_id.clone()),
    };
    let registry = plugins_lock.read().await;
    claudette::env_provider::resolve_with_registry(
        &registry,
        &state.env_cache,
        std::path::Path::new(worktree_path),
        &ws_info,
        &disabled,
    )
    .await
}

fn now_iso() -> String {
    let dur = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default();
    let secs = dur.as_secs();
    let days = secs / 86400;
    let time_secs = secs % 86400;
    let h = time_secs / 3600;
    let m = (time_secs % 3600) / 60;
    let s = time_secs % 60;
    // Hinnant's algorithm for epoch days → calendar date.
    let z = days + 719468;
    let era = z / 146097;
    let doe = z - era * 146097;
    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;
    let y = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let mo = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = if mo <= 2 { y + 1 } else { y };
    format!("{y:04}-{mo:02}-{d:02} {h:02}:{m:02}:{s:02}")
}

// ---- Command handlers ----

async fn handle_load_initial_data(state: &ServerState) -> Result<serde_json::Value, String> {
    let db = open_db(state)?;
    let repositories = db.list_repositories().map_err(|e| e.to_string())?;
    let workspaces = db.list_workspaces().map_err(|e| e.to_string())?;

    let worktree_base_dir = {
        let dir = state.worktree_base_dir.read().await;
        dir.to_string_lossy().to_string()
    };

    // Check repo path validity.
    let repositories: Vec<_> = repositories
        .into_iter()
        .map(|mut r| {
            r.path_valid = std::path::Path::new(&r.path).is_dir();
            r
        })
        .collect();

    // Resolve default branches concurrently.
    let branch_futures: Vec<_> = repositories
        .iter()
        .filter(|r| r.path_valid)
        .map(|r| {
            let id = r.id.clone();
            let path = r.path.clone();
            let base = r.base_branch.clone();
            let remote = r.default_remote.clone();
            async move {
                let branch = match base {
                    Some(b) => Some(b),
                    None => claudette::git::default_branch(&path, remote.as_deref())
                        .await
                        .ok(),
                };
                branch.map(|b| (id, b))
            }
        })
        .collect();
    let branch_results = futures_util::future::join_all(branch_futures).await;
    let default_branches: std::collections::HashMap<String, String> =
        branch_results.into_iter().flatten().collect();

    // Resolve current branch for each active workspace with a valid worktree path.
    let workspace_branch_futures: Vec<_> = workspaces
        .iter()
        .filter(|ws| ws.status == claudette::model::WorkspaceStatus::Active)
        .filter_map(|ws| {
            ws.worktree_path
                .as_ref()
                .filter(|path| std::path::Path::new(path).is_dir())
                .map(|path| {
                    let id = ws.id.clone();
                    let path = path.clone();
                    async move {
                        match claudette::git::current_branch(&path).await {
                            Ok(branch) => (id, branch),
                            Err(_) => (id, "(detached)".to_string()),
                        }
                    }
                })
        })
        .collect();
    let workspace_branch_results = futures_util::future::join_all(workspace_branch_futures).await;
    let workspace_current_branches: std::collections::HashMap<String, String> =
        workspace_branch_results.into_iter().collect();

    // Update workspace branch_name with current branch from worktree.
    let workspaces: Vec<_> = workspaces
        .into_iter()
        .map(|mut ws| {
            if let Some(current) = workspace_current_branches.get(&ws.id) {
                ws.branch_name = current.clone();
            }
            ws
        })
        .collect();

    let last_messages = db.last_message_per_workspace().map_err(|e| e.to_string())?;

    Ok(json!({
        "repositories": repositories,
        "workspaces": workspaces,
        "worktree_base_dir": worktree_base_dir,
        "default_branches": default_branches,
        "last_messages": last_messages,
    }))
}

async fn handle_load_chat_history(
    state: &ServerState,
    chat_session_id: &str,
) -> Result<serde_json::Value, String> {
    let db = open_db(state)?;
    let messages = db
        .list_chat_messages_for_session(chat_session_id)
        .map_err(|e| e.to_string())?;
    Ok(serde_json::to_value(messages).unwrap_or_default())
}

#[allow(clippy::too_many_arguments)]
async fn handle_send_chat_message(
    state: &Arc<ServerState>,
    writer: &Arc<Writer>,
    chat_session_id: &str,
    content: &str,
    permission_level: Option<&str>,
    model: Option<String>,
    fast_mode: Option<bool>,
    thinking_enabled: Option<bool>,
    plan_mode: Option<bool>,
    effort: Option<String>,
    chrome_enabled: Option<bool>,
    disable_1m_context: Option<bool>,
    backend_id: Option<String>,
    mentioned_files: Option<Vec<String>>,
) -> Result<serde_json::Value, String> {
    let db = open_db(state)?;

    let chat_session_id = chat_session_id.to_string();
    let chat_session = db
        .get_chat_session(&chat_session_id)
        .map_err(|e| e.to_string())?
        .ok_or("Chat session not found")?;
    let workspace_id = chat_session.workspace_id.clone();

    let workspaces = db.list_workspaces().map_err(|e| e.to_string())?;
    let ws = workspaces
        .iter()
        .find(|w| w.id == workspace_id)
        .ok_or("Workspace not found")?;
    let worktree_path = ws
        .worktree_path
        .as_ref()
        .ok_or("Workspace has no worktree")?
        .clone();

    // Save user message.
    let user_msg = ChatMessage {
        id: uuid::Uuid::new_v4().to_string(),
        workspace_id: workspace_id.clone(),
        chat_session_id: chat_session_id.clone(),
        role: ChatRole::User,
        content: content.to_string(),
        cost_usd: None,
        duration_ms: None,
        created_at: now_iso(),
        thinking: None,
        input_tokens: None,
        output_tokens: None,
        cache_read_tokens: None,
        cache_creation_tokens: None,
    };
    db.insert_chat_message(&user_msg)
        .map_err(|e| e.to_string())?;

    let level = permission_level.unwrap_or("full");
    let allowed_tools = tools_for_level(level);

    let repo = db
        .get_repository(&ws.repository_id)
        .map_err(|e| e.to_string())?;

    // Expand @-file mentions into inline file content for the agent prompt.
    let prompt = claudette::file_expand::expand_file_mentions(
        std::path::Path::new(&worktree_path),
        content,
        mentioned_files.as_deref().unwrap_or(&[]),
    )
    .await;

    // Build workspace env vars for the agent subprocess.
    let repo_path = repo.as_ref().map(|r| r.path.as_str()).unwrap_or("");
    let default_branch = match repo.as_ref().and_then(|r| r.base_branch.as_deref()) {
        Some(b) => b.to_string(),
        None => claudette::git::default_branch(
            repo_path,
            repo.as_ref().and_then(|r| r.default_remote.as_deref()),
        )
        .await
        .unwrap_or_else(|_| "main".to_string()),
    };
    let ws_env = claudette::env::WorkspaceEnv::from_workspace(ws, repo_path, default_branch);

    // Resolve the env-provider layer (direnv / mise / dotenv / nix-devshell)
    // once per turn. Mirrors the Tauri path in src-tauri/src/commands/chat.rs:
    // the mtime-keyed cache makes this essentially free on quiet turns; on
    // the first turn or after the user edits `.envrc` / `mise.toml` / etc.,
    // it re-runs the affected plugin. Resolution happens before any
    // `state.agents` lock is taken so the registry RwLock can be acquired
    // without nested-lock concerns. The disabled-provider set is read
    // synchronously here from the already-open `db` so we avoid both a
    // duplicate SQLite connection and a non-`Send` borrow across `.await`.
    let disabled_env_providers =
        load_disabled_providers(&db, repo.as_ref().map(|r| r.id.as_str()).unwrap_or(""));
    let resolved_env = resolve_workspace_env(
        state,
        ws,
        repo.as_ref(),
        &worktree_path,
        disabled_env_providers,
    )
    .await;

    let mut agents = state.agents.write().await;
    let workspace_id_owned = workspace_id.clone();
    let session = agents.entry(chat_session_id.clone()).or_insert_with(|| {
        let instructions = {
            let from_config = repo.as_ref().and_then(|r| {
                let path = r.path.clone();
                claudette::config::load_config(std::path::Path::new(&path))
                    .ok()
                    .flatten()
                    .and_then(|c| c.instructions)
            });
            from_config.or_else(|| repo.as_ref().and_then(|r| r.custom_instructions.clone()))
        };
        AgentSessionState {
            workspace_id: workspace_id_owned,
            session_id: uuid::Uuid::new_v4().to_string(),
            turn_count: 0,
            active_pid: None,
            custom_instructions: instructions,
            session_resolved_env: Default::default(),
            persistent_session: None,
            pending_permissions: std::collections::HashMap::new(),
            pending_message_queue: std::collections::VecDeque::new(),
        }
    });

    // Env-provider drift teardown: the env baked into a running agent's
    // env is fixed at spawn time, so the subprocess won't see `.envrc` /
    // `mise.toml` / `direnv allow` changes until it's respawned. Compare
    // the freshly-resolved vars against the snapshot stored at spawn and
    // reset the session on divergence so this turn launches with the new
    // env. Dropping `persistent_session` here lets `Arc::drop` kill the
    // running `claude` subprocess; the next turn will start a fresh one
    // with `is_resume = false` so the new env is picked up cleanly.
    if session.turn_count > 0 && session.session_resolved_env != resolved_env.vars {
        tracing::info!(
            target: "claudette::ws",
            workspace_id = %workspace_id,
            vars_before = session.session_resolved_env.len(),
            vars_after = resolved_env.vars.len(),
            "env-provider output changed — resetting session"
        );
        session.session_id = uuid::Uuid::new_v4().to_string();
        session.turn_count = 0;
        session.active_pid = None;
        session.session_resolved_env = Default::default();
        session.persistent_session = None;
        session.pending_permissions.clear();
        session.pending_message_queue.clear();
    }

    let is_resume = session.turn_count > 0;
    let session_id = session.session_id.clone();
    let custom_instructions = session.custom_instructions.clone();
    session.turn_count += 1;

    // Load repository MCP configs for injection on EVERY turn. Each `claude
    // --print` is an independent process — MCP connections are per-process and
    // NOT restored from session state on `--resume`.
    let mcp_config = {
        let db_rows = db
            .list_repository_mcp_servers(&ws.repository_id)
            .unwrap_or_default();
        claudette::mcp::cli_config_from_rows(&db_rows)
    };

    // Lazily resolve user-toggled `claude --help` flags. The server has no
    // UI to render the catalog, so we don't pre-warm at boot — instead we
    // discover on first need and cache the result for subsequent turns.
    // Fetch the defs (which may .await on discovery) BEFORE touching the
    // `!Send` `Database`, then resolve synchronously, so this future stays
    // `Send` for `tokio::spawn`.
    let claude_flag_defs = cached_claude_flag_defs(state).await;
    let extra_claude_flags = resolve_extra_claude_flags(&claude_flag_defs, &db, &ws.repository_id);

    let agent_settings = AgentSettings {
        model: if !is_resume { model } else { None },
        fast_mode: fast_mode.unwrap_or(false),
        thinking_enabled: thinking_enabled.unwrap_or(false),
        plan_mode: plan_mode.unwrap_or(false),
        effort,
        chrome_enabled: chrome_enabled.unwrap_or(false),
        mcp_config,
        disable_1m_context: disable_1m_context.unwrap_or(false),
        // The standalone server binary does not implement Claudette's teammate
        // bridge child-mode args, so do not ask Claude Code to launch it as the
        // teammate command from headless remote sessions.
        team_agent_session_tabs_enabled: false,
        backend_runtime: Default::default(),
        hook_bridge: None,
        extra_claude_flags,
    };
    if backend_id.as_deref().is_some_and(|id| id != "anthropic") {
        tracing::warn!(
            target: "claudette::ws",
            backend_id = backend_id.as_deref(),
            "alternate backends are not supported over remote transport yet"
        );
    }

    // Lazily start (or reuse) the long-lived `claude` subprocess for this
    // chat session. Unlike the previous one-shot `run_turn` path, the CLI
    // stays alive with stdin piped so `submit_agent_answer` /
    // `submit_plan_approval` can write `control_response` lines back —
    // which is the only way `AskUserQuestion` / `ExitPlanMode` work over
    // a remote connection.
    let ps_arc = match session.persistent_session.as_ref() {
        Some(existing) => Arc::clone(existing),
        None => {
            tracing::info!(
                target: "claudette::ws",
                chat_session_id = %chat_session_id,
                workspace_id = %workspace_id,
                session_id = %session_id,
                is_resume,
                "starting persistent claude session for remote chat"
            );
            // If `PersistentSession::start` fails (missing `claude`
            // binary, invalid worktree, env-resolution error baked into
            // a bad spawn) the per-session entry we just created above
            // is left in `state.agents` with `turn_count == 1` but no
            // running CLI. Drop it on the failure path so the next
            // attempt re-enters the session-init branch cleanly.
            let ps = match PersistentSession::start(
                std::path::Path::new(&worktree_path),
                &session_id,
                is_resume,
                &allowed_tools,
                custom_instructions.as_deref(),
                &agent_settings,
                Some(&ws_env),
                Some(&resolved_env),
            )
            .await
            {
                Ok(ps) => ps,
                Err(e) => {
                    agents.remove(&chat_session_id);
                    return Err(e);
                }
            };
            let arc = Arc::new(ps);
            session.persistent_session = Some(Arc::clone(&arc));
            arc
        }
    };

    session.active_pid = Some(ps_arc.pid());
    session.session_resolved_env = resolved_env.vars.clone();
    drop(agents);

    // Same cleanup discipline for `send_turn`: if the CLI rejects the
    // user message (e.g. the persistent process died between turns),
    // we drop the now-half-broken session so the next attempt
    // respawns rather than retrying against a corpse.
    let turn_handle = match ps_arc.send_turn(&prompt, &[]).await {
        Ok(handle) => handle,
        Err(e) => {
            let mut agents = state.agents.write().await;
            agents.remove(&chat_session_id);
            return Err(e);
        }
    };

    spawn_turn_event_bridge(
        Arc::clone(state),
        Arc::clone(writer),
        TurnBridgeContext {
            workspace_id: workspace_id.clone(),
            chat_session_id: chat_session_id.clone(),
            worktree_path: worktree_path.clone(),
            user_msg_id: user_msg.id.clone(),
        },
        turn_handle,
    );

    Ok(json!(null))
}

/// Bookkeeping passed into [`spawn_turn_event_bridge`]. Bundled into a
/// struct so the two call sites (initial `send_chat_message` and the
/// post-Result `dispatch_queued_message` follow-up) can't accidentally
/// disagree on what each turn forwards / persists.
pub struct TurnBridgeContext {
    pub workspace_id: String,
    pub chat_session_id: String,
    pub worktree_path: String,
    /// Anchor for the per-turn checkpoint when the assistant emits no
    /// text (e.g. a turn that only produced tool calls). Mirrors the
    /// desktop's `anchor_msg_id` fallback in
    /// `src-tauri/src/commands/chat/send.rs`.
    pub user_msg_id: String,
}

/// Spawn the per-turn event-bridge task. Reads `AgentEvent`s from
/// the running `claude` subprocess and:
///   - Forwards each as an `agent-stream` event on the WS.
///   - Routes `ControlRequest::CanUseTool` events to per-session
///     `pending_permissions` and emits `agent-permission-prompt` so
///     the client can render AskUserQuestion / ExitPlanMode UI.
///   - Persists assistant messages + usage to the DB.
///   - On `Result` (turn complete) creates a checkpoint and dispatches
///     the head of `pending_message_queue` if a steered message is
///     waiting.
///   - On `ProcessExited` clears the persistent session and drops the
///     map entry on a non-zero / pre-init exit.
///
/// Extracted so the queued-message path also gets all of the above —
/// before this lived only inline in `handle_send_chat_message`, so
/// steered turns silently lost their AskUserQuestion routing.
pub fn spawn_turn_event_bridge(
    state: Arc<ServerState>,
    writer: Arc<Writer>,
    ctx: TurnBridgeContext,
    turn_handle: claudette::agent::TurnHandle,
) {
    let TurnBridgeContext {
        workspace_id: ws_id,
        chat_session_id: chat_session_id_for_stream,
        worktree_path: wt_path,
        user_msg_id,
    } = ctx;
    let db_path = state.db_path.clone();
    tokio::spawn(async move {
        let mut rx = turn_handle.event_rx;
        let mut got_init = false;
        let mut pending_thinking: Option<String> = None;
        // Tracks the most recent per-message usage observed on a MessageDelta
        // event. Written into the next persisted assistant ChatMessage and
        // reset to None after each persistence so per-message counts stay
        // distinct across multi-message turns. Mirrors the Tauri bridge.
        let mut latest_usage: Option<claudette::agent::TokenUsage> = None;
        // Tracks the most recently persisted assistant message id so the
        // Result-event arm can update its cost/duration and use it as the
        // checkpoint anchor without re-querying the DB. Mirrors the Tauri
        // bridge.
        let mut last_assistant_msg_id: Option<String> = None;
        while let Some(event) = rx.recv().await {
            if let AgentEvent::Stream(StreamEvent::System { ref subtype, .. }) = event
                && subtype == "init"
            {
                got_init = true;
            }

            // Persist the captured `claude` invocation on first emit.
            // Mirrors the desktop path's `set_session_cli_invocation` Tauri
            // command — the underlying DB helper is first-emit-wins, so a
            // respawn within the same session leaves the original recorded
            // value in place. Banner is UX-cosmetic; drop on error.
            if let AgentEvent::Stream(StreamEvent::System {
                ref subtype,
                ref command_line,
                ..
            }) = event
                && subtype == "command_line"
                && let Some(line) = command_line.as_deref()
                && let Ok(db) = Database::open(&db_path)
                && let Err(e) =
                    db.set_chat_session_cli_invocation(&chat_session_id_for_stream, line)
            {
                tracing::warn!(target: "claudette::ws", error = %e, "persist cli_invocation failed");
            }

            // Capture pending `can_use_tool` requests (AskUserQuestion,
            // ExitPlanMode) into per-session state so `submit_agent_answer`
            // / `submit_plan_approval` can look them up by `tool_use_id` and
            // write the matching `control_response` back to the CLI's
            // stdin. Mirrors the desktop's `queue_control_prompt` path in
            // `src-tauri/src/commands/chat/send.rs`. The same event is also
            // forwarded in the `agent-stream` payload below so the client
            // sees the tool call in the transcript and renders the prompt
            // UI off the snapshot it already maintains.
            if let AgentEvent::Stream(StreamEvent::ControlRequest {
                ref request_id,
                ref request,
            }) = event
                && let ControlRequestInner::CanUseTool {
                    tool_name,
                    tool_use_id,
                    input,
                } = request
            {
                let mut agents = state.agents.write().await;
                if let Some(session) = agents.get_mut(&chat_session_id_for_stream) {
                    session.pending_permissions.insert(
                        tool_use_id.clone(),
                        PendingPermission {
                            request_id: request_id.clone(),
                            tool_name: tool_name.clone(),
                            original_input: input.clone(),
                        },
                    );
                }
                drop(agents);

                let prompt_msg = json!({
                    "event": "agent-permission-prompt",
                    "payload": {
                        "workspace_id": ws_id,
                        "chat_session_id": chat_session_id_for_stream,
                        "tool_use_id": tool_use_id,
                        "tool_name": tool_name,
                        "request_id": request_id,
                        "input": input,
                    }
                });
                send_message(&writer, &prompt_msg).await;
            }

            if let AgentEvent::ProcessExited(code) = &event {
                // The CLI subprocess died — drop the session entry entirely
                // so the next `send_chat_message` starts a fresh
                // `PersistentSession` instead of trying to write to a
                // dead stdin. Doing the full `remove(...)` here (rather
                // than the previous "clear-then-conditionally-remove"
                // dance) avoids leaving a half-stale `AgentSessionState`
                // around when the CLI exits with `code == Some(0)` after
                // a normal turn — `PersistentSession` only stays alive
                // when the CLI is alive, so an exit is the canonical
                // "session ended" signal regardless of code. The unused
                // `got_init` / `code` are kept for the trace-level log
                // so a regression in process startup is still visible.
                tracing::info!(
                    target: "claudette::ws",
                    chat_session_id = %chat_session_id_for_stream,
                    got_init,
                    exit_code = ?code,
                    "claude subprocess exited — dropping session entry"
                );
                let mut agents = state.agents.write().await;
                agents.remove(&chat_session_id_for_stream);
            }

            // Track per-assistant-message cumulative usage as the CLI streams
            // it. The final MessageDelta before message_stop carries the
            // authoritative per-message total; we overwrite on every delta and
            // consume it when the assistant message is persisted below.
            if let AgentEvent::Stream(StreamEvent::Stream {
                event: InnerStreamEvent::MessageDelta { usage: Some(u) },
            }) = &event
            {
                latest_usage = Some(u.clone());
            }

            // Persist assistant messages. The CLI may fire multiple assistant
            // events per turn (thinking-only, then text). Accumulate thinking
            // and save only when text content arrives.
            if let AgentEvent::Stream(StreamEvent::Assistant { ref message }) = event {
                let full_text = extract_assistant_text(message);

                if let Some(t) = extract_event_thinking(message) {
                    pending_thinking = Some(match pending_thinking.take() {
                        Some(mut existing) => {
                            existing.push_str(&t);
                            existing
                        }
                        None => t,
                    });
                }

                if !full_text.trim().is_empty()
                    && let Ok(db) = Database::open(&db_path)
                {
                    let msg = build_assistant_chat_message(BuildAssistantArgs {
                        workspace_id: &ws_id,
                        chat_session_id: &chat_session_id_for_stream,
                        content: full_text,
                        thinking: pending_thinking.take(),
                        usage: latest_usage.take(),
                        created_at: now_iso(),
                    });
                    let msg_id = msg.id.clone();
                    if db.insert_chat_message(&msg).is_ok() {
                        last_assistant_msg_id = Some(msg_id);
                    }
                }
            }

            // Update cost/duration on result events, then create a checkpoint.
            if let AgentEvent::Stream(StreamEvent::Result {
                ref total_cost_usd,
                ref duration_ms,
                ..
            }) = event
            {
                // Turn complete — clear `active_pid` (the persistent session
                // stays alive for the next turn) and dispatch the head of
                // the queued-message queue if the client steered while the
                // turn was running. This mirrors the desktop's
                // `process_pending_message_queue` post-turn handoff.
                let next_queued = {
                    let mut agents = state.agents.write().await;
                    if let Some(session) = agents.get_mut(&chat_session_id_for_stream) {
                        session.active_pid = None;
                        session.pending_message_queue.pop_front()
                    } else {
                        None
                    }
                };
                if let Some(queued) = next_queued
                    && let Err(e) = dispatch_queued_message(
                        Arc::clone(&state),
                        Arc::clone(&writer),
                        chat_session_id_for_stream.clone(),
                        queued,
                    )
                    .await
                {
                    tracing::warn!(
                        target: "claudette::ws",
                        error = %e,
                        "failed to dispatch queued message after turn end"
                    );
                }

                if let (Some(cost), Some(dur)) = (total_cost_usd, duration_ms)
                    && let Some(ref msg_id) = last_assistant_msg_id
                    && let Ok(db) = Database::open(&db_path)
                {
                    let _ = db.update_chat_message_cost(msg_id, *cost, *dur);
                }

                let anchor_msg_id = last_assistant_msg_id.as_deref().unwrap_or(&user_msg_id);
                if let Some(cp) = create_turn_checkpoint(CheckpointArgs {
                    db_path: &db_path,
                    workspace_id: &ws_id,
                    chat_session_id: &chat_session_id_for_stream,
                    anchor_msg_id,
                    worktree_path: &wt_path,
                    created_at: now_iso(),
                })
                .await
                {
                    let event_msg = json!({
                        "event": "checkpoint-created",
                        "payload": {
                            "workspace_id": &ws_id,
                            "chat_session_id": &chat_session_id_for_stream,
                            "checkpoint": &cp,
                        }
                    });
                    send_message(&writer, &event_msg).await;
                }
            }

            // Emit event over WebSocket.
            let event_msg = json!({
                "event": "agent-stream",
                "payload": {
                    "workspace_id": ws_id,
                    "session_id": chat_session_id_for_stream,
                    "event": event,
                }
            });
            send_message(&writer, &event_msg).await;
        }
    });
}

async fn handle_stop_agent(
    state: &ServerState,
    chat_session_id: &str,
) -> Result<serde_json::Value, String> {
    let db = open_db(state)?;
    let chat_session_id = chat_session_id.to_string();
    let chat_session = db
        .get_chat_session(&chat_session_id)
        .map_err(|e| e.to_string())?
        .ok_or("Chat session not found")?;
    let workspace_id = chat_session.workspace_id.clone();

    // Capture the pid and drop the persistent session under the lock, then
    // run the async kill outside it. Dropping `persistent_session` here
    // ensures the next `send_chat_message` starts a fresh CLI instead of
    // trying to write to a stdin attached to a process we just killed.
    let pid_to_stop = {
        let mut agents = state.agents.write().await;
        if let Some(session) = agents.get_mut(&chat_session_id) {
            session.persistent_session = None;
            session.pending_permissions.clear();
            session.pending_message_queue.clear();
            session.active_pid.take()
        } else {
            None
        }
    };
    if let Some(pid) = pid_to_stop {
        agent::stop_agent(pid).await?;
    }

    let msg = ChatMessage {
        id: uuid::Uuid::new_v4().to_string(),
        workspace_id,
        chat_session_id,
        role: ChatRole::System,
        content: "Agent stopped".to_string(),
        cost_usd: None,
        duration_ms: None,
        created_at: now_iso(),
        thinking: None,
        input_tokens: None,
        output_tokens: None,
        cache_read_tokens: None,
        cache_creation_tokens: None,
    };
    db.insert_chat_message(&msg).map_err(|e| e.to_string())?;
    Ok(json!(null))
}

/// Resolve a pending `AskUserQuestion` request. Mirrors the desktop's
/// `submit_agent_answer` in `src-tauri/src/commands/chat/interaction.rs`:
/// look up the pending entry by `tool_use_id`, verify it's an
/// AskUserQuestion (not an approval-style tool), drain it, build a
/// `behavior: "allow"` response with `answers` layered onto the original
/// input, and write a `control_response` line to the CLI's stdin.
pub async fn handle_submit_agent_answer(
    state: &ServerState,
    chat_session_id: &str,
    tool_use_id: &str,
    answers: std::collections::HashMap<String, String>,
    annotations: Option<serde_json::Value>,
) -> Result<serde_json::Value, String> {
    let (pending, ps) = {
        let mut agents = state.agents.write().await;
        let session = agents.get_mut(chat_session_id).ok_or("Session not found")?;
        let ps = session
            .persistent_session
            .clone()
            .ok_or("Agent session is not active")?;
        match session.pending_permissions.get(tool_use_id) {
            None => {
                let pending_ids: Vec<String> =
                    session.pending_permissions.keys().cloned().collect();
                return Err(format!(
                    "No pending permission request for tool_use_id {tool_use_id} (pending: {pending_ids:?})"
                ));
            }
            Some(p) if p.tool_name != "AskUserQuestion" => {
                return Err(format!(
                    "Pending tool for {tool_use_id} is {}, not AskUserQuestion",
                    p.tool_name
                ));
            }
            _ => {}
        }
        let pending = session
            .pending_permissions
            .remove(tool_use_id)
            .expect("checked above");
        (pending, ps)
    };

    let mut updated_input = pending.original_input.clone();
    if !updated_input.is_object() {
        updated_input = serde_json::Value::Object(serde_json::Map::new());
    }
    if let Some(obj) = updated_input.as_object_mut() {
        let answers_value =
            serde_json::to_value(&answers).map_err(|e| format!("Failed to encode answers: {e}"))?;
        obj.insert("answers".to_string(), answers_value);
        if let Some(ann) = annotations {
            obj.insert("annotations".to_string(), ann);
        }
    }

    let response = json!({
        "behavior": "allow",
        "updatedInput": updated_input,
    });
    ps.send_control_response(&pending.request_id, response)
        .await?;
    Ok(json!(null))
}

/// Which approval-style tool kinds a given handler accepts. Mirrors the
/// desktop's split between `submit_plan_approval` (ExitPlanMode only —
/// the plan-mode UX shows specialized copy) and `submit_agent_approval`
/// (any approval-style tool — Codex variants etc.).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ApprovalKind {
    PlanModeOnly,
    AnyApprovalTool,
}

impl ApprovalKind {
    fn accepts(&self, tool_name: &str) -> bool {
        match self {
            ApprovalKind::PlanModeOnly => tool_name == "ExitPlanMode",
            // Match the desktop's `is_user_approval_tool` predicate
            // (`src-tauri/src/commands/chat/interaction.rs`) — ExitPlanMode
            // plus the Codex approval tool family. The Codex names are
            // gated behind the `alternative-backends` feature on the
            // desktop, but listing them here unconditionally is harmless:
            // if the WS server never spawns a Codex agent, these strings
            // simply never appear in `pending_permissions`.
            ApprovalKind::AnyApprovalTool => {
                matches!(
                    tool_name,
                    "ExitPlanMode"
                        | "codex_request_approval"
                        | "codex_request_command_approval"
                        | "codex_request_patch_approval"
                )
            }
        }
    }
}

/// Resolve a pending `ExitPlanMode` (or other approval-style) tool
/// request. `approved=true` allows with the model's original input;
/// `approved=false` denies with the supplied reason (or a sensible
/// default that nudges the model to revise and resubmit the plan).
/// `kind` controls which pending `tool_name` values are accepted —
/// callers from `submit_plan_approval` pass `PlanModeOnly` (the UX
/// shows plan-specific copy and would mis-render for a Codex
/// approval), while `submit_agent_approval` passes `AnyApprovalTool`.
pub async fn handle_submit_approval(
    state: &ServerState,
    chat_session_id: &str,
    tool_use_id: &str,
    approved: bool,
    reason: Option<String>,
    kind: ApprovalKind,
) -> Result<serde_json::Value, String> {
    let (pending, ps) = {
        let mut agents = state.agents.write().await;
        let session = agents.get_mut(chat_session_id).ok_or("Session not found")?;
        let ps = session
            .persistent_session
            .clone()
            .ok_or("Agent session is not active")?;
        match session.pending_permissions.get(tool_use_id) {
            None => {
                let pending_ids: Vec<String> =
                    session.pending_permissions.keys().cloned().collect();
                return Err(format!(
                    "No pending permission request for tool_use_id {tool_use_id} (pending: {pending_ids:?})"
                ));
            }
            Some(p) if !kind.accepts(&p.tool_name) => {
                return Err(format!(
                    "Pending tool for {tool_use_id} is {}, not an approval request accepted by this handler",
                    p.tool_name
                ));
            }
            _ => {}
        }
        let pending = session
            .pending_permissions
            .remove(tool_use_id)
            .expect("checked above");
        (pending, ps)
    };

    let response = if approved {
        json!({
            "behavior": "allow",
            "updatedInput": pending.original_input,
        })
    } else {
        let feedback = reason.unwrap_or_else(|| "Plan denied. Please revise the approach.".into());
        let message = format!(
            "{feedback}\n\nRevise the plan to address this feedback, then call ExitPlanMode again to present the updated plan for approval. Do not begin implementation until the user approves the revised plan."
        );
        json!({
            "behavior": "deny",
            "message": message,
        })
    };
    ps.send_control_response(&pending.request_id, response)
        .await?;
    Ok(json!(null))
}

/// Steer a user message into the chat session. If a turn is currently
/// running (`active_pid.is_some()`), the message is queued and dispatched
/// after the running turn completes. If no turn is active, it's queued
/// for the caller's responsibility to follow up with `send_chat_message`
/// (we don't auto-dispatch from here because that path needs the full
/// agent settings context which only `send_chat_message` carries).
pub async fn handle_steer_queued_chat_message(
    state: &ServerState,
    chat_session_id: &str,
    content: &str,
) -> Result<serde_json::Value, String> {
    if content.trim().is_empty() {
        return Err("Cannot steer an empty message".into());
    }
    let mut agents = state.agents.write().await;
    let session = agents.get_mut(chat_session_id).ok_or("Session not found")?;
    session.pending_message_queue.push_back(QueuedMessage {
        content: content.to_string(),
    });
    Ok(json!({
        "queued": true,
        "queue_depth": session.pending_message_queue.len(),
    }))
}

/// Send a previously-queued user message into the active persistent
/// session. Called from the post-turn `Result` handler. We can't reuse
/// the full `handle_send_chat_message` path because that re-resolves env,
/// re-applies settings, and re-spawns — we want the lighter "next turn
/// on the same persistent session" path. The user message is persisted
/// so the chat transcript matches an interactive send.
///
/// Reuses [`spawn_turn_event_bridge`] so a queued turn gets the same
/// `ControlRequest::CanUseTool` routing as an initial turn — i.e. an
/// AskUserQuestion or ExitPlanMode emitted by the model during a
/// steered turn still surfaces as `agent-permission-prompt` to the
/// client and can be answered via `submit_agent_answer` /
/// `submit_plan_approval`. Before the helper was extracted, queued
/// turns silently dropped those prompts.
async fn dispatch_queued_message(
    state: Arc<ServerState>,
    writer: Arc<Writer>,
    chat_session_id: String,
    queued: QueuedMessage,
) -> Result<(), String> {
    let (ps_arc, workspace_id, worktree_path) = {
        let agents = state.agents.read().await;
        let session = agents
            .get(&chat_session_id)
            .ok_or("Session not found in dispatch_queued_message")?;
        let ps = session
            .persistent_session
            .as_ref()
            .ok_or("No persistent session for queued message")?;
        (
            Arc::clone(ps),
            session.workspace_id.clone(),
            // Re-resolve the worktree path so the checkpoint anchor in
            // the event bridge has the same value the initial turn used.
            Database::open(&state.db_path)
                .ok()
                .and_then(|db| db.list_workspaces().ok())
                .and_then(|ws| {
                    ws.into_iter()
                        .find(|w| w.id == session.workspace_id)
                        .and_then(|w| w.worktree_path)
                })
                .unwrap_or_default(),
        )
    };

    let db = Database::open(&state.db_path).map_err(|e| e.to_string())?;
    let user_msg = ChatMessage {
        id: uuid::Uuid::new_v4().to_string(),
        workspace_id: workspace_id.clone(),
        chat_session_id: chat_session_id.clone(),
        role: ChatRole::User,
        content: queued.content.clone(),
        cost_usd: None,
        duration_ms: None,
        created_at: now_iso(),
        thinking: None,
        input_tokens: None,
        output_tokens: None,
        cache_read_tokens: None,
        cache_creation_tokens: None,
    };
    db.insert_chat_message(&user_msg)
        .map_err(|e| e.to_string())?;

    let turn_handle = ps_arc.send_turn(&queued.content, &[]).await?;

    {
        let mut agents = state.agents.write().await;
        if let Some(session) = agents.get_mut(&chat_session_id) {
            session.active_pid = Some(ps_arc.pid());
            session.turn_count += 1;
        }
    }

    spawn_turn_event_bridge(
        Arc::clone(&state),
        writer,
        TurnBridgeContext {
            workspace_id,
            chat_session_id,
            worktree_path,
            user_msg_id: user_msg.id,
        },
        turn_handle,
    );

    Ok(())
}

async fn handle_create_workspace(
    state: &ServerState,
    repository_id: &str,
    name: &str,
    preserve_supplied_name: bool,
) -> Result<serde_json::Value, String> {
    use claudette::ops::{NoopHooks, workspace as ops_workspace};

    let mut db = open_db(state)?;
    let repo = db
        .get_repository(repository_id)
        .map_err(|e| e.to_string())?
        .ok_or("Repository not found")?;

    let worktree_base_dir = state.worktree_base_dir.read().await.clone();
    // Server uses the repo's path_slug as the prefix — historical behavior
    // that survives the refactor unchanged. The GUI resolves a per-user
    // prefix from app settings; harmonizing the two is a follow-up.
    let branch_prefix = format!("{}/", repo.path_slug);

    let params = ops_workspace::CreateParams {
        repo_id: repository_id,
        name,
        branch_prefix: &branch_prefix,
    };
    let out = if preserve_supplied_name {
        ops_workspace::create_preserving_supplied_name(
            &mut db,
            &NoopHooks,
            worktree_base_dir.as_path(),
            params,
        )
        .await
    } else {
        ops_workspace::create(&mut db, &NoopHooks, worktree_base_dir.as_path(), params).await
    }
    .map_err(|e| e.to_string())?;

    // Run the setup script (if configured) for parity with the GUI path.
    // The server has no plugin registry, so env-provider stack is not
    // applied — adding that requires loading plugins server-side.
    let setup_result = ops_workspace::resolve_and_run_setup(
        &out.workspace,
        std::path::Path::new(&repo.path),
        std::path::Path::new(&out.worktree_path),
        repo.setup_script.as_deref(),
        repo.base_branch.as_deref(),
        repo.default_remote.as_deref(),
        None,
    )
    .await;

    Ok(json!({
        "workspace": out.workspace,
        "default_session_id": out.default_session_id,
        "setup_result": setup_result,
    }))
}

async fn handle_archive_workspace(
    state: &ServerState,
    workspace_id: &str,
) -> Result<serde_json::Value, String> {
    use claudette::ops::{NoopHooks, workspace as ops_workspace};

    // Stop any running agents for sessions in this workspace. Collect the
    // PIDs to stop under the lock, then drop the lock before awaiting any
    // process teardowns to avoid blocking unrelated requests. Removing
    // each `AgentSessionState` from the map also drops its
    // `Arc<PersistentSession>` reference, which (assuming nothing else
    // holds an Arc clone) tears down the CLI subprocess via the existing
    // RAII path.
    let pids_to_stop: Vec<u32> = {
        let mut agents = state.agents.write().await;
        let to_remove: Vec<String> = agents
            .iter()
            .filter(|(_, s)| s.workspace_id == workspace_id)
            .map(|(k, _)| k.clone())
            .collect();
        to_remove
            .into_iter()
            .filter_map(|key| agents.remove(&key).and_then(|s| s.active_pid))
            .collect()
    };
    for pid in pids_to_stop {
        let _ = agent::stop_agent(pid).await;
    }

    let mut db = open_db(state)?;
    let _ = ops_workspace::archive(
        &mut db,
        &NoopHooks,
        ops_workspace::ArchiveParams {
            workspace_id,
            // The server doesn't surface a delete-branch toggle today;
            // archive without branch deletion mirrors prior behavior so
            // remote clients can still restore archived workspaces.
            delete_branch: false,
        },
    )
    .await
    .map_err(|e| e.to_string())?;

    Ok(json!(null))
}

pub async fn handle_restore_workspace(
    state: &ServerState,
    workspace_id: &str,
) -> Result<serde_json::Value, String> {
    use claudette::ops::{NoopHooks, workspace as ops_workspace};

    let mut db = open_db(state)?;
    let worktree_base = state.worktree_base_dir.read().await.clone();

    let out = ops_workspace::restore(
        &mut db,
        &NoopHooks,
        worktree_base.as_path(),
        ops_workspace::RestoreParams { workspace_id },
    )
    .await
    .map_err(|e| e.to_string())?;

    Ok(json!({
        "workspace": out.workspace,
        "worktree_path": out.worktree_path,
    }))
}

async fn handle_load_diff_files(
    state: &ServerState,
    workspace_id: &str,
) -> Result<serde_json::Value, String> {
    let db = open_db(state)?;
    let workspaces = db.list_workspaces().map_err(|e| e.to_string())?;
    let ws = workspaces
        .iter()
        .find(|w| w.id == workspace_id)
        .ok_or("Workspace not found")?;
    let worktree_path = ws
        .worktree_path
        .as_ref()
        .ok_or("Workspace has no worktree")?;

    let repo = db
        .get_repository(&ws.repository_id)
        .map_err(|e| e.to_string())?
        .ok_or("Repository not found")?;

    let base_branch = match repo.base_branch.as_deref() {
        Some(b) => b.to_string(),
        None => claudette::git::default_branch(&repo.path, repo.default_remote.as_deref())
            .await
            .map_err(|e| format!("{e:?}"))?,
    };

    let merge_base = claudette::diff::merge_base(worktree_path, "HEAD", &base_branch)
        .await
        .map_err(|e| format!("{e:?}"))?;

    let (files, staged_files) = tokio::join!(
        claudette::diff::changed_files(worktree_path, &merge_base),
        claudette::diff::staged_changed_files(worktree_path, &merge_base),
    );

    let files = files.map_err(|e| format!("{e:?}"))?;
    let staged_files = staged_files.ok();

    Ok(json!({
        "files": files,
        "merge_base": merge_base,
        "staged_files": staged_files,
    }))
}

async fn handle_load_file_diff(
    worktree_path: &str,
    file_path: &str,
    merge_base: &str,
    diff_layer: Option<&str>,
) -> Result<serde_json::Value, String> {
    let raw =
        claudette::diff::file_diff_for_layer(worktree_path, merge_base, file_path, diff_layer)
            .await
            .map_err(|e| format!("{e:?}"))?;
    let parsed = claudette::diff::parse_unified_diff(&raw, file_path);
    Ok(serde_json::to_value(parsed).unwrap_or_default())
}

async fn handle_spawn_pty(
    state: &Arc<ServerState>,
    writer: &Arc<Writer>,
    workspace_id: &str,
    cwd: &str,
    rows: u16,
    cols: u16,
) -> Result<serde_json::Value, String> {
    let pty_system = native_pty_system();
    let pair = pty_system
        .openpty(PtySize {
            rows,
            cols,
            pixel_width: 0,
            pixel_height: 0,
        })
        .map_err(|e| e.to_string())?;

    let shell = std::env::var("SHELL").unwrap_or_else(|_| "/bin/sh".to_string());
    let mut cmd = CommandBuilder::new(&shell);
    cmd.cwd(cwd);

    // Inject workspace context env vars (best-effort — DB lookup may fail).
    if let Ok(db) = open_db(state)
        && let Ok(wss) = db.list_workspaces()
        && let Some(ws) = wss.iter().find(|w| w.id == workspace_id)
    {
        let repo = db.get_repository(&ws.repository_id).ok().flatten();
        let repo_path = repo.as_ref().map(|r| r.path.as_str()).unwrap_or_default();
        let default_branch = match repo.as_ref().and_then(|r| r.base_branch.as_deref()) {
            Some(b) => b.to_string(),
            None => claudette::git::default_branch(
                repo_path,
                repo.as_ref().and_then(|r| r.default_remote.as_deref()),
            )
            .await
            .unwrap_or_else(|_| "main".into()),
        };
        let ws_env = claudette::env::WorkspaceEnv::from_workspace(ws, repo_path, default_branch);
        for (k, v) in ws_env.vars() {
            cmd.env(k, v);
        }
    }

    let child = pair.slave.spawn_command(cmd).map_err(|e| e.to_string())?;
    let pty_reader = pair.master.try_clone_reader().map_err(|e| e.to_string())?;
    let pty_writer = pair.master.take_writer().map_err(|e| e.to_string())?;

    let pty_id = state.next_pty_id();

    {
        let mut ptys = state.ptys.write().await;
        ptys.insert(
            pty_id,
            PtyHandle {
                writer: std::sync::Mutex::new(pty_writer),
                master: std::sync::Mutex::new(pair.master),
                child: std::sync::Mutex::new(child),
            },
        );
    }

    // Spawn a reader task that forwards PTY output over WebSocket.
    let writer = Arc::clone(writer);
    let state_clone = Arc::clone(state);
    tokio::task::spawn_blocking(move || {
        use std::io::Read;
        let mut reader = pty_reader;
        let mut buf = [0u8; 4096];
        loop {
            match reader.read(&mut buf) {
                Ok(0) => break,
                Ok(n) => {
                    let data: Vec<u8> = buf[..n].to_vec();
                    let writer = writer.clone();
                    let rt = tokio::runtime::Handle::current();
                    rt.block_on(async {
                        let msg = json!({
                            "event": "pty-output",
                            "payload": {
                                "pty_id": pty_id,
                                "data": data,
                            }
                        });
                        send_message(&writer, &msg).await;
                    });
                }
                Err(_) => break,
            }
        }
        // Clean up PTY on exit.
        let rt = tokio::runtime::Handle::current();
        rt.block_on(async {
            let mut ptys = state_clone.ptys.write().await;
            ptys.remove(&pty_id);
        });
    });

    Ok(json!({"pty_id": pty_id}))
}

async fn handle_write_pty(
    state: &ServerState,
    pty_id: u64,
    data: &[u8],
) -> Result<serde_json::Value, String> {
    let ptys = state.ptys.read().await;
    let handle = ptys.get(&pty_id).ok_or("PTY not found")?;
    let mut writer = handle.writer.lock().map_err(|e| e.to_string())?;
    writer.write_all(data).map_err(|e| e.to_string())?;
    writer.flush().map_err(|e| e.to_string())?;
    Ok(json!(null))
}

async fn handle_resize_pty(
    state: &ServerState,
    pty_id: u64,
    rows: u16,
    cols: u16,
) -> Result<serde_json::Value, String> {
    let ptys = state.ptys.read().await;
    let handle = ptys.get(&pty_id).ok_or("PTY not found")?;
    let master = handle.master.lock().map_err(|e| e.to_string())?;
    master
        .resize(PtySize {
            rows,
            cols,
            pixel_width: 0,
            pixel_height: 0,
        })
        .map_err(|e| e.to_string())?;
    Ok(json!(null))
}

async fn handle_close_pty(state: &ServerState, pty_id: u64) -> Result<serde_json::Value, String> {
    let mut ptys = state.ptys.write().await;
    if let Some(handle) = ptys.remove(&pty_id)
        && let Ok(mut child) = handle.child.lock()
    {
        let _ = child.kill();
    }
    Ok(json!(null))
}

fn handle_get_app_setting(state: &ServerState, key: &str) -> Result<serde_json::Value, String> {
    let db = open_db(state)?;
    let value = db.get_app_setting(key).map_err(|e| e.to_string())?;
    Ok(json!(value))
}

async fn handle_set_app_setting(
    state: &ServerState,
    key: &str,
    value: &str,
) -> Result<serde_json::Value, String> {
    let db = open_db(state)?;
    db.set_app_setting(key, value).map_err(|e| e.to_string())?;
    if key == "worktree_base_dir" {
        let mut dir = state.worktree_base_dir.write().await;
        *dir = std::path::PathBuf::from(value);
    }
    Ok(json!(null))
}

fn handle_list_chat_sessions(
    state: &ServerState,
    workspace_id: &str,
    include_archived: bool,
) -> Result<serde_json::Value, String> {
    let db = open_db(state)?;
    let sessions = db
        .list_chat_sessions_for_workspace(workspace_id, include_archived)
        .map_err(|e| e.to_string())?;
    serde_json::to_value(sessions).map_err(|e| e.to_string())
}

fn handle_load_completed_turns(
    state: &ServerState,
    chat_session_id: &str,
) -> Result<serde_json::Value, String> {
    let db = open_db(state)?;
    let turns = db
        .list_completed_turns_for_session(chat_session_id)
        .map_err(|e| e.to_string())?;
    serde_json::to_value(turns).map_err(|e| e.to_string())
}

fn handle_get_chat_session(
    state: &ServerState,
    chat_session_id: &str,
) -> Result<serde_json::Value, String> {
    let db = open_db(state)?;
    let session = db
        .get_chat_session(chat_session_id)
        .map_err(|e| e.to_string())?
        .ok_or("Session not found")?;
    serde_json::to_value(session).map_err(|e| e.to_string())
}

fn handle_create_chat_session(
    state: &ServerState,
    workspace_id: &str,
) -> Result<serde_json::Value, String> {
    let db = open_db(state)?;
    let session = db
        .create_chat_session(workspace_id)
        .map_err(|e| e.to_string())?;
    serde_json::to_value(session).map_err(|e| e.to_string())
}

fn handle_rename_chat_session(
    state: &ServerState,
    chat_session_id: &str,
    name: &str,
) -> Result<serde_json::Value, String> {
    let capped = claudette::model::validate_session_name(name).map_err(String::from)?;
    let db = open_db(state)?;
    db.rename_chat_session(chat_session_id, &capped)
        .map_err(|e| e.to_string())?;
    Ok(json!(null))
}

async fn handle_archive_chat_session(
    state: &ServerState,
    chat_session_id: &str,
) -> Result<serde_json::Value, String> {
    let db = open_db(state)?;
    let session = db
        .get_chat_session(chat_session_id)
        .map_err(|e| e.to_string())?
        .ok_or("Session not found")?;
    let workspace_id = session.workspace_id.clone();

    // Stop and remove the live agent for this session.
    // Capture the PID under the lock, then drop the lock before the async stop.
    let pid_to_stop = {
        let mut agents = state.agents.write().await;
        agents
            .remove(chat_session_id)
            .and_then(|mut agent| agent.active_pid.take())
    };
    if let Some(pid) = pid_to_stop {
        let _ = agent::stop_agent(pid).await;
    }

    let fresh = db
        .archive_chat_session_ensuring_active(chat_session_id, &workspace_id)
        .map_err(|e| e.to_string())?;
    if let Some(fresh) = fresh {
        return serde_json::to_value(fresh).map_err(|e| e.to_string());
    }

    Ok(json!(null))
}

/// Resolve the cached `claude --help` defs (running discovery on first
/// call) into a clone the caller can use synchronously alongside a
/// `!Send` `Database`. Returning a clone — rather than holding the
/// `RwLock` guard across the resolver call — keeps the handler future
/// `Send`, which `tokio::spawn` requires.
async fn cached_claude_flag_defs(
    state: &Arc<crate::ws::ServerState>,
) -> Result<Vec<claudette::claude_help::ClaudeFlagDef>, String> {
    {
        let guard = state.claude_flag_defs.read().await;
        if let Some(cached) = guard.as_ref() {
            return cached.clone();
        }
    }
    let result = claudette::claude_help::discover_claude_flags().await;
    let mut guard = state.claude_flag_defs.write().await;
    if guard.is_none() {
        *guard = Some(result.clone());
    }
    guard.as_ref().expect("just populated").clone()
}

/// Synchronous resolver: combine the cached defs with persisted enable /
/// value state and produce the `(flag, optional_value)` list ready to
/// feed into `AgentSettings.extra_claude_flags`. Pulled out so the call
/// site can keep the `!Send` `Database` reference off the await stack.
fn resolve_extra_claude_flags(
    defs: &Result<Vec<claudette::claude_help::ClaudeFlagDef>, String>,
    db: &Database,
    repo_id: &str,
) -> Vec<(String, Option<String>)> {
    match defs {
        Ok(defs) => {
            match claudette::claude_flags_store::resolve_for_repo(db, defs, Some(repo_id)) {
                Ok(v) => v,
                Err(e) => {
                    tracing::warn!(
                        target: "claudette::ws",
                        repo_id = %repo_id,
                        error = %e,
                        "failed to resolve claude flags"
                    );
                    Vec::new()
                }
            }
        }
        Err(msg) => {
            tracing::warn!(target: "claudette::ws", error = %msg, "claude flag discovery failed");
            Vec::new()
        }
    }
}
