//! The workspace's side of remote connections: serving this device's
//! projects and threads to other devices, and showing other devices'
//! projects and threads in the sidebar.

use std::{
    collections::{HashMap, HashSet},
    hash::{Hash as _, Hasher as _},
    path::PathBuf,
    rc::Rc,
    sync::atomic::{AtomicU64, Ordering},
    time::{Duration, SystemTime},
};

use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
use gpui_kit::component::{WindowExt as _, notification::Notification};
use gpui_kit::*;

use super::{Project, ThreadSettings, Workspace, hostname, initials, tint_for};
use crate::claude::PermissionMode;
use crate::git::GitInfo;
use crate::remote::{
    ClientMessage, ProjectInfo, ServerMessage, ThreadAction, ThreadSummary,
    client::{ClientEvent, ConnectionStatus, RemoteClient},
    new_key,
    server::{Peer, RemoteServer, ServerEvent},
    shell::{HostShell, ShellEvent},
};
use crate::session::{
    NewSession, PermissionDecision, Reasoning, RemoteLink, Session, from_millis, store_uploads,
};
use crate::projects;
use crate::settings::{self, RemoteHost, Settings, SettingsPatch, WorkspaceMode};
use crate::terminal::RemoteRequest;
use crate::ui::folder_browser;
use crate::ui::metal::MetalButton;
use crate::ui::terminal::TerminalView;

/// How often changed threads are sent to subscribed clients.
pub(super) const FLUSH: Duration = Duration::from_millis(150);

/// A connection to another device's app.
pub(super) struct Environment {
    pub(super) host: RemoteHost,
    pub(super) status: ConnectionStatus,
    pub(super) client: Rc<RemoteClient>,
    /// Its environment settings, once it shared them.
    pub(super) settings: Option<Settings>,
    _events: Task<()>,
}

/// One client's interest in a thread: the hashes of the transcript items it
/// was last sent, or `None` before the first send.
type Subscription = Option<Vec<u64>>;

#[derive(Default)]
pub(super) struct RemoteState {
    server: Option<RemoteServer>,
    /// The port and key the running server was started with.
    server_config: Option<(u16, String)>,
    server_error: Option<String>,
    peers: Vec<Peer>,
    subscriptions: HashMap<u64, HashMap<String, Subscription>>,
    dirty_threads: HashSet<String>,
    state_dirty: bool,
    last_state: String,
    /// The environment settings clients were last sent.
    last_settings: String,
    pub(super) environments: Vec<Environment>,
    /// Shells clients run here, by connection and the client's id for them.
    shells: HashMap<(u64, u64), HostShell>,
    /// This device's terminals whose shells run on another device, by the
    /// connection's id and the terminal's id there.
    pub(super) remote_terminals: HashMap<(String, u64), WeakEntity<TerminalView>>,
    _server_events: Option<Task<()>>,
}

/// What the settings show about remote connections.
pub struct RemoteOverview {
    /// The port this device listens on, when it does.
    pub serving: Option<u16>,
    pub error: Option<String>,
    pub peers: Vec<Peer>,
    /// Each connection to another device, by its id in the settings.
    pub environments: Vec<(String, ConnectionStatus)>,
}

/// An environment the settings view edits: this device or a connected one.
pub struct SettingsEnvironment {
    /// The connection's id in the settings; `None` for this device.
    pub id: Option<String>,
    pub name: String,
    pub connected: bool,
    /// `None` until the device shared them; older versions never do.
    pub settings: Option<Settings>,
    /// Its projects, by name and path there.
    pub projects: Vec<(String, PathBuf)>,
}

impl Workspace {
    /// Starts, stops or restarts the server and the connections to match the
    /// settings.
    pub(super) fn sync_remote(&mut self, window: &mut Window, cx: &mut Context<Self>) {
        let (serve, port, mut key, hosts) = settings::get(|settings| {
            (settings.remote_server, settings.remote_port, settings.remote_key.clone(), settings.remotes.clone())
        });
        if serve && key.is_empty() {
            key = new_key();
            let saved = key.clone();
            settings::update(|settings| settings.remote_key = saved);
        }
        let wanted = serve.then(|| (port, key));
        if wanted != self.remote.server_config {
            self.remote.server = None;
            self.remote._server_events = None;
            self.remote.peers.clear();
            self.remote.subscriptions.clear();
            self.remote.shells.clear();
            self.remote.server_error = None;
            self.remote.server_config = wanted.clone();
            if let Some((port, key)) = wanted {
                match RemoteServer::start(port, key, hostname()) {
                    Ok(server) => {
                        let events = server.events.clone();
                        self.remote._server_events = Some(cx.spawn_in(window, async move |this, cx| {
                            while let Ok(event) = events.recv().await {
                                if this.update_in(cx, |this, window, cx| this.on_server_event(event, window, cx)).is_err() {
                                    break;
                                }
                            }
                        }));
                        self.remote.server = Some(server);
                        self.remote.state_dirty = true;
                    }
                    Err(err) => self.remote.server_error = Some(format!("Couldn't listen on port {port}: {err}")),
                }
            }
        }

        let wanted: Vec<RemoteHost> = hosts.into_iter().filter(|host| host.enabled && !host.address.trim().is_empty()).collect();
        let stale: Vec<String> = self
            .remote
            .environments
            .iter()
            .filter(|env| !wanted.iter().any(|host| *host == env.host))
            .map(|env| env.host.id.clone())
            .collect();
        for id in stale {
            self.drop_environment(&id, window, cx);
        }
        for host in wanted {
            if !self.remote.environments.iter().any(|env| env.host.id == host.id) {
                self.connect_environment(host, window, cx);
            }
        }
        cx.notify();
    }

    fn connect_environment(&mut self, host: RemoteHost, window: &mut Window, cx: &mut Context<Self>) {
        let client = Rc::new(RemoteClient::connect(&host.address, host.key.clone(), hostname()));
        let events = client.events.clone();
        let id = host.id.clone();
        let task = cx.spawn_in(window, async move |this, cx| {
            while let Ok(event) = events.recv().await {
                let id = id.clone();
                if this.update_in(cx, |this, window, cx| this.on_client_event(&id, event, window, cx)).is_err() {
                    break;
                }
            }
        });
        self.remote.environments.push(Environment {
            host,
            status: ConnectionStatus::Connecting,
            client,
            settings: None,
            _events: task,
        });
    }

    /// Disconnects from a device and takes its projects and threads away.
    fn drop_environment(&mut self, id: &str, window: &mut Window, cx: &mut Context<Self>) {
        self.remote.remote_terminals.retain(|(env, _), _| env != id);
        self.forget_devices_link(id, cx);
        self.remote.environments.retain(|env| env.host.id != id);
        let removed: Vec<Entity<Session>> = self
            .projects
            .iter()
            .filter(|project| project.environment.as_deref() == Some(id))
            .flat_map(|project| project.sessions.clone())
            .collect();
        for session in &removed {
            if let Some(ix) = self.project_of(session) {
                self.remove_thread(ix, session, cx);
            }
        }
        let active = self.active.clone();
        self.projects.retain(|project| project.environment.as_deref() != Some(id));
        self.active = active.filter(|session| self.project_of(session).is_some());
        match &self.active {
            Some(session) => self.active_project = self.project_of(session).unwrap_or(0),
            None => {
                self.active_project = 0;
                if let Some((ix, next)) = self.threads(cx).0.into_iter().next() {
                    self.select(ix, next, window, cx);
                }
            }
        }
    }

    /// This device's remote server, while it serves.
    pub(super) fn remote_server(&self) -> Option<&RemoteServer> {
        self.remote.server.as_ref()
    }

    pub fn remote_overview(&self) -> RemoteOverview {
        RemoteOverview {
            serving: self.remote.server.as_ref().map(|server| server.port),
            error: self.remote.server_error.clone(),
            peers: self.remote.peers.clone(),
            environments: self.remote.environments.iter().map(|env| (env.host.id.clone(), env.status.clone())).collect(),
        }
    }

    /// This device, then every connection, with the settings each one keeps.
    pub fn settings_environments(&self) -> Vec<SettingsEnvironment> {
        let projects = |env: Option<&str>| -> Vec<(String, PathBuf)> {
            let scratch = super::scratch_dir();
            self.projects
                .iter()
                .filter(|project| project.environment.as_deref() == env && Some(&project.path) != scratch.as_ref())
                .map(|project| (project.name.to_string(), project.path.clone()))
                .collect()
        };
        let this_device = SettingsEnvironment {
            id: None,
            name: hostname(),
            connected: true,
            settings: Some(settings::load()),
            projects: projects(None),
        };
        std::iter::once(this_device)
            .chain(self.remote.environments.iter().map(|env| SettingsEnvironment {
                id: Some(env.host.id.clone()),
                name: env.host.label(),
                connected: matches!(env.status, ConnectionStatus::Connected(_)),
                settings: env.settings.clone(),
                projects: projects(Some(&env.host.id)),
            }))
            .collect()
    }

    /// Changes an environment's settings: this device's (`None`) at once,
    /// another's by asking it. The caller tells the open conversations.
    pub fn write_settings(&mut self, env: Option<&str>, patch: SettingsPatch, cx: &mut Context<Self>) {
        if patch.is_empty() {
            return;
        }
        match env {
            None => settings::apply(&patch),
            Some(env) => {
                let Some(environment) = self.remote.environments.iter_mut().find(|candidate| candidate.host.id == env) else {
                    return;
                };
                environment.client.send(&ClientMessage::UpdateSettings { patch: patch.clone() });
                // Show the change now; the device confirms it with its settings.
                if let Some(settings) = environment.settings.as_mut() {
                    settings.apply_patch(&patch);
                }
            }
        }
        cx.notify();
    }

    /// The first time a device shares its settings and they differ from
    /// this device's, asks whether to give it this device's settings or
    /// leave its own as they are.
    fn offer_settings(&mut self, env: &str, window: &mut Window, cx: &mut Context<Self>) {
        let Some(environment) = self.remote.environments.iter_mut().find(|candidate| candidate.host.id == env) else { return };
        let Some(theirs) = environment.settings.as_ref().map(Settings::environment_values) else { return };
        // Asked once, whatever the answer.
        environment.host.settings_offered = true;
        let name = environment.host.label();
        settings::update(|settings| {
            if let Some(host) = settings.remotes.iter_mut().find(|host| host.id == env) {
                host.settings_offered = true;
            }
        });
        let mut ours = settings::get(Settings::environment_values);
        ours.remove("project_overrides");
        let differing: Vec<String> =
            ours.iter().filter(|(key, value)| theirs.get(*key) != Some(value)).map(|(key, _)| key.clone()).collect();
        if differing.is_empty() {
            return;
        }
        let list = crate::ui::settings::row_titles(&differing).join(", ");
        let workspace = cx.weak_entity();
        let env = env.to_string();
        window.open_dialog(cx, move |dialog, _, cx| {
            let p = crate::ui::metal::palette(cx);
            let (workspace, env, ours) = (workspace.clone(), env.clone(), ours.clone());
            let text = |text: String| div().text_size(px(13.)).line_height(px(19.)).text_color(p.ink2).child(text);
            dialog
                .title(format!("Use your settings on {name}?"))
                .w(px(480.))
                .child(
                    gpui_kit::component::v_flex()
                        .gap(px(10.))
                        .child(text(format!(
                            "{name} is connected. Its settings for new threads, worktrees and source control differ from this device's: {list}."
                        )))
                        .child(text(format!(
                            "Use my settings gives {name} this device's. Leave as is keeps {name}'s own. Either way, you can change them later in Settings with {name} picked."
                        ))),
                )
                .footer(
                    gpui_kit::component::h_flex()
                        .w_full()
                        .justify_end()
                        .gap(px(8.))
                        .child(
                            MetalButton::new("offer-settings-leave")
                                .ghost()
                                .compact()
                                .label("Leave as is")
                                .on_click(|_, window, cx| window.close_dialog(cx)),
                        )
                        .child(MetalButton::new("offer-settings-use").primary().compact().label("Use my settings").on_click(
                            move |_, window, cx| {
                                window.close_dialog(cx);
                                let patch = SettingsPatch { values: ours.clone(), ..Default::default() };
                                let _ = workspace.update(cx, |workspace, cx| workspace.write_settings(Some(&env), patch, cx));
                            },
                        )),
                )
        });
    }

    /// Sends connected devices this device's environment settings, if they
    /// changed since they were last sent.
    pub(super) fn broadcast_settings(&mut self) {
        let Some(server) = &self.remote.server else { return };
        let values = settings::get(Settings::environment_values);
        let text = serde_json::to_string(&values).unwrap_or_default();
        if text != self.remote.last_settings {
            self.remote.last_settings = text;
            server.broadcast(&ServerMessage::Settings { values });
        }
    }

    /// The settings a thread in this project starts with: its environment's,
    /// with the project's overrides.
    pub(super) fn project_settings(&self, project_ix: usize) -> Settings {
        let Some(project) = self.projects.get(project_ix) else { return settings::load() };
        match &project.environment {
            None => settings::get(|settings| settings.for_project(&project.path)),
            Some(env) => self
                .remote
                .environments
                .iter()
                .find(|candidate| candidate.host.id == *env)
                .and_then(|env| env.settings.as_ref())
                .map(|settings| settings.for_project(&project.path))
                // Until the device says, its threads start as this one's would.
                .unwrap_or_else(settings::load),
        }
    }

    pub fn disconnect_peer(&mut self, conn: u64, cx: &mut Context<Self>) {
        if let Some(server) = &self.remote.server {
            server.disconnect(conn);
        }
        self.remote.peers.retain(|peer| peer.conn != conn);
        self.remote.subscriptions.remove(&conn);
        self.remote.shells.retain(|(owner, _), _| *owner != conn);
        cx.notify();
    }

    /// Notes a change to a local thread, for the clients watching.
    pub(super) fn remote_thread_changed(&mut self, session: &Entity<Session>, cx: &App) {
        if self.remote.server.is_none() {
            return;
        }
        let session = session.read(cx);
        if !session.is_remote() {
            self.remote.dirty_threads.insert(session.id.clone());
            self.remote.state_dirty = true;
        }
    }

    /// The local thread with a given id.
    pub(super) fn local_thread(&self, id: &str, cx: &App) -> Option<(usize, Entity<Session>)> {
        self.projects.iter().enumerate().filter(|(_, project)| project.environment.is_none()).find_map(|(ix, project)| {
            project.sessions.iter().find(|session| session.read(cx).id == id).map(|session| (ix, session.clone()))
        })
    }

    // ---- serving ------------------------------------------------------------

    fn host_state(&self, cx: &App) -> ServerMessage {
        let local = self.projects.iter().filter(|project| project.environment.is_none());
        let projects = local
            .clone()
            .map(|project| {
                let git = project.git.clone();
                ProjectInfo {
                    path: project.path.clone(),
                    name: project.name.to_string(),
                    git: git.is_some(),
                    branch: git.as_ref().and_then(|git| git.branch.clone()),
                    branches: git.as_ref().map(|git| git.branches.clone()).unwrap_or_default(),
                    added: git.as_ref().map_or(0, |git| git.added),
                    removed: git.as_ref().map_or(0, |git| git.removed),
                }
            })
            .collect();
        let threads = local
            .flat_map(|project| &project.sessions)
            .map(|session| session.read(cx))
            .filter(|session| !session.is_blank())
            .map(Session::summary)
            .collect();
        ServerMessage::State { projects, threads }
    }

    fn on_server_event(&mut self, event: ServerEvent, window: &mut Window, cx: &mut Context<Self>) {
        let Some(server) = &self.remote.server else { return };
        match event {
            ServerEvent::Connected { conn, name, address } => {
                server.send(conn, &self.host_state(cx));
                server.send(conn, &ServerMessage::Settings { values: settings::get(Settings::environment_values) });
                self.remote.peers.push(Peer { conn, name: name.clone(), address });
                self.remote.subscriptions.insert(conn, HashMap::new());
                if settings::get(|settings| settings.in_app_notifications) {
                    window.push_notification(Notification::info(format!("{name} connected.")), cx);
                }
                cx.notify();
            }
            ServerEvent::Disconnected { conn } => {
                self.forget_device_watcher(conn);
                self.remote.peers.retain(|peer| peer.conn != conn);
                self.remote.subscriptions.remove(&conn);
                self.remote.shells.retain(|(owner, _), _| *owner != conn);
                cx.notify();
            }
            ServerEvent::Message { conn, message } => self.on_client_message(conn, message, window, cx),
        }
    }

    fn on_client_message(&mut self, conn: u64, message: ClientMessage, window: &mut Window, cx: &mut Context<Self>) {
        match message {
            ClientMessage::Hello { .. } => {}
            ClientMessage::Device { request, call } => self.answer_device_call(conn, request, call, cx),
            ClientMessage::Subscribe { thread } => {
                self.remote.subscriptions.entry(conn).or_default().insert(thread.clone(), None);
                self.remote.dirty_threads.insert(thread);
            }
            ClientMessage::Unsubscribe { thread } => {
                if let Some(threads) = self.remote.subscriptions.get_mut(&conn) {
                    threads.remove(&thread);
                }
            }
            ClientMessage::Action { thread, action } => {
                let Some((_, session)) = self.local_thread(&thread, cx) else {
                    self.reply(conn, ServerMessage::Error { message: "That thread is gone.".into() });
                    return;
                };
                self.run_action(session, action, window, cx);
            }
            ClientMessage::Create {
                reference,
                project,
                instance,
                provider,
                model,
                permission_mode,
                reasoning,
                fast_mode,
                worktree,
                base,
                text,
                attachments,
            } => {
                let Some(project_ix) =
                    self.projects.iter().position(|candidate| candidate.environment.is_none() && candidate.path == project)
                else {
                    self.reply(conn, ServerMessage::Error { message: format!("{} isn't a project here.", project.display()) });
                    return;
                };
                let thread_settings = ThreadSettings {
                    instance,
                    provider,
                    model,
                    mode: PermissionMode::from_cli(&permission_mode).unwrap_or_default(),
                    reasoning: Reasoning::from_key(&reasoning).unwrap_or_default(),
                    fast_mode,
                };
                let mode = if worktree { WorkspaceMode::Worktree } else { WorkspaceMode::Local };
                let plan = self.worktree_plan(project_ix, mode).map(|mut plan| {
                    if base.is_some() {
                        plan.base = base;
                    }
                    plan
                });
                let session = self.spawn_thread(project_ix, thread_settings, plan, window, cx);
                let paths = store_uploads(attachments);
                session.update(cx, |session, cx| session.send_message(text, paths, cx));
                let id = session.read(cx).id.clone();
                self.remote.subscriptions.entry(conn).or_default().insert(id.clone(), None);
                self.remote.dirty_threads.insert(id.clone());
                self.remote.state_dirty = true;
                self.reply(conn, ServerMessage::Created { reference, thread: id });
            }
            ClientMessage::UpdateSettings { patch } => {
                settings::apply(&patch);
                // Tells the open conversations, and every client, this one too.
                self.settings_changed(window, cx);
            }
            ClientMessage::ListFolders { path } => {
                let home = folder_browser::home_dir();
                let dir = path.clone();
                let listing = cx.background_executor().spawn(async move { folder_browser::list(&dir, &home) });
                cx.spawn(async move |this, cx| {
                    let (folders, truncated, error) = match listing.await {
                        Ok((folders, truncated)) => (folders, truncated, None),
                        Err(error) => (Vec::new(), false, Some(error)),
                    };
                    let _ = this.update(cx, |this, _| this.reply(conn, ServerMessage::Folders { path, folders, truncated, error }));
                })
                .detach();
            }
            ClientMessage::AddProject { source } => {
                let saved = settings::get(|settings| settings.add_project_dir.clone());
                let base = folder_browser::start_dir(Some(&saved));
                let home = folder_browser::home_dir();
                let added = cx.background_executor().spawn(async move { projects::resolve(&source, &base, &home) });
                cx.spawn(async move |this, cx| {
                    let result = added.await;
                    let _ = this.update(cx, |this, cx| match result {
                        Ok(path) => {
                            this.ensure_project(path.clone(), None);
                            this.remote.state_dirty = true;
                            cx.notify();
                            this.reply(conn, ServerMessage::ProjectAdded { path });
                        }
                        Err(message) => this.reply(conn, ServerMessage::Error { message }),
                    });
                })
                .detach();
            }
            ClientMessage::TerminalOpen { terminal, thread, project, columns, lines } => {
                self.open_host_shell(conn, terminal, thread, project, columns, lines, cx)
            }
            ClientMessage::TerminalInput { terminal, data } => {
                if let (Some(shell), Ok(bytes)) = (self.remote.shells.get(&(conn, terminal)), BASE64.decode(data)) {
                    shell.write(bytes);
                }
            }
            ClientMessage::TerminalResize { terminal, columns, lines } => {
                if let Some(shell) = self.remote.shells.get(&(conn, terminal)) {
                    shell.resize(columns, lines);
                }
            }
            ClientMessage::TerminalClose { terminal } => {
                self.remote.shells.remove(&(conn, terminal));
            }
        }
    }

    /// Starts a shell for a client's terminal, in the thread's folder here.
    #[allow(clippy::too_many_arguments)]
    fn open_host_shell(
        &mut self,
        conn: u64,
        terminal: u64,
        thread: Option<String>,
        project: PathBuf,
        columns: u16,
        lines: u16,
        cx: &mut Context<Self>,
    ) {
        let Some(server) = &self.remote.server else { return };
        let sender = server.sender(conn);
        let exited = |error: String| ServerMessage::TerminalExited { terminal, error: Some(error) };
        let is_project = self.projects.iter().any(|candidate| candidate.environment.is_none() && candidate.path == project);
        let cwd = match thread.as_deref().map(|id| self.local_thread(id, cx)) {
            Some(Some((_, session))) => session.read(cx).cwd.clone(),
            Some(None) => return sender.send(&exited("That thread is gone.".into())),
            None if is_project => project,
            None => return sender.send(&exited(format!("{} isn't a project here.", project.display()))),
        };
        if !cwd.is_dir() {
            return sender.send(&exited(format!("{} isn't there anymore.", cwd.display())));
        }
        let output = sender.clone();
        let started = HostShell::spawn(&cwd, columns, lines, move |event| match event {
            ShellEvent::Output(bytes) => output.send(&ServerMessage::TerminalOutput { terminal, data: BASE64.encode(bytes) }),
            ShellEvent::Exited => output.send(&ServerMessage::TerminalExited { terminal, error: None }),
        });
        match started {
            Ok(shell) => {
                self.remote.shells.insert((conn, terminal), shell);
            }
            Err(err) => sender.send(&exited(format!("Couldn't start the shell: {err}"))),
        }
    }

    fn reply(&self, conn: u64, message: ServerMessage) {
        if let Some(server) = &self.remote.server {
            server.send(conn, &message);
        }
    }

    /// Carries out what a client asked of a local thread.
    fn run_action(&mut self, session: Entity<Session>, action: ThreadAction, window: &mut Window, cx: &mut Context<Self>) {
        match action {
            ThreadAction::Send { text, attachments } => {
                let paths = store_uploads(attachments);
                session.update(cx, |session, cx| session.send_message(text, paths, cx));
            }
            ThreadAction::Queue { text, attachments } => {
                let paths = store_uploads(attachments);
                session.update(cx, |session, cx| session.queue_message(text, paths, cx));
            }
            ThreadAction::SteerQueued { ix } => session.update(cx, |session, cx| session.steer_queued(ix, cx)),
            ThreadAction::RemoveQueued { ix } => session.update(cx, |session, cx| session.remove_queued(ix, cx)),
            ThreadAction::Interrupt => session.update(cx, |session, cx| session.interrupt(cx)),
            ThreadAction::Permission { ix, decision } => {
                let decision = match decision.as_str() {
                    "always" => PermissionDecision::AllowAlways,
                    "deny" => PermissionDecision::Deny,
                    _ => PermissionDecision::Allow,
                };
                session.update(cx, |session, cx| session.respond_permission(ix, decision, cx));
            }
            ThreadAction::Answer { ix, option } => session.update(cx, |session, cx| session.answer_question(ix, option, cx)),
            ThreadAction::AskSubmit { ix, answers } => session.update(cx, |session, cx| session.ask_submit(ix, answers, cx)),
            ThreadAction::AskSkip { ix } => session.update(cx, |session, cx| session.ask_skip(ix, cx)),
            ThreadAction::Background { ix } => session.update(cx, |session, cx| session.background_tool(ix, cx)),
            ThreadAction::StopTask { task_id } => session.update(cx, |session, cx| session.stop_task(&task_id, cx)),
            ThreadAction::SetModel { model } => session.update(cx, |session, cx| session.set_model(model, cx)),
            ThreadAction::SetPermissionMode { mode } => {
                if let Some(mode) = PermissionMode::from_cli(&mode) {
                    session.update(cx, |session, cx| session.set_permission_mode(mode, cx));
                }
            }
            ThreadAction::SetReasoning { reasoning } => {
                if let Some(reasoning) = Reasoning::from_key(&reasoning) {
                    session.update(cx, |session, cx| session.set_reasoning(reasoning, cx));
                }
            }
            ThreadAction::SetFastMode { on } => session.update(cx, |session, cx| session.set_fast_mode(on, cx)),
            ThreadAction::Settle => self.settle_threads(vec![session], window, cx),
            ThreadAction::Unsettle => session.update(cx, |session, cx| session.unsettle(cx)),
            ThreadAction::Pin { pinned } => session.update(cx, |session, cx| session.set_pinned(pinned, cx)),
            ThreadAction::Snooze { until } => session.update(cx, |session, cx| session.snooze(until.map(from_millis), cx)),
            ThreadAction::Rename { title } => session.update(cx, |session, cx| session.rename(title, cx)),
            ThreadAction::CancelResume => session.update(cx, |session, cx| session.cancel_resume(cx)),
            ThreadAction::ScheduleResume { at } => {
                session.update(cx, |session, cx| session.schedule_resume(from_millis(at), cx))
            }
            ThreadAction::Delete => self.forget_thread(session, window, cx),
        }
    }

    /// Sends clients the thread list when it changed, and each subscribed
    /// thread that changed from the first item that did.
    pub(super) fn flush_remote(&mut self, cx: &mut Context<Self>) {
        let Some(server) = &self.remote.server else { return };
        if std::mem::take(&mut self.remote.state_dirty) {
            let state = self.host_state(cx);
            let text = serde_json::to_string(&state).unwrap_or_default();
            if text != self.remote.last_state {
                self.remote.last_state = text;
                server.broadcast(&state);
            }
        }
        let dirty = std::mem::take(&mut self.remote.dirty_threads);
        if dirty.is_empty() {
            return;
        }
        let mut snapshots = HashMap::new();
        for (conn, threads) in &mut self.remote.subscriptions {
            for (id, sent) in threads.iter_mut() {
                if !dirty.contains(id) && sent.is_some() {
                    continue;
                }
                let Some(session) = self
                    .projects
                    .iter()
                    .filter(|project| project.environment.is_none())
                    .flat_map(|project| &project.sessions)
                    .find(|session| session.read(cx).id == *id)
                else {
                    continue;
                };
                let (summary, detail, items, hashes) = snapshots
                    .entry(id.clone())
                    .or_insert_with(|| {
                        let session = session.read(cx);
                        let items = session.saved_items(0);
                        let hashes: Vec<u64> = items.iter().map(item_hash).collect();
                        (session.summary(), session.detail(), items, hashes)
                    })
                    .clone();
                let previous = sent.take().unwrap_or_default();
                let from = previous.iter().zip(&hashes).take_while(|(a, b)| a == b).count();
                server.send(
                    *conn,
                    &ServerMessage::Thread { summary, detail, from, len: items.len(), items: items[from..].to_vec() },
                );
                *sent = Some(hashes);
            }
        }
    }

    // ---- connecting to other devices -------------------------------------------

    fn on_client_event(&mut self, env: &str, event: ClientEvent, window: &mut Window, cx: &mut Context<Self>) {
        match event {
            ClientEvent::Status(status) => {
                let connected = matches!(status, ConnectionStatus::Connected(_));
                if !connected {
                    // A shell there ends with the connection.
                    self.stop_remote_terminals(env, "The connection to the other device dropped; this terminal's shell ended.", cx);
                }
                if let Some(environment) = self.remote.environments.iter_mut().find(|candidate| candidate.host.id == env) {
                    environment.status = status;
                }
                if connected {
                    if let Some(link) = self.devices_link(env) {
                        link.watch();
                    }
                    // A new connection knows nothing of what this one watched.
                    for session in self.environment_sessions(env) {
                        session.update(cx, |session, _| {
                            if let Some(remote) = session.remote.as_mut()
                                && remote.subscribed
                            {
                                remote.subscribed = false;
                                session.subscribe_remote();
                            }
                        });
                    }
                }
                cx.notify();
            }
            ClientEvent::Message(ServerMessage::State { projects, threads }) => {
                self.reconcile_environment(env, projects, threads, window, cx)
            }
            ClientEvent::Message(ServerMessage::Thread { summary, detail, from, len, items }) => {
                if let Some(session) = self.remote_session(env, &summary.id, cx) {
                    session.update(cx, |session, cx| session.apply_remote(&summary, detail, from, len, items, cx));
                }
            }
            ClientEvent::Message(ServerMessage::Created { reference, thread }) => {
                let session = self.environment_sessions(env).into_iter().find(|session| {
                    session.read(cx).remote.as_ref().is_some_and(|remote| remote.creating == Some(reference))
                });
                if let Some(session) = session {
                    session.update(cx, |session, cx| session.remote_created(thread, cx));
                }
            }
            ClientEvent::Message(ServerMessage::Error { message }) => {
                let name = self.environment_name(env);
                window.push_notification(Notification::error(message).title(name), cx);
            }
            ClientEvent::Message(ServerMessage::Settings { values }) => {
                if let Some(environment) = self.remote.environments.iter_mut().find(|candidate| candidate.host.id == env) {
                    environment.settings = Some(settings::with_values(&Settings::default(), &values));
                    if !environment.host.settings_offered {
                        self.offer_settings(env, window, cx);
                    }
                }
                cx.notify();
            }
            ClientEvent::Message(ServerMessage::Folders { path, folders, truncated, error }) => {
                if let Some((palette, _)) = &self.palette {
                    let folders = error.map_or(Ok((folders, truncated)), Err);
                    palette.update(cx, |palette, cx| palette.receive_folders(env, &path, folders, cx));
                }
            }
            ClientEvent::Message(ServerMessage::ProjectAdded { path }) => {
                let ix = self.ensure_remote_project(env, path, None);
                self.new_thread(ix, window, cx);
            }
            ClientEvent::Message(ServerMessage::Welcome { .. } | ServerMessage::Denied { .. }) => {}
            ClientEvent::Message(ServerMessage::DeviceReply { request, ok, error }) => {
                if let Some(link) = self.devices_link(env) {
                    link.resolve(request, error.map_or_else(|| Ok(ok.unwrap_or_default()), Err));
                }
            }
            ClientEvent::Message(ServerMessage::DeviceState { state }) => self.remote_device_state(env, state, cx),
            ClientEvent::Message(ServerMessage::DeviceOpened { thread, device }) => {
                if let Some(session) = self.remote_session(env, &thread, cx) {
                    let local = session.read(cx).id.clone();
                    self.agent_opened_device(&local, device, window, cx);
                }
            }
            ClientEvent::Message(ServerMessage::DeviceClosed { thread, device }) => {
                if let Some(session) = self.remote_session(env, &thread, cx) {
                    let local = session.read(cx).id.clone();
                    self.agent_closed_device(&local, device.as_deref(), cx);
                }
            }
            ClientEvent::Message(ServerMessage::TerminalOutput { terminal, data }) => {
                let view = self.remote.remote_terminals.get(&(env.to_string(), terminal)).and_then(WeakEntity::upgrade);
                match (view, BASE64.decode(data)) {
                    (Some(view), Ok(bytes)) => view.update(cx, |view, cx| view.receive(&bytes, cx)),
                    (None, _) => {
                        // A terminal closed here whose shell still talks.
                        self.remote.remote_terminals.remove(&(env.to_string(), terminal));
                        self.send_to_environment(env, &ClientMessage::TerminalClose { terminal });
                    }
                    _ => {}
                }
            }
            ClientEvent::Message(ServerMessage::TerminalExited { terminal, error }) => {
                if let Some(view) = self.remote.remote_terminals.remove(&(env.to_string(), terminal)).and_then(|view| view.upgrade()) {
                    view.update(cx, |view, cx| view.remote_exited(error, cx));
                }
            }
        }
    }

    fn send_to_environment(&self, env: &str, message: &ClientMessage) {
        if let Some(environment) = self.remote.environments.iter().find(|candidate| candidate.host.id == env) {
            environment.client.send(message);
        }
    }

    /// A terminal for a thread on another device, whose shell starts there
    /// in the thread's folder. `None` while the device isn't connected.
    pub(super) fn open_remote_terminal(
        &mut self,
        session: &Entity<Session>,
        label: String,
        window: &mut Window,
        cx: &mut Context<Self>,
    ) -> Option<Entity<TerminalView>> {
        static NEXT_TERMINAL: AtomicU64 = AtomicU64::new(1);
        let (env, thread, project) = {
            let session = session.read(cx);
            let remote = session.remote.as_ref()?;
            (remote.env.clone(), remote.thread.clone(), session.project.clone())
        };
        let environment = self.remote.environments.iter().find(|candidate| candidate.host.id == env)?;
        if !matches!(environment.status, ConnectionStatus::Connected(_)) {
            return None;
        }
        let client = environment.client.clone();
        let terminal = NEXT_TERMINAL.fetch_add(1, Ordering::Relaxed);
        let size = crate::ui::terminal::INITIAL_SIZE;
        client.send(&ClientMessage::TerminalOpen {
            terminal,
            thread,
            project,
            columns: size.columns as u16,
            lines: size.lines as u16,
        });
        let link = move |request: RemoteRequest| {
            client.send(&match request {
                RemoteRequest::Input(bytes) => ClientMessage::TerminalInput { terminal, data: BASE64.encode(bytes) },
                RemoteRequest::Resize { columns, lines } => ClientMessage::TerminalResize { terminal, columns, lines },
                RemoteRequest::Close => ClientMessage::TerminalClose { terminal },
            });
        };
        let view = cx.new(|cx| TerminalView::remote(label, link, window, cx));
        self.remote.remote_terminals.insert((env, terminal), view.downgrade());
        Some(view)
    }

    /// Stops a device's terminals, whose shells are out of reach now.
    fn stop_remote_terminals(&mut self, env: &str, reason: &str, cx: &mut Context<Self>) {
        let ids: Vec<(String, u64)> = self.remote.remote_terminals.keys().filter(|(owner, _)| owner == env).cloned().collect();
        for id in ids {
            if let Some(view) = self.remote.remote_terminals.remove(&id).and_then(|view| view.upgrade()) {
                view.update(cx, |view, cx| view.stop(reason.to_string(), cx));
            }
        }
    }

    /// Asks another device for a folder's listing, for the Add project palette.
    pub(super) fn request_folders(&self, env: &str, path: String) {
        if let Some(environment) = self.remote.environments.iter().find(|candidate| candidate.host.id == env) {
            environment.client.send(&ClientMessage::ListFolders { path });
        }
    }

    pub(super) fn environment_sessions(&self, env: &str) -> Vec<Entity<Session>> {
        self.projects
            .iter()
            .filter(|project| project.environment.as_deref() == Some(env))
            .flat_map(|project| project.sessions.clone())
            .collect()
    }

    pub(super) fn remote_session(&self, env: &str, id: &str, cx: &App) -> Option<Entity<Session>> {
        self.environment_sessions(env).into_iter().find(|session| {
            session.read(cx).remote.as_ref().and_then(|remote| remote.thread.as_deref()) == Some(id)
        })
    }

    pub(super) fn environment_name(&self, env: &str) -> String {
        self.remote
            .environments
            .iter()
            .find(|candidate| candidate.host.id == env)
            .map(|env| env.host.label())
            .unwrap_or_else(|| env.to_string())
    }

    /// The link a new mirrored thread on `env` uses.
    pub(super) fn remote_link(&self, env: &str, thread: Option<String>) -> Option<RemoteLink> {
        let environment = self.remote.environments.iter().find(|candidate| candidate.host.id == env)?;
        Some(RemoteLink {
            env: env.to_string(),
            thread,
            client: environment.client.clone(),
            subscribed: false,
            loaded: false,
            attention: false,
            creating: None,
        })
    }

    /// The index of a device's project, adding it if needed.
    fn ensure_remote_project(&mut self, env: &str, path: PathBuf, name: Option<&str>) -> usize {
        if let Some(ix) =
            self.projects.iter().position(|project| project.environment.as_deref() == Some(env) && project.path == path)
        {
            return ix;
        }
        let name = name.map(str::to_string).unwrap_or_else(|| {
            path.file_name().map(|name| name.to_string_lossy().to_string()).unwrap_or_else(|| path.display().to_string())
        });
        self.projects.push(Project {
            path,
            initials: initials(&name).into(),
            tint: tint_for(&name),
            name: name.into(),
            added: SystemTime::now(),
            sessions: Vec::new(),
            git: None,
            environment: Some(env.to_string()),
            repo: None,
        });
        self.projects.len() - 1
    }

    /// Brings a device's projects and threads in line with its latest list.
    fn reconcile_environment(
        &mut self,
        env: &str,
        projects: Vec<ProjectInfo>,
        threads: Vec<ThreadSummary>,
        window: &mut Window,
        cx: &mut Context<Self>,
    ) {
        for info in &projects {
            let ix = self.ensure_remote_project(env, info.path.clone(), Some(&info.name));
            self.projects[ix].git = info.git.then(|| GitInfo {
                branch: info.branch.clone(),
                branches: info.branches.clone(),
                added: info.added,
                removed: info.removed,
            });
        }
        let known: HashSet<&str> = threads.iter().map(|summary| summary.id.as_str()).collect();
        for session in self.environment_sessions(env) {
            let gone = session
                .read(cx)
                .remote
                .as_ref()
                .and_then(|remote| remote.thread.as_deref())
                .is_some_and(|id| !known.contains(id));
            if gone && let Some(ix) = self.project_of(&session) {
                let was_active = self.active.as_ref() == Some(&session);
                self.remove_thread(ix, &session, cx);
                if was_active {
                    self.active = None;
                    if let Some((ix, next)) = self.threads(cx).0.into_iter().next() {
                        self.select(ix, next, window, cx);
                    }
                }
            }
        }
        for summary in threads {
            match self.remote_session(env, &summary.id, cx) {
                Some(session) => session.update(cx, |session, cx| {
                    session.apply_summary(&summary);
                    cx.notify();
                }),
                None => {
                    let Some(link) = self.remote_link(env, Some(summary.id.clone())) else { continue };
                    let project_ix = self.ensure_remote_project(env, summary.project.clone(), None);
                    let session = cx.new(|cx| Session::remote(&summary, link, cx));
                    self.attach(project_ix, session, window, cx);
                }
            }
        }
        cx.notify();
    }

    /// A new, empty thread for one of a device's projects.
    pub(super) fn new_remote_session(
        &mut self,
        env: &str,
        project: PathBuf,
        settings: ThreadSettings,
        worktree: bool,
        cx: &mut Context<Self>,
    ) -> Option<Entity<Session>> {
        let link = self.remote_link(env, None)?;
        let plan = worktree.then(|| crate::worktree::WorktreePlan {
            project: project.clone(),
            base: None,
            from_origin: false,
            submodules: settings::Submodules::Skip,
        });
        let new = NewSession {
            instance: settings.instance,
            provider: settings.provider,
            project,
            model: settings.model,
            permission_mode: settings.mode,
            reasoning: settings.reasoning,
            fast_mode: settings.fast_mode,
            worktree: plan,
        };
        Some(cx.new(|_| Session::new_remote(new, link)))
    }
}

fn item_hash<T: serde::Serialize>(item: &T) -> u64 {
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    serde_json::to_string(item).unwrap_or_default().hash(&mut hasher);
    hasher.finish()
}