//! 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,
    time::{Duration, SystemTime},
};

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},
};
use crate::session::{
    NewSession, PermissionDecision, Reasoning, RemoteLink, Session, from_millis, store_uploads,
};
use crate::settings::{self, RemoteHost, Settings, SettingsPatch, WorkspaceMode};
use crate::ui::metal::MetalButton;

/// 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>,
    _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.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.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);
                }
            }
        }
    }

    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);
        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.
    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.remote.peers.retain(|peer| peer.conn != conn);
                self.remote.subscriptions.remove(&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::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);
            }
        }
    }

    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 let Some(environment) = self.remote.environments.iter_mut().find(|candidate| candidate.host.id == env) {
                    environment.status = status;
                }
                if connected {
                    // 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::Welcome { .. } | ServerMessage::Denied { .. }) => {}
        }
    }

    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()
    }

    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()
}