//! 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.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);
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.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::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 {
// 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::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 });
}
}
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()
}