vectrumcodePublic

Merge remote device streaming: a remote thread's Device panel uses its device's simulators

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Browse Files

JjohannSocommitted 4 days ago2 parents 388a3aa 93a7f0ccommit3a7cb6c

Showing 14 changed files with 832 additions and 100 deletions.

Merge commit — showing changes against the first parent.

+4−0README.mdView File

@@ -159,6 +159,10 @@ Gruvbox, Rosé Pine, Solarized and Dracula.
159	159	panel with Auto-show floating preview off), and its activity shows in the timeline.
160	160	Both tools are npm packages the app installs at pinned versions into its data folder and
161	161	runs with your Node; quitting stops them, while simulators and emulators keep running.
	162	+ A thread on another device uses that device's simulators and emulators: its Device panel
	163	+ lists, starts and controls them over the remote connection, and the video and touches go
	164	+ through the other device's remote server, which passes only the stream routes on to its
	165	+ hub, and only with its key.
162	166	- A terminal drawer under the conversation, like T3 Code's (Ctrl+J, or the panel toggle in
163	167	the chat header): each thread has its own terminals, which start in the thread's folder
164	168	(its worktree when it has one), on the device the thread runs on. Drag the drawer's top edge to resize it. Split a terminal

+2−2src/device/actions.rsView File

@@ -138,7 +138,7 @@ fn run_ios(ready: &Ready, udid: &str, action: &Action) -> Result<()> {
138	138	}
139	139	Action::ClearLocation => simctl(udid, &["location", "clear"], "location")?,
140	140	Action::SetPermission { app_id, permission, decision } => {
141		- if *permission == "notifications" {
	141	+ if permission == "notifications" {
142	142	// simctl has no notification verb; serve-sim's CLI edits the plist for it.
143	143	let Some(cli) = &ready.helpers.serve_sim_cli else {
144	144	bail!("Device permission requires a helper missing from this install. Set up device support again.");
@@ -243,7 +243,7 @@ fn run_android(serial: &str, action: &Action) -> Result<()> {
243	243	if permissions.is_empty() {
244	244	return Err(unsupported(action, Platform::Android));
245	245	}
246		- let verb = if *decision == "grant" { "grant" } else { "revoke" };
	246	+ let verb = if decision == "grant" { "grant" } else { "revoke" };
247	247	for permission in permissions {
248	248	// Not every app declares every permission of a group.
249	249	let _ = shell(serial, &["pm", verb, app_id, permission], "permission");

+20−11src/device/mod.rsView File

@@ -167,7 +167,8 @@ pub struct DeviceSession {
167	167	}
168	168
169	169	/// Everything the Device panel renders from.
170		- #[derive(Clone, Debug, Default)]
	170	+ #[derive(Clone, Debug, Default, Serialize, Deserialize)]
	171	+ #[serde(default)]
171	172	pub struct State {
172	173	pub enabled: bool,
173	174	pub agent_access: bool,
@@ -228,7 +229,7 @@ impl State {
228	229	}
229	230	}
230	231
231		- #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
	232	+ #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
232	233	pub enum TextSize {
233	234	Small,
234	235	#[default]
@@ -250,7 +251,7 @@ impl TextSize {
250	251	}
251	252	}
252	253
253		- #[derive(Clone, Copy, Debug, PartialEq, Eq)]
	254	+ #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
254	255	pub enum ColorFilter {
255	256	None,
256	257	Grayscale,
@@ -288,7 +289,7 @@ impl ColorFilter {
288	289	}
289	290	}
290	291
291		- #[derive(Clone, Copy, Debug, PartialEq, Eq)]
	292	+ #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
292	293	pub enum Orientation {
293	294	Portrait,
294	295	LandscapeLeft,
@@ -332,7 +333,7 @@ impl Orientation {
332	333	}
333	334	}
334	335
335		- #[derive(Clone, Copy, Debug, PartialEq, Eq)]
	336	+ #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
336	337	pub enum Toggle {
337	338	ReduceMotion,
338	339	IncreaseContrast,
@@ -343,7 +344,8 @@ pub enum Toggle {
343	344	}
344	345
345	346	/// Current values as read from the device; `None` is unsupported or unread.
346		- #[derive(Clone, Debug, Default, PartialEq)]
	347	+ #[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
	348	+ #[serde(default)]
347	349	pub struct Settings {
348	350	pub dark: Option<bool>,
349	351	pub text_size: Option<TextSize>,
@@ -370,20 +372,21 @@ impl Settings {
370	372	}
371	373	}
372	374
373		- #[derive(Clone, Debug, PartialEq)]
	375	+ #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
374	376	pub struct ForegroundApp {
375	377	pub id: String,
376	378	}
377	379
378	380	/// One device's settings and foreground app, as last read.
379		- #[derive(Clone, Debug, PartialEq)]
	381	+ #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
380	382	pub struct Detail {
381	383	pub settings: Settings,
382	384	pub foreground: Option<ForegroundApp>,
383	385	}
384	386
385	387	/// One control in the Tools drawer or the rail.
386		- #[derive(Clone, Debug, PartialEq)]
	388	+ #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
	389	+ #[serde(rename_all = "snake_case")]
387	390	pub enum Action {
388	391	SetAppearance(bool),
389	392	SetTextSize(TextSize),
@@ -394,7 +397,7 @@ pub enum Action {
394	397	SetLocation { latitude: f64, longitude: f64 },
395	398	ClearLocation,
396	399	/// `permission` is a shared name (`camera`), `decision` grant, revoke or reset.
397		- SetPermission { app_id: String, permission: &'static str, decision: &'static str },
	400	+ SetPermission { app_id: String, permission: String, decision: String },
398	401	OpenUrl(String),
399	402	LaunchApp(String),
400	403	TerminateApp(String),
@@ -624,6 +627,12 @@ impl DeviceService {
624	627	}
625	628	}
626	629
	630	+ /// The hub's origin for another device's stream: running, or started
	631	+ /// when device support is on.
	632	+ pub fn hub_for_remote(&self) -> Option<String> {
	633	+ self.hub_origin().or_else(\|\| self.enabled().then(\|\| self.readiness().ok().map(\|ready\| ready.origin)).flatten())
	634	+ }
	635	+
627	636	/// The hub's origin if it runs, without starting it.
628	637	pub fn hub_origin(&self) -> Option<String> {
629	638	self.host.current().map(\|ready\| ready.origin)
@@ -1030,7 +1039,7 @@ impl DeviceService {
1030	1039	}
1031	1040
1032	1041	/// One accessibility element, normalized to the displayed screen.
1033		- #[derive(Clone, Debug, PartialEq)]
	1042	+ #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
1034	1043	pub struct AxElement {
1035	1044	pub label: String,
1036	1045	pub x: f32,

+59−21src/device/stream.rsView File

@@ -131,6 +131,41 @@ struct Shared {
131	131	generation: std::sync::atomic::AtomicU64,
132	132	}
133	133
	134	+ /// Where a device hub answers: this machine's on loopback, or another
	135	+ /// device's through its remote server, which passes allowlisted stream
	136	+ /// routes on to its hub for anyone with its key.
	137	+ #[derive(Clone, Debug, PartialEq)]
	138	+ pub struct HubTarget {
	139	+ /// `host:port`.
	140	+ pub address: String,
	141	+ /// Put before every route: `/device-hub` through a remote server.
	142	+ pub prefix: String,
	143	+ /// The remote server's key, sent as a bearer token.
	144	+ pub key: Option<String>,
	145	+ }
	146	+
	147	+ impl HubTarget {
	148	+ /// This machine's hub at its loopback origin, `http://127.0.0.1:<port>`.
	149	+ pub fn local(origin: &str) -> Self {
	150	+ Self { address: origin.trim_start_matches("http://").to_string(), prefix: String::new(), key: None }
	151	+ }
	152	+
	153	+ fn connect(&self, timeout: Duration) -> Result<TcpStream> {
	154	+ use std::net::ToSocketAddrs as _;
	155	+ let address: SocketAddr =
	156	+ self.address.to_socket_addrs()?.next().ok_or_else(\|\| anyhow!("{} can't be reached", self.address))?;
	157	+ let tcp = TcpStream::connect_timeout(&address, timeout)?;
	158	+ tcp.set_nodelay(true)?;
	159	+ Ok(tcp)
	160	+ }
	161	+
	162	+ /// The request head of a GET, with the key when there is one.
	163	+ fn get_head(&self, path: &str, extra: &str) -> String {
	164	+ let auth = self.key.as_ref().map(\|key\| format!("Authorization: Bearer {key}\r\n")).unwrap_or_default();
	165	+ format!("GET {}{path} HTTP/1.1\r\nHost: {}\r\n{auth}{extra}Connection: close\r\n\r\n", self.prefix, self.address)
	166	+ }
	167	+ }
	168	+
134	169	/// One device's live screen, as its panel watches and drives it.
135	170	pub struct DeviceStream {
136	171	shared: Arc<Shared>,
@@ -139,7 +174,7 @@ pub struct DeviceStream {
139	174
140	175	impl DeviceStream {
141	176	/// Starts the stream's thread. Nothing connects until `set_watching(true)`.
142		- pub fn new(origin: String, platform: Platform, device: String, events: smol::channel::Sender<StreamEvent>) -> Self {
	177	+ pub fn new(hub: HubTarget, platform: Platform, device: String, events: smol::channel::Sender<StreamEvent>) -> Self {
143	178	let shared = Arc::new(Shared {
144	179	stop: AtomicBool::new(false),
145	180	watching: AtomicBool::new(false),
@@ -150,7 +185,7 @@ impl DeviceStream {
150	185	let (input, input_rx) = mpsc::channel();
151	186	let thread_shared = shared.clone();
152	187	let _ = thread::Builder::new().name("device-stream".into()).spawn(move \|\| {
153		- let stream = Streamer { origin, device, shared: thread_shared, events, input: input_rx };
	188	+ let stream = Streamer { hub, device, shared: thread_shared, events, input: input_rx };
154	189	match platform {
155	190	Platform::Android => stream.run_android(),
156	191	Platform::Ios => stream.run_ios(),
@@ -203,7 +238,7 @@ impl Drop for DeviceStream {
203	238	}
204	239
205	240	struct Streamer {
206		- origin: String,
	241	+ hub: HubTarget,
207	242	device: String,
208	243	shared: Arc<Shared>,
209	244	events: smol::channel::Sender<StreamEvent>,
@@ -235,10 +270,6 @@ impl Streamer {
235	270	let _ = self.events.send_blocking(event);
236	271	}
237	272
238		- fn port(&self) -> u16 {
239		- self.origin.rsplit(':').next().and_then(\|port\| port.parse().ok()).unwrap_or(80)
240		- }
241		-
242	273	/// Keeps a connection up while the panel shows: `connect` runs one
243	274	/// connection and says how it ended.
244	275	fn supervise(&self, mut connect: impl FnMut(&Self, u64) -> End) {
@@ -271,12 +302,22 @@ impl Streamer {
271	302	}
272	303
273	304	fn socket(&self, path: &str) -> Result<WebSocket<TcpStream>> {
274		- let address = SocketAddr::from(([127, 0, 0, 1], self.port()));
275		- let tcp = TcpStream::connect_timeout(&address, Duration::from_secs(3))?;
276		- tcp.set_nodelay(true)?;
277		- tcp.set_read_timeout(Some(Duration::from_secs(5)))?;
278		- let url = format!("ws://127.0.0.1:{}{path}", self.port());
279		- let (socket, _) = tungstenite::client(url, tcp).map_err(\|err\| anyhow!("stream handshake failed: {err}"))?;
	305	+ use tungstenite::client::IntoClientRequest as _;
	306	+ let tcp = self.hub.connect(Duration::from_secs(4))?;
	307	+ tcp.set_read_timeout(Some(Duration::from_secs(8)))?;
	308	+ let mut request = format!("ws://{}{}{path}", self.hub.address, self.hub.prefix).into_client_request()?;
	309	+ if let Some(key) = &self.hub.key {
	310	+ request.headers_mut().insert("Authorization", format!("Bearer {key}").parse()?);
	311	+ }
	312	+ let (socket, _) = tungstenite::client(request, tcp).map_err(\|err\| match err {
	313	+ tungstenite::HandshakeError::Failure(tungstenite::Error::Http(response)) if response.status() == 401 => {
	314	+ anyhow!("The other device refused the key.")
	315	+ }
	316	+ tungstenite::HandshakeError::Failure(tungstenite::Error::Http(response)) if response.status() == 503 => {
	317	+ anyhow!("Device support isn't running on the other device.")
	318	+ }
	319	+ err => anyhow!("stream handshake failed: {err}"),
	320	+ })?;
280	321	socket.get_ref().set_read_timeout(Some(POLL))?;
281	322	Ok(socket)
282	323	}
@@ -419,8 +460,8 @@ impl Streamer {
419	460	let video_done = Arc::new(AtomicBool::new(false));
420	461	let (video_tx, video_rx) = mpsc::channel::<VideoMessage>();
421	462	{
422		- let (port, device, done) = (self.port(), device.to_string(), video_done.clone());
423		- thread::spawn(move \|\| read_mjpeg(port, &device, &done, &video_tx));
	463	+ let (hub, device, done) = (self.hub.clone(), device.to_string(), video_done.clone());
	464	+ thread::spawn(move \|\| read_mjpeg(&hub, &device, &done, &video_tx));
424	465	}
425	466	let mut socket: Option<WebSocket<TcpStream>> = None;
426	467	let mut admitted = false;
@@ -546,14 +587,11 @@ enum VideoMessage {
546	587	}
547	588
548	589	/// Reads serve-sim's MJPEG body and passes each JPEG on.
549		- fn read_mjpeg(port: u16, device: &str, done: &AtomicBool, frames: &mpsc::Sender<VideoMessage>) {
	590	+ fn read_mjpeg(hub: &HubTarget, device: &str, done: &AtomicBool, frames: &mpsc::Sender<VideoMessage>) {
550	591	let result = (\|\| -> Result<()> {
551		- let address = SocketAddr::from(([127, 0, 0, 1], port));
552		- let mut tcp = TcpStream::connect_timeout(&address, Duration::from_secs(3))?;
	592	+ let mut tcp = hub.connect(Duration::from_secs(4))?;
553	593	tcp.set_read_timeout(Some(Duration::from_millis(500)))?;
554		- let request = format!(
555		- "GET /vendor/serve-sim/helper/{device}/stream.mjpeg HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\nAccept: multipart/x-mixed-replace\r\nConnection: close\r\n\r\n"
556		- );
	594	+ let request = hub.get_head(&format!("/vendor/serve-sim/helper/{device}/stream.mjpeg"), "Accept: multipart/x-mixed-replace\r\n");
557	595	std::io::Write::write_all(&mut tcp, request.as_bytes())?;
558	596	let mut buffer: Vec<u8> = Vec::with_capacity(512 * 1024);
559	597	let mut chunk = vec![0u8; 64 * 1024];

+16−0src/remote/client.rsView File

@@ -42,7 +42,23 @@ pub struct RemoteClient {
42	42	pub events: smol::channel::Receiver<ClientEvent>,
43	43	}
44	44
	45	+ /// A way to the host that works off the UI thread.
	46	+ #[derive(Clone)]
	47	+ pub struct ClientSender(mpsc::Sender<String>);
	48	+
	49	+ impl ClientSender {
	50	+ pub fn send(&self, message: &ClientMessage) {
	51	+ if let Ok(text) = serde_json::to_string(message) {
	52	+ let _ = self.0.send(text);
	53	+ }
	54	+ }
	55	+ }
	56	+
45	57	impl RemoteClient {
	58	+ pub fn sender(&self) -> ClientSender {
	59	+ ClientSender(self.outbox.clone())
	60	+ }
	61	+
46	62	/// Connects to `address` (`host` or `host:port`) and keeps reconnecting
47	63	/// until dropped.
48	64	pub fn connect(address: &str, key: String, name: String) -> Self {

+163−0src/remote/device_proxy.rsaddedView File

++ //! Device streams for other devices, as T3 Code passes them through its
++ //! environment server. A client's Device panel connects to this device's
++ //! remote server under `/device-hub/…` with the key; the request goes on to
++ //! the device hub on loopback and the connection is spliced through, so
++ //! video (H.264 over a WebSocket, or MJPEG) and touches flow straight
++ //! between the two.
++ //!
++ //! The hub never faces the network itself: it has routes that run commands.
++ //! Only the stream and input routes pass, and only GETs (WebSocket upgrades
++ //! are GETs too). Everything else a client does with devices goes through
++ //! typed messages on the remote connection.
++ use std::{
++     io::{Read as _, Write as _},
++     net::{Shutdown, TcpStream},
++     thread,
++     time::Duration,
++ };
++ use super::keys_match;
++ /// Where device routes start on the remote server.
++ pub const PREFIX: &str = "/device-hub";
++ /// The most a request head may take.
++ const MAX_HEAD: usize = 16 * 1024;
++ /// Looks at a new connection's request head without taking it, and says
++ /// whether it's for the device hub.
++ pub fn is_device_request(stream: &TcpStream) -> bool {
++     let mut buffer = [0u8; 512];
++     let deadline = std::time::Instant::now() + Duration::from_secs(5);
++     loop {
++         match stream.peek(&mut buffer) {
++             Ok(0) => return false,
++             Ok(n) => {
++                 let head = String::from_utf8_lossy(&buffer[..n]);
++                 if let Some(line) = head.split("\r\n").next().filter(|_| head.contains("\r\n")) {
++                     return line.split_whitespace().nth(1).is_some_and(|path| path.starts_with(&format!("{PREFIX}/")));
++                 }
++                 if n == buffer.len() || std::time::Instant::now() > deadline {
++                     return false;
++                 }
++                 thread::sleep(Duration::from_millis(10));
++             }
++             Err(_) => return false,
++         }
++     }
++ }
++ /// Whether a hub route may be reached from another device.
++ fn allowed(path: &str) -> bool {
++     let route = path.split('?').next().unwrap_or_default();
++     match route {
++         // Android: H.264 out, gestures in.
++         "/vendor/serve-emu/ws" => true,
++         // iOS: input.
++         "/vendor/serve-sim/helper/ws" => true,
++         // iOS: video, `/vendor/serve-sim/helper/<udid>/stream.mjpeg`.
++         _ => route
++             .strip_prefix("/vendor/serve-sim/helper/")
++             .and_then(|rest| rest.strip_suffix("/stream.mjpeg"))
++             .is_some_and(|udid| !udid.is_empty() && !udid.contains('/')),
++     }
++ }
++ fn respond(stream: &mut TcpStream, status: &str) {
++     let body = status.split_once(' ').map(|(_, text)| text).unwrap_or(status);
++     let _ = write!(stream, "HTTP/1.1 {status}\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len());
++     let _ = stream.flush();
++ }
++ /// Serves one device request: checks the key and the route, then splices
++ /// the connection through to the hub. `hub` finds the hub's loopback
++ /// origin, starting it if device support is on.
++ pub fn serve(mut stream: TcpStream, key: &str, hub: impl FnOnce() -> Option<String>) {
++     let _ = stream.set_read_timeout(Some(Duration::from_secs(10)));
++     let Some(head) = read_head(&mut stream) else { return };
++     let mut lines = head.split("\r\n");
++     let request_line = lines.next().unwrap_or_default();
++     let mut parts = request_line.split_whitespace();
++     let (method, path, version) = (parts.next().unwrap_or_default(), parts.next().unwrap_or_default(), parts.next().unwrap_or("HTTP/1.1"));
++     let headers: Vec<(&str, &str)> = lines
++         .filter(|line| !line.is_empty())
++         .filter_map(|line| line.split_once(':').map(|(name, value)| (name.trim(), value.trim())))
++         .collect();
++     let offered = headers
++         .iter()
++         .find(|(name, _)| name.eq_ignore_ascii_case("authorization"))
++         .and_then(|(_, value)| value.strip_prefix("Bearer "))
++         .unwrap_or_default();
++     if !keys_match(offered, key) {
++         // Slow down anyone guessing keys.
++         thread::sleep(Duration::from_secs(1));
++         return respond(&mut stream, "401 Unauthorized");
++     }
++     let route = path.strip_prefix(PREFIX).unwrap_or_default();
++     if method != "GET" || !allowed(route) {
++         return respond(&mut stream, "403 Forbidden");
++     }
++     let Some(origin) = hub() else { return respond(&mut stream, "503 Service Unavailable") };
++     let address = origin.trim_start_matches("http://").to_string();
++     let Ok(mut upstream) = TcpStream::connect(&address) else { return respond(&mut stream, "502 Bad Gateway") };
++     let _ = upstream.set_nodelay(true);
++     let _ = stream.set_nodelay(true);
++     // The same request, for the hub: its own path, its own host, no key.
++     let mut forwarded = format!("{method} {route} {version}\r\n");
++     for (name, value) in &headers {
++         if name.eq_ignore_ascii_case("authorization") || name.eq_ignore_ascii_case("host") {
++             continue;
++         }
++         forwarded.push_str(&format!("{name}: {value}\r\n"));
++     }
++     forwarded.push_str(&format!("Host: {address}\r\n\r\n"));
++     if upstream.write_all(forwarded.as_bytes()).is_err() {
++         return respond(&mut stream, "502 Bad Gateway");
++     }
++     splice(stream, upstream);
++ }
++ /// Reads the request head, up to and including its blank line.
++ fn read_head(stream: &mut TcpStream) -> Option<String> {
++     let mut head = Vec::new();
++     let mut byte = [0u8; 1];
++     while !head.ends_with(b"\r\n\r\n") {
++         if head.len() >= MAX_HEAD || stream.read(&mut byte).ok()? == 0 {
++             return None;
++         }
++         head.push(byte[0]);
++     }
++     Some(String::from_utf8_lossy(&head).into_owned())
++ }
++ /// Copies both ways until either side hangs up.
++ fn splice(client: TcpStream, hub: TcpStream) {
++     let _ = client.set_read_timeout(None);
++     let (Ok(mut client_read), Ok(mut hub_write)) = (client.try_clone(), hub.try_clone()) else { return };
++     let upward = thread::spawn(move || {
++         let _ = std::io::copy(&mut client_read, &mut hub_write);
++         let _ = hub_write.shutdown(Shutdown::Both);
++     });
++     let (mut hub_read, mut client_write) = (hub, client);
++     let _ = std::io::copy(&mut hub_read, &mut client_write);
++     let _ = client_write.shutdown(Shutdown::Both);
++     let _ = upward.join();
++ }
++ #[cfg(test)]
++ mod tests {
++     use super::allowed;
++     #[::core::prelude::v1::test]
++     fn passes_only_stream_routes() {
++         assert!(allowed("/vendor/serve-emu/ws?device=emulator-5554&frame-meta=1"));
++         assert!(allowed("/vendor/serve-sim/helper/ws?device=ABC"));
++         assert!(allowed("/vendor/serve-sim/helper/ABC-123/stream.mjpeg"));
++         assert!(!allowed("/vendor/serve-sim/helper/a/b/stream.mjpeg"));
++         assert!(!allowed("/vendor/serve-sim/api"));
++         assert!(!allowed("/vendor/serve-sim/exec"));
++         assert!(!allowed("/api/devices/boot"));
++         assert!(!allowed("/vendor/serve-emu/api/screenshot?device=x"));
++     }
++ }

+57−0src/remote/mod.rsView File

@@ -16,8 +16,13 @@
16	16	//! shells on the host: [`ClientMessage::TerminalOpen`] starts one in a
17	17	//! thread's folder there, keys go over as [`ClientMessage::TerminalInput`],
18	18	//! and the shell's output comes back as [`ServerMessage::TerminalOutput`].
	19	+ //! A client's Device panel uses the host's simulators and emulators: it asks
	20	+ //! with [`ClientMessage::Device`], gets [`ServerMessage::DeviceReply`], and
	21	+ //! follows [`ServerMessage::DeviceState`]; the video and touches go through
	22	+ //! the host's [`device_proxy`] on the same port.
19	23
20	24	pub mod client;
	25	+ pub mod device_proxy;
21	26	pub mod server;
22	27	pub mod shell;
23	28
@@ -124,6 +129,25 @@ pub struct Upload {
124	129	pub data: String,
125	130	}
126	131
	132	+ /// What a client asks of the host's devices. Threads are the host's ids.
	133	+ #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
	134	+ #[serde(tag = "call", rename_all = "snake_case")]
	135	+ pub enum DeviceCall {
	136	+ /// Send the device state now and whenever it changes.
	137	+ Watch,
	138	+ /// Refresh the devices, starting the hub once device support is on.
	139	+ List,
	140	+ Configure { enabled: Option<bool>, agent_access: Option<bool>, onboarding: Option<bool> },
	141	+ Retry,
	142	+ Open { thread: String, device: String, boot: bool },
	143	+ Close { thread: String, device: Option<String>, shutdown: bool },
	144	+ Detail { device: String },
	145	+ Action { device: String, action: crate::device::Action },
	146	+ /// A PNG of the screen, base64.
	147	+ Screenshot { device: String },
	148	+ Accessibility { device: String },
	149	+ }
	150	+
127	151	/// What a client asks of a thread on the host.
128	152	#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
129	153	#[serde(tag = "action", rename_all = "snake_case")]
@@ -195,6 +219,9 @@ pub enum ClientMessage {
195	219	TerminalResize { terminal: u64, columns: u16, lines: u16 },
196	220	/// Hangs up on a terminal's shell.
197	221	TerminalClose { terminal: u64 },
	222	+ /// Something for the host's devices; `request` pairs it with its reply.
	223	+ /// Older hosts don't answer.
	224	+ Device { request: u64, call: DeviceCall },
198	225	}
199	226
200	227	#[derive(Clone, Debug, Serialize, Deserialize)]
@@ -219,6 +246,14 @@ pub enum ServerMessage {
219	246	TerminalOutput { terminal: u64, data: String },
220	247	/// A terminal's shell exited, or with `error`, couldn't start.
221	248	TerminalExited { terminal: u64, error: Option<String> },
	249	+ /// The answer to a [`ClientMessage::Device`]: its result or what went wrong.
	250	+ DeviceReply { request: u64, ok: Option<serde_json::Value>, error: Option<String> },
	251	+ /// The host's devices, for clients that watch them.
	252	+ DeviceState { state: crate::device::State },
	253	+ /// An agent on the host opened a device in one of its threads.
	254	+ DeviceOpened { thread: String, device: crate::device::DeviceSummary },
	255	+ /// An agent closed a thread's device, or all of them without one.
	256	+ DeviceClosed { thread: String, device: Option<String> },
222	257	}
223	258
224	259	/// A new random key: 24 bytes, hex.
@@ -324,6 +359,28 @@ mod tests {
324	359	serde_json::from_str::<ServerMessage>(&text).unwrap(),
325	360	ServerMessage::TerminalExited { terminal: 3, error: None }
326	361	));
	362	+ let message = ClientMessage::Device {
	363	+ request: 4,
	364	+ call: DeviceCall::Action {
	365	+ device: "emulator-5554".into(),
	366	+ action: crate::device::Action::SetPermission {
	367	+ app_id: "com.example".into(),
	368	+ permission: "camera".into(),
	369	+ decision: "grant".into(),
	370	+ },
	371	+ },
	372	+ };
	373	+ let text = serde_json::to_string(&message).unwrap();
	374	+ assert_eq!(serde_json::from_str::<ClientMessage>(&text).unwrap(), message);
	375	+ let message = ClientMessage::Device { request: 5, call: DeviceCall::Action { device: "d".into(), action: crate::device::Action::SetAppearance(true) } };
	376	+ let text = serde_json::to_string(&message).unwrap();
	377	+ assert_eq!(serde_json::from_str::<ClientMessage>(&text).unwrap(), message);
	378	+ let state = crate::device::State { enabled: true, host_status: crate::device::HostStatus::Ready, ..Default::default() };
	379	+ let text = serde_json::to_string(&ServerMessage::DeviceState { state }).unwrap();
	380	+ match serde_json::from_str::<ServerMessage>(&text).unwrap() {
	381	+ ServerMessage::DeviceState { state } => assert_eq!(state.host_status, crate::device::HostStatus::Ready),
	382	+ _ => panic!("not a device state"),
	383	+ }
327	384	let values = crate::settings::Settings::default().environment_values();
328	385	let text = serde_json::to_string(&ServerMessage::Settings { values: values.clone() }).unwrap();
329	386	match serde_json::from_str::<ServerMessage>(&text).unwrap() {

+5−0src/remote/server.rsView File

@@ -137,6 +137,11 @@ fn serve(
137	137	let _ = stream.set_nonblocking(false);
138	138	let _ = stream.set_nodelay(true);
139	139	let _ = stream.set_read_timeout(Some(HELLO_TIMEOUT));
	140	+ // A client's Device panel streaming one of this device's simulators.
	141	+ if super::device_proxy::is_device_request(&stream) {
	142	+ super::device_proxy::serve(stream, key, \|\| crate::device::service().hub_for_remote());
	143	+ return;
	144	+ }
140	145	let Ok(mut socket) = tungstenite::accept(stream) else { return };
141	146
142	147	let name = match read_message(&mut socket) {

+218−0src/ui/device/backend.rsaddedView File

++ //! Where a Device panel's devices are: on this machine, through the device
++ //! service, or on the device a remote thread runs on, through its remote
++ //! connection. Calls block; make them from a background thread.
++ use std::{
++     collections::HashMap,
++     sync::{
++         Arc, Mutex,
++         atomic::{AtomicU64, Ordering},
++     },
++     time::Duration,
++ };
++ use base64::Engine as _;
++ use gpui_kit::App;
++ use serde_json::Value;
++ use super::DeviceWatch;
++ use crate::device::{self, Action, AxElement, Detail, DeviceSession, HostStatus, State, stream::HubTarget};
++ use crate::remote::{ClientMessage, DeviceCall, client::ClientSender, device_proxy};
++ /// How long a call to another device may take; booting takes the longest.
++ const CALL_TIMEOUT: Duration = Duration::from_secs(240);
++ /// Another device's devices, reached through its remote connection.
++ pub struct RemoteDevices {
++     pub env: String,
++     /// The device's name, shown where this machine would say "This machine".
++     pub name: String,
++     sender: ClientSender,
++     hub: HubTarget,
++     pending: Mutex<HashMap<u64, smol::channel::Sender<Result<Value, String>>>>,
++     next: AtomicU64,
++ }
++ impl RemoteDevices {
++     /// `address` and `key` are the connection's, which the device's stream
++     /// proxy answers on too.
++     pub fn new(env: String, name: String, sender: ClientSender, address: &str, key: String) -> Self {
++         let hub = HubTarget { address: crate::remote::with_port(address), prefix: device_proxy::PREFIX.into(), key: Some(key) };
++         Self { env, name, sender, hub, pending: Mutex::default(), next: AtomicU64::new(1) }
++     }
++     /// Asks the device to send its device state now and on every change.
++     pub fn watch(&self) {
++         self.sender.send(&ClientMessage::Device { request: 0, call: DeviceCall::Watch });
++     }
++     /// Hands a reply to the call waiting for it.
++     pub fn resolve(&self, request: u64, result: Result<Value, String>) {
++         let waiting = self.pending.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).remove(&request);
++         if let Some(waiting) = waiting {
++             let _ = waiting.try_send(result);
++         }
++     }
++     fn call(&self, call: DeviceCall) -> Result<Value, String> {
++         let request = self.next.fetch_add(1, Ordering::Relaxed);
++         let (tx, rx) = smol::channel::bounded(1);
++         self.pending.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).insert(request, tx);
++         self.sender.send(&ClientMessage::Device { request, call });
++         let answer = smol::block_on(async {
++             smol::future::or(async { rx.recv().await.ok() }, async {
++                 smol::Timer::after(CALL_TIMEOUT).await;
++                 None
++             })
++             .await
++         });
++         self.pending.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).remove(&request);
++         answer.unwrap_or_else(|| {
++             Err(format!("{} didn't answer. Its Vectrum Code may be too old for devices, or the connection dropped.", self.name))
++         })
++     }
++     fn call_as<T: serde::de::DeserializeOwned>(&self, call: DeviceCall) -> Result<T, String> {
++         serde_json::from_value(self.call(call)?).map_err(|err| format!("Unreadable answer from {}: {err}", self.name))
++     }
++ }
++ /// A Device panel's devices.
++ #[derive(Clone)]
++ pub enum Backend {
++     Local,
++     Remote(Arc<RemoteDevices>),
++ }
++ fn text(err: anyhow::Error) -> String {
++     format!("{err:#}")
++ }
++ impl Backend {
++     /// The devices as last heard of. Another device's read as starting up
++     /// until it has said.
++     pub fn state(&self, cx: &App) -> State {
++         let watch = cx.global::<DeviceWatch>();
++         match self {
++             Backend::Local => watch.state.clone(),
++             Backend::Remote(remote) => watch.remote.get(&remote.env).cloned().unwrap_or_else(|| State {
++                 enabled: true,
++                 onboarding_completed: true,
++                 host_status: HostStatus::Starting,
++                 ..State::default()
++             }),
++         }
++     }
++     /// Where the device hub's streams are, if it runs.
++     pub fn hub(&self) -> Option<HubTarget> {
++         match self {
++             Backend::Local => device::service().hub_origin().map(|origin| HubTarget::local(&origin)),
++             Backend::Remote(remote) => Some(remote.hub.clone()),
++         }
++     }
++     pub fn list(&self) {
++         match self {
++             Backend::Local => {
++                 device::service().list();
++             }
++             Backend::Remote(remote) => {
++                 let _ = remote.call(DeviceCall::List);
++             }
++         }
++     }
++     pub fn retry(&self) {
++         match self {
++             Backend::Local => {
++                 device::service().retry();
++             }
++             Backend::Remote(remote) => {
++                 let _ = remote.call(DeviceCall::Retry);
++             }
++         }
++     }
++     pub fn configure(&self, enabled: Option<bool>, agent_access: Option<bool>, onboarding: Option<bool>) -> Result<(), String> {
++         match self {
++             Backend::Local => device::service().configure(enabled, agent_access, onboarding).map(|_| ()).map_err(text),
++             Backend::Remote(remote) => remote.call(DeviceCall::Configure { enabled, agent_access, onboarding }).map(|_| ()),
++         }
++     }
++     pub fn open(&self, thread: &str, device: &str, boot: bool) -> Result<DeviceSession, String> {
++         match self {
++             Backend::Local => device::service().open(thread, device, boot).map_err(text),
++             Backend::Remote(remote) => {
++                 remote.call_as(DeviceCall::Open { thread: thread.into(), device: device.into(), boot })
++             }
++         }
++     }
++     pub fn close(&self, thread: &str, device: Option<&str>, shutdown: bool) -> Result<(), String> {
++         match self {
++             Backend::Local => device::service().close(thread, device, shutdown).map_err(text),
++             Backend::Remote(remote) => remote
++                 .call(DeviceCall::Close { thread: thread.into(), device: device.map(str::to_string), shutdown })
++                 .map(|_| ()),
++         }
++     }
++     pub fn detail(&self, device: &str) -> Result<Detail, String> {
++         match self {
++             Backend::Local => device::service().detail(device).map_err(text),
++             Backend::Remote(remote) => remote.call_as(DeviceCall::Detail { device: device.into() }),
++         }
++     }
++     pub fn action(&self, device: &str, action: &Action) -> Result<Detail, String> {
++         match self {
++             Backend::Local => device::service().action(device, action).map_err(text),
++             Backend::Remote(remote) => remote.call_as(DeviceCall::Action { device: device.into(), action: action.clone() }),
++         }
++     }
++     pub fn screenshot(&self, device: &str) -> Result<Vec<u8>, String> {
++         match self {
++             Backend::Local => device::service().screenshot(device).map(|(_, png)| png).map_err(text),
++             Backend::Remote(remote) => {
++                 let data: String = remote.call_as(DeviceCall::Screenshot { device: device.into() })?;
++                 base64::engine::general_purpose::STANDARD.decode(data).map_err(|err| err.to_string())
++             }
++         }
++     }
++     pub fn accessibility(&self, device: &str) -> Result<Vec<AxElement>, String> {
++         match self {
++             Backend::Local => device::service().accessibility(device).map_err(text),
++             Backend::Remote(remote) => remote.call_as(DeviceCall::Accessibility { device: device.into() }),
++         }
++     }
++ }
++ /// Answers a client's device call on the host. Blocks.
++ pub fn answer(call: DeviceCall) -> Result<Value, String> {
++     let service = device::service();
++     let json = |value: Result<Value, serde_json::Error>| value.map_err(|err| err.to_string());
++     match call {
++         DeviceCall::Watch => json(serde_json::to_value(service.state())),
++         DeviceCall::List => json(serde_json::to_value(service.list())),
++         DeviceCall::Retry => json(serde_json::to_value(service.retry())),
++         DeviceCall::Configure { enabled, agent_access, onboarding } => {
++             json(serde_json::to_value(service.configure(enabled, agent_access, onboarding).map_err(text)?))
++         }
++         DeviceCall::Open { thread, device, boot } => json(serde_json::to_value(service.open(&thread, &device, boot).map_err(text)?)),
++         DeviceCall::Close { thread, device, shutdown } => {
++             service.close(&thread, device.as_deref(), shutdown).map_err(text)?;
++             Ok(Value::Null)
++         }
++         DeviceCall::Detail { device } => json(serde_json::to_value(service.detail(&device).map_err(text)?)),
++         DeviceCall::Action { device, action } => json(serde_json::to_value(service.action(&device, &action).map_err(text)?)),
++         DeviceCall::Screenshot { device } => {
++             let (_, png) = service.screenshot(&device).map_err(text)?;
++             Ok(Value::String(base64::engine::general_purpose::STANDARD.encode(png)))
++         }
++         DeviceCall::Accessibility { device } => json(serde_json::to_value(service.accessibility(&device).map_err(text)?)),
++     }
++ }

+49−17src/ui/device/mod.rsView File

@@ -3,9 +3,11 @@
3	3	//! open or start; one with a device shows it live. A device can float over
4	4	//! the chat instead, in a small window of its own.
5	5
	6	+ mod backend;
6	7	mod setup;
7	8	mod view;
8	9
	10	+ pub use backend::{Backend, RemoteDevices, answer as answer_device_call};
9	11	pub use setup::{AGENT_DESCRIPTION, DeviceSetup, HUB_DESCRIPTION, agent_status, hub_status, platform_line};
10	12	pub use view::{DeviceView, DeviceViewEvent};
11	13
@@ -26,16 +28,18 @@ use crate::ui::metal::{self, Led, MetalButton, MetalStyled as _, Palette};
26	28	const SHARE: f32 = 0.42;
27	29	const MIN_WIDTH: f32 = 380.;
28	30
29		- /// The device service's state, as the UI last heard of it.
	31	+ /// The device service's state, as the UI last heard of it, and each
	32	+ /// connected device's, by its connection's id.
30	33	pub struct DeviceWatch {
31	34	pub state: State,
	35	+ pub remote: std::collections::HashMap<String, State>,
32	36	}
33	37
34	38	impl Global for DeviceWatch {}
35	39
36	40	impl DeviceWatch {
37	41	pub fn init(cx: &mut App) {
38		- cx.set_global(DeviceWatch { state: device::service().state() });
	42	+ cx.set_global(DeviceWatch { state: device::service().state(), remote: Default::default() });
39	43	let events = device::service().subscribe();
40	44	cx.spawn(async move \|cx\| {
41	45	while let Ok(event) = events.recv().await {
@@ -50,6 +54,7 @@ impl DeviceWatch {
50	54	.detach();
51	55	}
52	56
	57	+ /// This machine's devices.
53	58	pub fn state(cx: &App) -> &State {
54	59	&cx.global::<DeviceWatch>().state
55	60	}
@@ -117,7 +122,9 @@ struct Tab {
117	122	}
118	123
119	124	pub struct DevicePanel {
	125	+ /// The thread, by the id its devices' machine knows it by.
120	126	thread_id: String,
	127	+ backend: Backend,
121	128	tabs: Vec<Tab>,
122	129	active: u64,
123	130	next_id: u64,
@@ -137,10 +144,11 @@ pub struct DevicePanel {
137	144	impl EventEmitter<DevicePanelEvent> for DevicePanel {}
138	145
139	146	impl DevicePanel {
140		- pub fn new(thread_id: String, _: &mut Window, cx: &mut Context<Self>) -> Self {
	147	+ pub fn new(thread_id: String, backend: Backend, _: &mut Window, cx: &mut Context<Self>) -> Self {
141	148	let subscriptions = vec![cx.observe_global::<DeviceWatch>(\|this, cx\| this.sync(cx))];
142	149	Self {
143	150	thread_id,
	151	+ backend,
144	152	tabs: vec![Tab { id: 1, name: None, view: None }],
145	153	active: 1,
146	154	next_id: 2,
@@ -182,17 +190,31 @@ impl DevicePanel {
182	190
183	191	/// Lists the devices; never installs or starts anything before setup.
184	192	fn refresh(&mut self, cx: &mut Context<Self>) {
185		- let state = DeviceWatch::state(cx);
	193	+ let state = self.backend.state(cx);
186	194	if state.host_status == HostStatus::Disabled \|\| !state.onboarding_completed {
187	195	return;
188	196	}
189	197	self.listed = true;
190		- background(cx, \|\| device::service().list(), \|_, _, _\| {});
	198	+ let backend = self.backend.clone();
	199	+ background(cx, move \|\| backend.list(), \|_, _, _\| {});
	200	+ }
	201	+
	202	+ pub fn backend(&self) -> &Backend {
	203	+ &self.backend
	204	+ }
	205	+
	206	+ pub fn thread_id(&self) -> &str {
	207	+ &self.thread_id
	208	+ }
	209	+
	210	+ /// The thread's id where its devices are, once a new remote thread has one.
	211	+ pub fn set_thread_id(&mut self, thread_id: String) {
	212	+ self.thread_id = thread_id;
191	213	}
192	214
193	215	/// Keeps open views' devices current.
194	216	fn sync(&mut self, cx: &mut Context<Self>) {
195		- let state = DeviceWatch::state(cx).clone();
	217	+ let state = self.backend.state(cx);
196	218	for tab in &self.tabs {
197	219	if let Some(view) = &tab.view {
198	220	let current = view.read(cx).device.clone();
@@ -235,8 +257,9 @@ impl DevicePanel {
235	257	let id = match self.view_for(&device.id, cx) {
236	258	Some(id) => id,
237	259	None => {
238		- let host_label = DeviceWatch::state(cx).host_label(&device.host_id);
239		- let view = cx.new(\|cx\| DeviceView::new(device, host_label, window, cx));
	260	+ let host_label = self.backend.state(cx).host_label(&device.host_id);
	261	+ let backend = self.backend.clone();
	262	+ let view = cx.new(\|cx\| DeviceView::new(device, host_label, backend, window, cx));
240	263	self._subscriptions.push(cx.subscribe_in(&view, window, Self::on_view_event));
241	264	let empty = self.tabs.iter_mut().find(\|tab\| tab.id == self.active && tab.view.is_none());
242	265	match empty {
@@ -290,14 +313,15 @@ impl DevicePanel {
290	313	DeviceViewEvent::PowerOff => {
291	314	let device = view.read(cx).device.clone();
292	315	let thread = self.thread_id.clone();
	316	+ let backend = self.backend.clone();
293	317	self.error = None;
294	318	background(
295	319	cx,
296		- move \|\| device::service().close(&thread, Some(&device.id), true),
	320	+ move \|\| backend.close(&thread, Some(&device.id), true),
297	321	move \|this, result, cx\| match result {
298	322	Ok(()) => this.remove_tab(id, cx),
299	323	Err(err) => {
300		- this.error = Some(format!("{err:#}").into());
	324	+ this.error = Some(err.into());
301	325	cx.notify();
302	326	}
303	327	},
@@ -377,18 +401,19 @@ impl DevicePanel {
377	401	self.pending = Some(device.clone());
378	402	let thread = self.thread_id.clone();
379	403	let id = device.id.clone();
	404	+ let backend = self.backend.clone();
380	405	cx.spawn_in(window, async move \|this, cx\| {
381		- let result = cx.background_executor().spawn(async move { device::service().open(&thread, &id, true) }).await;
	406	+ let result = cx.background_executor().spawn(async move { backend.open(&thread, &id, true) }).await;
382	407	let _ = this.update_in(cx, \|this, window, cx\| {
383	408	this.pending = None;
384	409	match result {
385	410	Ok(session) => {
386		- let opened = device::service().state().device(&session.host_id, &session.device_id).cloned().unwrap_or(device);
	411	+ let opened = this.backend.state(cx).device(&session.host_id, &session.device_id).cloned().unwrap_or(device);
387	412	this.show_device(opened, false, window, cx);
388	413	let (panel, floating) = (this.visible, this.float_visible);
389	414	this.set_visible(panel, floating, window, cx);
390	415	}
391		- Err(err) => this.error = Some(format!("{err:#}").into()),
	416	+ Err(err) => this.error = Some(err.into()),
392	417	}
393	418	cx.notify();
394	419	});
@@ -482,7 +507,7 @@ impl DevicePanel {
482	507
483	508	fn render_body(&self, cx: &mut Context<Self>) -> AnyElement {
484	509	let p = metal::palette(cx);
485		- let state = DeviceWatch::state(cx).clone();
	510	+ let state = self.backend.state(cx);
486	511	let tab = self.tabs.iter().find(\|tab\| tab.id == self.active);
487	512	let view = tab.and_then(\|tab\| tab.view.clone()).filter(\|_\| self.floating != Some(self.active));
488	513	if let Some(view) = view {
@@ -532,7 +557,10 @@ impl DevicePanel {
532	557	.primary()
533	558	.compact()
534	559	.label("Set up devices")
535		- .on_click(\|_, window, cx\| DeviceSetup::open(window, cx)),
	560	+ .on_click({
	561	+ let backend = self.backend.clone();
	562	+ move \|_, window, cx\| DeviceSetup::open(backend.clone(), window, cx)
	563	+ }),
536	564	)
537	565	.into_any_element();
538	566	}
@@ -588,7 +616,10 @@ impl DevicePanel {
588	616	MetalButton::new("device-retry")
589	617	.compact()
590	618	.label("Retry")
591		- .on_click(cx.listener(\|_, _, _, cx\| background(cx, \|\| device::service().retry(), \|_, _, _\| {}))),
	619	+ .on_click(cx.listener(\|this, _, _, cx\| {
	620	+ let backend = this.backend.clone();
	621	+ background(cx, move \|\| backend.retry(), \|_, _, _\| {})
	622	+ })),
592	623	);
593	624	}
594	625	}
@@ -675,7 +706,8 @@ impl DevicePanel {
675	706
676	707	fn render_notices(&self, cx: &mut Context<Self>) -> Vec<AnyElement> {
677	708	let p = metal::palette(cx);
678		- let state = DeviceWatch::state(cx);
	709	+ let state = self.backend.state(cx);
	710	+ let state = &state;
679	711	let mut notices = Vec::new();
680	712	let notice = \|text: String, p: &Palette\| {
681	713	div()

+31−16src/ui/device/setup.rsView File

@@ -7,8 +7,8 @@ use gpui_kit::component::{Icon, IconName, WindowExt as _, h_flex, v_flex};
7	7	use gpui_kit::prelude::FluentBuilder as _;
8	8	use gpui_kit::*;
9	9
10		- use super::{DeviceWatch, background};
11		- use crate::device::{self, HostStatus, Platform, State};
	10	+ use super::{Backend, background};
	11	+ use crate::device::{HostStatus, Platform, State};
12	12	use crate::ui::metal::{self, Led, MetalButton, MetalStyled as _, Palette};
13	13
14	14	pub const HUB_DESCRIPTION: &str =
@@ -27,33 +27,46 @@ enum Pending {
27	27	}
28	28
29	29	pub struct DeviceSetup {
	30	+ backend: Backend,
30	31	step: usize,
31	32	pending: Option<Pending>,
32	33	_watch: Subscription,
33	34	}
34	35
35	36	impl DeviceSetup {
36		- fn new(cx: &mut Context<Self>) -> Self {
37		- Self { step: 0, pending: None, _watch: cx.observe_global::<DeviceWatch>(\|_, cx\| cx.notify()) }
	37	+ fn new(backend: Backend, cx: &mut Context<Self>) -> Self {
	38	+ Self { backend, step: 0, pending: None, _watch: cx.observe_global::<super::DeviceWatch>(\|_, cx\| cx.notify()) }
38	39	}
39	40
40		- /// Opens the dialog. `on_dismiss` runs if it closes before setup is done.
41		- pub fn open(window: &mut Window, cx: &mut App) {
42		- Self::open_with(window, cx, None::<fn(&mut Window, &mut App)>);
	41	+ /// Opens the dialog for the devices `backend` reaches.
	42	+ pub fn open(backend: Backend, window: &mut Window, cx: &mut App) {
	43	+ Self::open_with(backend, window, cx, None::<fn(&mut Window, &mut App)>);
43	44	}
44	45
45		- pub fn open_with(window: &mut Window, cx: &mut App, on_dismiss: Option<impl Fn(&mut Window, &mut App) + 'static>) {
46		- let setup = cx.new(Self::new);
	46	+ /// Opens the dialog. `on_dismiss` runs if it closes before setup is done.
	47	+ pub fn open_with(
	48	+ backend: Backend,
	49	+ window: &mut Window,
	50	+ cx: &mut App,
	51	+ on_dismiss: Option<impl Fn(&mut Window, &mut App) + 'static>,
	52	+ ) {
	53	+ let title = match &backend {
	54	+ Backend::Local => "Set up devices".to_string(),
	55	+ Backend::Remote(remote) => format!("Set up devices on {}", remote.name),
	56	+ };
	57	+ let done_backend = backend.clone();
	58	+ let setup = cx.new(\|cx\| Self::new(backend, cx));
47	59	let on_dismiss = on_dismiss.map(\|f\| std::rc::Rc::new(f) as std::rc::Rc<dyn Fn(&mut Window, &mut App)>);
48	60	window.open_dialog(cx, move \|dialog, _, _\| {
49	61	let on_dismiss = on_dismiss.clone();
	62	+ let done_backend = done_backend.clone();
50	63	dialog
51	64	.w(px(560.))
52		- .title("Set up devices")
	65	+ .title(title.clone())
53	66	.child(setup.clone())
54	67	.on_close(move \|_: &ClickEvent, window: &mut Window, cx: &mut App\| {
55	68	let done = {
56		- let state = DeviceWatch::state(cx);
	69	+ let state = done_backend.state(cx);
57	70	state.onboarding_completed && state.host_status != HostStatus::Disabled
58	71	};
59	72	if !done && let Some(on_dismiss) = &on_dismiss {
@@ -74,8 +87,9 @@ impl DeviceSetup {
74	87	) {
75	88	self.pending = Some(kind);
76	89	cx.notify();
	90	+ let backend = self.backend.clone();
77	91	cx.spawn_in(window, async move \|this, cx\| {
78		- let result = cx.background_executor().spawn(async move { device::service().configure(enabled, agent, done) }).await;
	92	+ let result = cx.background_executor().spawn(async move { backend.configure(enabled, agent, done) }).await;
79	93	let _ = this.update_in(cx, \|this, window, cx\| {
80	94	this.pending = None;
81	95	cx.notify();
@@ -90,7 +104,8 @@ impl DeviceSetup {
90	104	fn check(&mut self, cx: &mut Context<Self>) {
91	105	self.pending = Some(Pending::Check);
92	106	cx.notify();
93		- background(cx, \|\| device::service().list(), \|this, _, cx\| {
	107	+ let backend = self.backend.clone();
	108	+ background(cx, move \|\| backend.list(), \|this, _, cx\| {
94	109	this.pending = None;
95	110	cx.notify();
96	111	});
@@ -178,7 +193,7 @@ pub fn platform_line(state: &State, platform: Platform, compact: bool, p: &Palet
178	193	impl Render for DeviceSetup {
179	194	fn render(&mut self, _: &mut Window, cx: &mut Context<Self>) -> impl IntoElement {
180	195	let p = metal::palette(cx);
181		- let state = DeviceWatch::state(cx).clone();
	196	+ let state = self.backend.state(cx);
182	197	let enabled = state.host_status != HostStatus::Disabled;
183	198	let busy = state.busy();
184	199	let blocked = busy \|\| self.pending.is_some();
@@ -221,7 +236,7 @@ impl Render for DeviceSetup {
221	236	let weak = cx.weak_entity();
222	237	move \|on, window, cx\| {
223	238	let _ = weak.update(cx, \|this, cx\| {
224		- if this.pending.is_none() && !DeviceWatch::state(cx).busy() {
	239	+ if this.pending.is_none() && !this.backend.state(cx).busy() {
225	240	this.configure(Pending::Hub, Some(on), (!on).then_some(false), None, window, cx);
226	241	}
227	242	});
@@ -259,7 +274,7 @@ impl Render for DeviceSetup {
259	274	let weak = cx.weak_entity();
260	275	move \|on, window, cx\| {
261	276	let _ = weak.update(cx, \|this, cx\| {
262		- let state = DeviceWatch::state(cx);
	277	+ let state = this.backend.state(cx);
263	278	if this.pending.is_none() && !state.busy() && state.host_status != HostStatus::Disabled {
264	279	this.configure(Pending::Agent, None, Some(on), None, window, cx);
265	280	}

+22−16src/ui/device/view.rsView File

@@ -14,7 +14,7 @@ use gpui_kit::component::{
14	14	use gpui_kit::prelude::FluentBuilder as _;
15	15	use gpui_kit::*;
16	16
17		- use super::DeviceWatch;
	17	+ use super::Backend;
18	18	use crate::device::{
19	19	self, Action, AxElement, ColorFilter, Detail, DeviceSummary, Orientation, Platform, TextSize, Toggle,
20	20	stream::{Button, DeviceStream, Frame, Key, Screen, Status, StreamEvent, TouchPhase},
@@ -84,6 +84,7 @@ struct Fields {
84	84	pub struct DeviceView {
85	85	pub device: DeviceSummary,
86	86	host_label: String,
	87	+ backend: Backend,
87	88	stream: Option<DeviceStream>,
88	89	frame: Option<Frame>,
89	90	status: Status,
@@ -116,7 +117,7 @@ pub struct DeviceView {
116	117	impl EventEmitter<DeviceViewEvent> for DeviceView {}
117	118
118	119	impl DeviceView {
119		- pub fn new(device: DeviceSummary, host_label: String, window: &mut Window, cx: &mut Context<Self>) -> Self {
	120	+ pub fn new(device: DeviceSummary, host_label: String, backend: Backend, window: &mut Window, cx: &mut Context<Self>) -> Self {
120	121	let field = \|placeholder: &str, window: &mut Window, cx: &mut Context<Self>\| {
121	122	let placeholder = placeholder.to_string();
122	123	cx.new(\|cx\| InputState::new(window, cx).placeholder(placeholder))
@@ -138,6 +139,7 @@ impl DeviceView {
138	139	Self {
139	140	device,
140	141	host_label,
	142	+ backend,
141	143	stream: None,
142	144	frame: None,
143	145	status: Status::Connecting,
@@ -198,13 +200,13 @@ impl DeviceView {
198	200	if self.stream.is_some() {
199	201	return;
200	202	}
201		- let Some(origin) = device::service().hub_origin() else {
	203	+ let Some(hub) = self.backend.hub() else {
202	204	self.status = Status::Error;
203	205	self.status_detail = Some("The device hub is not running. Refresh devices to start it.".into());
204	206	return;
205	207	};
206	208	let (tx, rx) = smol::channel::unbounded();
207		- self.stream = Some(DeviceStream::new(origin, self.device.platform, self.device.id.clone(), tx));
	209	+ self.stream = Some(DeviceStream::new(hub, self.device.platform, self.device.id.clone(), tx));
208	210	self._events = cx.spawn_in(window, async move \|this, cx\| {
209	211	while let Ok(event) = rx.recv().await {
210	212	let mut batch = vec![event];
@@ -257,8 +259,9 @@ impl DeviceView {
257	259	/// Reads the device's settings; every action reads them again after.
258	260	fn read_detail(&mut self, cx: &mut Context<Self>) {
259	261	let id = self.device.id.clone();
	262	+ let backend = self.backend.clone();
260	263	self._detail = cx.spawn(async move \|this, cx\| {
261		- let result = cx.background_executor().spawn(async move { device::service().detail(&id) }).await;
	264	+ let result = cx.background_executor().spawn(async move { backend.detail(&id) }).await;
262	265	let _ = this.update(cx, \|this, cx\| {
263	266	match result {
264	267	Ok(detail) => {
@@ -267,7 +270,7 @@ impl DeviceView {
267	270	}
268	271	Err(err) => {
269	272	this.detail = None;
270		- this.error = Some(format!("{err:#}").into());
	273	+ this.error = Some(err.into());
271	274	}
272	275	}
273	276	cx.notify();
@@ -284,13 +287,14 @@ impl DeviceView {
284	287	self.pending = true;
285	288	self.error = None;
286	289	let id = self.device.id.clone();
	290	+ let backend = self.backend.clone();
287	291	cx.spawn(async move \|this, cx\| {
288		- let result = cx.background_executor().spawn(async move { device::service().action(&id, &action) }).await;
	292	+ let result = cx.background_executor().spawn(async move { backend.action(&id, &action) }).await;
289	293	let _ = this.update(cx, \|this, cx\| {
290	294	this.pending = false;
291	295	match result {
292	296	Ok(detail) => this.detail = Some(detail),
293		- Err(err) => this.error = Some(format!("{err:#}").into()),
	297	+ Err(err) => this.error = Some(err.into()),
294	298	}
295	299	cx.notify();
296	300	});
@@ -311,10 +315,11 @@ impl DeviceView {
311	315	return;
312	316	}
313	317	let id = self.device.id.clone();
	318	+ let backend = self.backend.clone();
314	319	self._ax = cx.spawn(async move \|this, cx\| {
315	320	loop {
316		- let id = id.clone();
317		- let tree = cx.background_executor().spawn(async move { device::service().accessibility(&id) }).await;
	321	+ let (id, backend) = (id.clone(), backend.clone());
	322	+ let tree = cx.background_executor().spawn(async move { backend.accessibility(&id) }).await;
318	323	let keep_going = this
319	324	.update(cx, \|this, cx\| {
320	325	// A failed poll keeps the last good tree.
@@ -339,6 +344,7 @@ impl DeviceView {
339	344	}
340	345	self.screenshot_pending = true;
341	346	let id = self.device.id.clone();
	347	+ let backend = self.backend.clone();
342	348	let name: String = self.device.name.chars().map(\|c\| if c.is_ascii_alphanumeric() \|\| c == '-' { c } else { '-' }).collect();
343	349	let file = format!("{name}-{}.png", chrono::Local::now().timestamp_millis());
344	350	let dir = std::env::var_os("HOME").map(std::path::PathBuf::from).map(\|home\| {
@@ -352,9 +358,8 @@ impl DeviceView {
352	358	Some(target) => {
353	359	cx.background_executor()
354	360	.spawn(async move {
355		- let (_, png) = device::service().screenshot(&id)?;
356		- std::fs::write(&target, png)?;
357		- anyhow::Ok(())
	361	+ let png = backend.screenshot(&id)?;
	362	+ std::fs::write(&target, png).map_err(\|err\| err.to_string())
358	363	})
359	364	.await
360	365	}
@@ -363,7 +368,7 @@ impl DeviceView {
363	368	let _ = this.update(cx, \|this, cx\| {
364	369	this.screenshot_pending = false;
365	370	if let Err(err) = result {
366		- this.error = Some(format!("{err:#}").into());
	371	+ this.error = Some(err.into());
367	372	}
368	373	cx.notify();
369	374	});
@@ -783,7 +788,8 @@ impl DeviceView {
783	788	.child("Reading device settings…"),
784	789	);
785	790	}
786		- if let Some(diagnostics) = cx.global::<DeviceWatch>().state.host_detail.clone().filter(\|_\| cx.global::<DeviceWatch>().state.host_status == device::HostStatus::Ready) {
	791	+ let host = self.backend.state(cx);
	792	+ if let Some(diagnostics) = host.host_detail.clone().filter(\|_\| host.host_status == device::HostStatus::Ready) {
787	793	body = body.child(section("Host diagnostics", p).child(small_text(diagnostics, p)));
788	794	}
789	795
@@ -1000,7 +1006,7 @@ impl DeviceView {
1000	1006	move \|this: &mut Self, _: &mut Window, cx: &mut Context<Self>\| {
1001	1007	if let Some(app_id) = app_id.clone() {
1002	1008	let permission = this.permission;
1003		- this.act(Action::SetPermission { app_id, permission, decision }, cx);
	1009	+ this.act(Action::SetPermission { app_id, permission: permission.into(), decision: decision.into() }, cx);
1004	1010	}
1005	1011	}
1006	1012	};

+155−15src/ui/workspace/devices.rsView File

@@ -1,8 +1,13 @@
1	1	//! The workspace's side of devices: each thread's Device panel, the device
2	2	//! that floats over the chat, devices agents open and close, and the
3		- //! `device_*` tools agents call.
	3	+ //! `device_*` tools agents call. A thread on another device uses that
	4	+ //! device's devices, over the remote connection; this device serves its own
	5	+ //! to the devices connected to it.
4	6
5		- use std::collections::HashMap;
	7	+ use std::{
	8	+ collections::{HashMap, HashSet},
	9	+ sync::Arc,
	10	+ };
6	11
7	12	use base64::Engine as _;
8	13	use gpui_kit::component::{Icon, h_flex, v_flex};
@@ -12,9 +17,11 @@ use serde_json::{Value, json};
12	17	use super::Workspace;
13	18	use super::delegation::Answer;
14	19	use super::surfaces::Surface;
15		- use crate::device::{self, DeviceSummary, Event, HostStatus, LOCAL_HOST, Platform};
	20	+ use crate::device::{self, DeviceSummary, Event, HostStatus, LOCAL_HOST, Platform, State};
	21	+ use crate::remote::{DeviceCall, ServerMessage};
	22	+ use crate::session::Session;
16	23	use crate::settings;
17		- use crate::ui::device::{DevicePanel, DevicePanelEvent, DeviceSetup, DeviceWatch};
	24	+ use crate::ui::device::{Backend, DevicePanel, DevicePanelEvent, DeviceSetup, DeviceWatch, RemoteDevices, answer_device_call};
18	25	use crate::ui::icons::Lucide;
19	26	use crate::ui::metal::{self, MetalButton, MetalStyled as _};
20	27
@@ -52,6 +59,10 @@ struct FloatDrag {
52	59	pub(super) struct Devices {
53	60	/// Each thread's Device panel, by thread id.
54	61	panels: HashMap<String, Entity<DevicePanel>>,
	62	+ /// Connected devices' devices, by the connection's id.
	63	+ links: HashMap<String, Arc<RemoteDevices>>,
	64	+ /// Clients of this device that follow its devices, by connection.
	65	+ watchers: HashSet<u64>,
55	66	/// The floating window, in window coordinates, once placed.
56	67	float_rect: Option<Bounds<Pixels>>,
57	68	float_drag: Option<FloatDrag>,
@@ -69,13 +80,22 @@ impl Workspace {
69	80	self.devices._events = Some(cx.spawn_in(window, async move \|this, cx\| {
70	81	while let Ok(event) = events.recv().await {
71	82	let handled = this.update_in(cx, \|this, window, cx\| match event {
72		- Event::AgentOpened { thread, device } => this.agent_opened_device(&thread, device, window, cx),
	83	+ Event::AgentOpened { thread, device } => {
	84	+ // Connected devices show it too, for their view of the thread.
	85	+ if let Some(server) = this.remote_server() {
	86	+ server.broadcast(&ServerMessage::DeviceOpened { thread: thread.clone(), device: device.clone() });
	87	+ }
	88	+ if this.local_thread(&thread, cx).is_some() {
	89	+ this.agent_opened_device(&thread, device, window, cx);
	90	+ }
	91	+ }
73	92	Event::AgentClosed { thread, device } => {
74		- if let Some(panel) = this.devices.panels.get(&thread).cloned() {
75		- panel.update(cx, \|panel, cx\| panel.forget_device(device.as_deref(), cx));
	93	+ if let Some(server) = this.remote_server() {
	94	+ server.broadcast(&ServerMessage::DeviceClosed { thread: thread.clone(), device: device.clone() });
76	95	}
	96	+ this.agent_closed_device(&thread, device.as_deref(), cx);
77	97	}
78		- Event::Changed => {}
	98	+ Event::Changed => this.send_device_state(),
79	99	});
80	100	if handled.is_err() {
81	101	break;
@@ -86,23 +106,133 @@ impl Workspace {
86	106
87	107	/// A device an agent opened floats over the chat, or opens in the right
88	108	/// panel when Auto-show floating preview is off.
89		- fn agent_opened_device(&mut self, thread: &str, device: DeviceSummary, window: &mut Window, cx: &mut Context<Self>) {
	109	+ pub(super) fn agent_opened_device(&mut self, thread: &str, device: DeviceSummary, window: &mut Window, cx: &mut Context<Self>) {
90	110	let float = settings::get(\|settings\| settings.auto_float_preview);
91	111	let panel = self.device_panel(thread, window, cx);
92	112	panel.update(cx, \|panel, cx\| panel.show_device(device, float, window, cx));
93	113	let active = self.active.as_ref().is_some_and(\|session\| session.read(cx).id == thread);
94	114	if active && !float {
95	115	self.open_surface(Surface::Device, window, cx);
	116	+ } else if active && self.surface_open(Surface::Device) {
	117	+ // Floating closes the panel, as the browser's floating preview does.
	118	+ self.right_panel = None;
	119	+ }
	120	+ cx.notify();
	121	+ }
	122	+
	123	+ pub(super) fn agent_closed_device(&mut self, thread: &str, device: Option<&str>, cx: &mut Context<Self>) {
	124	+ if let Some(panel) = self.devices.panels.get(thread).cloned() {
	125	+ panel.update(cx, \|panel, cx\| panel.forget_device(device, cx));
	126	+ }
	127	+ }
	128	+
	129	+ /// Any thread, here or on a connected device, by its id here.
	130	+ fn session_by_id(&self, id: &str, cx: &App) -> Option<Entity<Session>> {
	131	+ self.projects.iter().flat_map(\|project\| project.sessions.iter()).find(\|session\| session.read(cx).id == id).cloned()
	132	+ }
	133	+
	134	+ /// Where a thread's devices are, and its id there: a thread on another
	135	+ /// device uses that device's.
	136	+ fn device_target(&mut self, thread: &str, cx: &mut Context<Self>) -> (Backend, String) {
	137	+ let remote = self.session_by_id(thread, cx).and_then(\|session\| {
	138	+ let remote = session.read(cx).remote.as_ref()?;
	139	+ Some((remote.env.clone(), remote.thread.clone()))
	140	+ });
	141	+ match remote {
	142	+ Some((env, host_thread)) => match self.devices_link(&env) {
	143	+ // A thread not created there yet gets an id of its own.
	144	+ Some(link) => (Backend::Remote(link), host_thread.unwrap_or_else(\|\| format!("pending-{thread}"))),
	145	+ None => (Backend::Local, thread.to_string()),
	146	+ },
	147	+ None => (Backend::Local, thread.to_string()),
	148	+ }
	149	+ }
	150	+
	151	+ /// A connected device's devices, made on first use.
	152	+ pub(super) fn devices_link(&mut self, env: &str) -> Option<Arc<RemoteDevices>> {
	153	+ if let Some(link) = self.devices.links.get(env) {
	154	+ return Some(link.clone());
	155	+ }
	156	+ let environment = self.remote.environments.iter().find(\|candidate\| candidate.host.id == env)?;
	157	+ let link = Arc::new(RemoteDevices::new(
	158	+ env.to_string(),
	159	+ environment.host.label(),
	160	+ environment.client.sender(),
	161	+ &environment.host.address,
	162	+ environment.host.key.clone(),
	163	+ ));
	164	+ link.watch();
	165	+ self.devices.links.insert(env.to_string(), link.clone());
	166	+ Some(link)
	167	+ }
	168	+
	169	+ /// A connected device went away: its devices with it.
	170	+ pub(super) fn forget_devices_link(&mut self, env: &str, cx: &mut Context<Self>) {
	171	+ self.devices.links.remove(env);
	172	+ if cx.global_mut::<DeviceWatch>().remote.remove(env).is_some() {
	173	+ cx.notify();
96	174	}
	175	+ }
	176	+
	177	+ /// A connected device's devices changed.
	178	+ pub(super) fn remote_device_state(&mut self, env: &str, mut state: State, cx: &mut Context<Self>) {
	179	+ // Its "This machine" is that device here.
	180	+ let name = self.environment_name(env);
	181	+ for host in &mut state.hosts {
	182	+ if host.id == LOCAL_HOST {
	183	+ host.label = name.clone();
	184	+ }
	185	+ }
	186	+ cx.global_mut::<DeviceWatch>().remote.insert(env.to_string(), state);
97	187	cx.notify();
98	188	}
99	189
	190	+ // ---- serving this device's devices ----
	191	+
	192	+ /// Answers a connected device's call for this device's devices, off the UI thread.
	193	+ pub(super) fn answer_device_call(&mut self, conn: u64, request: u64, call: DeviceCall, cx: &mut Context<Self>) {
	194	+ let Some(server) = self.remote_server() else { return };
	195	+ let sender = server.sender(conn);
	196	+ if call == DeviceCall::Watch {
	197	+ self.devices.watchers.insert(conn);
	198	+ sender.send(&ServerMessage::DeviceState { state: device::service().state() });
	199	+ return;
	200	+ }
	201	+ self.devices.watchers.insert(conn);
	202	+ cx.background_executor()
	203	+ .spawn(async move {
	204	+ let (ok, error) = match answer_device_call(call) {
	205	+ Ok(value) => (Some(value), None),
	206	+ Err(error) => (None, Some(error)),
	207	+ };
	208	+ sender.send(&ServerMessage::DeviceReply { request, ok, error });
	209	+ })
	210	+ .detach();
	211	+ }
	212	+
	213	+ /// Sends this device's devices to the clients following them.
	214	+ fn send_device_state(&self) {
	215	+ let Some(server) = self.remote_server() else { return };
	216	+ if self.devices.watchers.is_empty() {
	217	+ return;
	218	+ }
	219	+ let message = ServerMessage::DeviceState { state: device::service().state() };
	220	+ for conn in &self.devices.watchers {
	221	+ server.send(*conn, &message);
	222	+ }
	223	+ }
	224	+
	225	+ pub(super) fn forget_device_watcher(&mut self, conn: u64) {
	226	+ self.devices.watchers.remove(&conn);
	227	+ }
	228	+
100	229	/// A thread's Device panel, made on first use.
101	230	pub(super) fn device_panel(&mut self, thread: &str, window: &mut Window, cx: &mut Context<Self>) -> Entity<DevicePanel> {
102	231	if let Some(panel) = self.devices.panels.get(thread) {
103	232	return panel.clone();
104	233	}
105		- let panel = cx.new(\|cx\| DevicePanel::new(thread.to_string(), window, cx));
	234	+ let (backend, target) = self.device_target(thread, cx);
	235	+ let panel = cx.new(\|cx\| DevicePanel::new(target, backend, window, cx));
106	236	let thread_id = thread.to_string();
107	237	cx.subscribe_in(&panel, window, move \|this, _, event: &DevicePanelEvent, window, cx\| {
108	238	let active = this.active.as_ref().is_some_and(\|session\| session.read(cx).id == thread_id);
@@ -140,15 +270,22 @@ impl Workspace {
140	270	(Some(thread), true) => Some(self.device_panel(thread, window, cx)),
141	271	_ => None,
142	272	};
	273	+ // A new thread on another device gets its id there once it's made.
	274	+ if let (Some(thread), Some(panel)) = (&thread, &shown) {
	275	+ let (_, target) = self.device_target(thread, cx);
	276	+ panel.update(cx, \|panel, _\| panel.set_thread_id(target));
	277	+ }
143	278	if !surface {
144	279	self.devices.setup_offered = false;
145		- } else if !self.devices.setup_offered {
	280	+ } else if let Some(panel) = shown.as_ref().filter(\|_\| !self.devices.setup_offered) {
146	281	self.devices.setup_offered = true;
147		- let state = DeviceWatch::state(cx);
	282	+ let backend = panel.read(cx).backend().clone();
	283	+ let state = backend.state(cx);
148	284	if state.host_status == HostStatus::Disabled \|\| !state.onboarding_completed {
149	285	let workspace = cx.weak_entity();
150	286	cx.defer_in(window, move \|_, window, cx\| {
151	287	DeviceSetup::open_with(
	288	+ backend,
152	289	window,
153	290	cx,
154	291	Some(move \|_: &mut Window, cx: &mut App\| {
@@ -176,9 +313,12 @@ impl Workspace {
176	313
177	314	/// Drops a deleted thread's panel and closes its devices.
178	315	pub(super) fn forget_thread_devices(&mut self, thread: &str, cx: &mut Context<Self>) {
179		- if self.devices.panels.remove(thread).is_some() {
180		- let thread = thread.to_string();
181		- cx.background_executor().spawn(async move { device::service().close(&thread, None, false) }).detach();
	316	+ if let Some(panel) = self.devices.panels.remove(thread) {
	317	+ let (backend, thread) = {
	318	+ let panel = panel.read(cx);
	319	+ (panel.backend().clone(), panel.thread_id().to_string())
	320	+ };
	321	+ cx.background_executor().spawn(async move { backend.close(&thread, None, false) }).detach();
182	322	}
183	323	}
184	324

+31−2src/ui/workspace/remote.rsView File

@@ -179,6 +179,7 @@ impl Workspace {
179	179	/// Disconnects from a device and takes its projects and threads away.
180	180	fn drop_environment(&mut self, id: &str, window: &mut Window, cx: &mut Context<Self>) {
181	181	self.remote.remote_terminals.retain(\|(env, _), _\| env != id);
	182	+ self.forget_devices_link(id, cx);
182	183	self.remote.environments.retain(\|env\| env.host.id != id);
183	184	let removed: Vec<Entity<Session>> = self
184	185	.projects
@@ -205,6 +206,11 @@ impl Workspace {
205	206	}
206	207	}
207	208
	209	+ /// This device's remote server, while it serves.
	210	+ pub(super) fn remote_server(&self) -> Option<&RemoteServer> {
	211	+ self.remote.server.as_ref()
	212	+ }
	213	+
208	214	pub fn remote_overview(&self) -> RemoteOverview {
209	215	RemoteOverview {
210	216	serving: self.remote.server.as_ref().map(\|server\| server.port),
@@ -429,6 +435,7 @@ impl Workspace {
429	435	cx.notify();
430	436	}
431	437	ServerEvent::Disconnected { conn } => {
	438	+ self.forget_device_watcher(conn);
432	439	self.remote.peers.retain(\|peer\| peer.conn != conn);
433	440	self.remote.subscriptions.remove(&conn);
434	441	self.remote.shells.retain(\|(owner, _), _\| *owner != conn);
@@ -441,6 +448,7 @@ impl Workspace {
441	448	fn on_client_message(&mut self, conn: u64, message: ClientMessage, window: &mut Window, cx: &mut Context<Self>) {
442	449	match message {
443	450	ClientMessage::Hello { .. } => {}
	451	+ ClientMessage::Device { request, call } => self.answer_device_call(conn, request, call, cx),
444	452	ClientMessage::Subscribe { thread } => {
445	453	self.remote.subscriptions.entry(conn).or_default().insert(thread.clone(), None);
446	454	self.remote.dirty_threads.insert(thread);
@@ -718,6 +726,9 @@ impl Workspace {
718	726	environment.status = status;
719	727	}
720	728	if connected {
	729	+ if let Some(link) = self.devices_link(env) {
	730	+ link.watch();
	731	+ }
721	732	// A new connection knows nothing of what this one watched.
722	733	for session in self.environment_sessions(env) {
723	734	session.update(cx, \|session, _\| {
@@ -772,6 +783,24 @@ impl Workspace {
772	783	self.new_thread(ix, window, cx);
773	784	}
774	785	ClientEvent::Message(ServerMessage::Welcome { .. } \| ServerMessage::Denied { .. }) => {}
	786	+ ClientEvent::Message(ServerMessage::DeviceReply { request, ok, error }) => {
	787	+ if let Some(link) = self.devices_link(env) {
	788	+ link.resolve(request, error.map_or_else(\|\| Ok(ok.unwrap_or_default()), Err));
	789	+ }
	790	+ }
	791	+ ClientEvent::Message(ServerMessage::DeviceState { state }) => self.remote_device_state(env, state, cx),
	792	+ ClientEvent::Message(ServerMessage::DeviceOpened { thread, device }) => {
	793	+ if let Some(session) = self.remote_session(env, &thread, cx) {
	794	+ let local = session.read(cx).id.clone();
	795	+ self.agent_opened_device(&local, device, window, cx);
	796	+ }
	797	+ }
	798	+ ClientEvent::Message(ServerMessage::DeviceClosed { thread, device }) => {
	799	+ if let Some(session) = self.remote_session(env, &thread, cx) {
	800	+ let local = session.read(cx).id.clone();
	801	+ self.agent_closed_device(&local, device.as_deref(), cx);
	802	+ }
	803	+ }
775	804	ClientEvent::Message(ServerMessage::TerminalOutput { terminal, data }) => {
776	805	let view = self.remote.remote_terminals.get(&(env.to_string(), terminal)).and_then(WeakEntity::upgrade);
777	806	match (view, BASE64.decode(data)) {
@@ -856,7 +885,7 @@ impl Workspace {
856	885	}
857	886	}
858	887
859		- fn environment_sessions(&self, env: &str) -> Vec<Entity<Session>> {
	888	+ pub(super) fn environment_sessions(&self, env: &str) -> Vec<Entity<Session>> {
860	889	self.projects
861	890	.iter()
862	891	.filter(\|project\| project.environment.as_deref() == Some(env))
@@ -864,7 +893,7 @@ impl Workspace {
864	893	.collect()
865	894	}
866	895
867		- fn remote_session(&self, env: &str, id: &str, cx: &App) -> Option<Entity<Session>> {
	896	+ pub(super) fn remote_session(&self, env: &str, id: &str, cx: &App) -> Option<Entity<Session>> {
868	897	self.environment_sessions(env).into_iter().find(\|session\| {
869	898	session.read(cx).remote.as_ref().and_then(\|remote\| remote.thread.as_deref()) == Some(id)
870	899	})