tenant.openagents/omega

1use std::io::Write;
2
3use crate::{
4    RemoteArch, RemoteOs, RemotePlatform,
5    json_log::LogRecord,
6    protocol::{MESSAGE_LEN_SIZE, message_len_from_buffer, read_message_with_len, write_message},
7};
8use anyhow::{Context as _, Result};
9use futures::{
10    AsyncReadExt as _, FutureExt as _, StreamExt as _,
11    channel::mpsc::{Sender, UnboundedReceiver, UnboundedSender},
12};
13use gpui::{AppContext as _, AsyncApp, Task};
14use rpc::proto::Envelope;
15use util::command::Child;
16
17pub mod docker;
18#[cfg(any(test, feature = "test-support"))]
19pub mod mock;
20pub mod ssh;
21pub mod wsl;
22
23/// Parses the output of `uname -sm` to determine the remote platform.
24/// Takes the last line to skip possible shell initialization output.
25fn parse_platform(output: &str) -> Result<RemotePlatform> {
26    let output = output.trim();
27    let uname = output.rsplit_once('\n').map_or(output, |(_, last)| last);
28    let Some((os, arch)) = uname.split_once(" ") else {
29        anyhow::bail!("unknown uname: {uname:?}")
30    };
31
32    let os = match os {
33        "Darwin" => RemoteOs::MacOs,
34        "Linux" => RemoteOs::Linux,
35        _ => anyhow::bail!(
36            "Prebuilt remote servers are not yet available for {os:?}. See https://zed.dev/docs/remote-development"
37        ),
38    };
39
40    // exclude armv5,6,7 as they are 32-bit.
41    let arch = if arch.starts_with("armv8")
42        || arch.starts_with("armv9")
43        || arch.starts_with("arm64")
44        || arch.starts_with("aarch64")
45    {
46        RemoteArch::Aarch64
47    } else if arch.starts_with("x86") {
48        RemoteArch::X86_64
49    } else {
50        anyhow::bail!(
51            "Prebuilt remote servers are not yet available for {arch:?}. See https://zed.dev/docs/remote-development"
52        )
53    };
54
55    Ok(RemotePlatform { os, arch })
56}
57
58/// The command (program + args) used to read a remote host's OS version, given
59/// its detected OS.
60///
61/// The output is parsed by [`parse_os_version`].
62pub(crate) fn os_version_command(os: RemoteOs) -> (&'static str, &'static [&'static str]) {
63    match os {
64        // Matches the `/etc/os-release` parsing in `client::telemetry::os_version`.
65        RemoteOs::Linux => ("cat", &["/etc/os-release"]),
66        RemoteOs::MacOs => ("sw_vers", &["-productVersion"]),
67        // Prints e.g. "Microsoft Windows [Version 10.0.19045.5011]".
68        RemoteOs::Windows => ("cmd.exe", &["/c", "ver"]),
69    }
70}
71
72/// Parses the output of [`os_version_command`] into a human-readable version
73/// string, matching the conventions used by `client::telemetry::os_version`.
74///
75/// For Linux this is `"{ID} {VERSION_ID}"` (e.g. `"ubuntu 24.04"`); for macOS it
76/// is the product version (e.g. `"15.6.1"`); for Windows it is the
77/// `major.minor.build` version (e.g. `"10.0.19045"`). Returns `None` if nothing
78/// usable could be parsed.
79pub(crate) fn parse_os_version(os: RemoteOs, output: &str) -> Option<String> {
80    let output = output.trim();
81    if output.is_empty() {
82        return None;
83    }
84    match os {
85        RemoteOs::Linux => util::parse_os_release(output),
86        RemoteOs::MacOs => {
87            // `sw_vers -productVersion` prints a single version line.
88            output
89                .lines()
90                .next_back()
91                .map(|line| line.trim().to_string())
92                .filter(|line| !line.is_empty())
93        }
94        RemoteOs::Windows => parse_windows_version(output),
95    }
96}
97
98/// Extracts a `major.minor.build` version from the output of `cmd.exe /c ver`,
99/// e.g. `"Microsoft Windows [Version 10.0.19045.5011]"` -> `"10.0.19045"`.
100///
101/// Scans for the first dotted run of integers (rather than relying on the
102/// surrounding, potentially localized, text) and drops the trailing revision so
103/// the format matches `client::telemetry::os_version` on Windows.
104fn parse_windows_version(output: &str) -> Option<String> {
105    output
106        .split(|c: char| !c.is_ascii_digit() && c != '.')
107        .filter_map(|token| {
108            let parts: Vec<&str> = token.split('.').filter(|part| !part.is_empty()).collect();
109            (parts.len() >= 3 && parts.iter().all(|part| part.parse::<u32>().is_ok()))
110                .then(|| parts[..3].join("."))
111        })
112        .next()
113}
114
115/// Parses the output of `echo $SHELL` to determine the remote shell.
116/// Takes the last line to skip possible shell initialization output.
117fn parse_shell(output: &str, fallback_shell: &str) -> String {
118    let output = output.trim();
119    let shell = output.rsplit_once('\n').map_or(output, |(_, last)| last);
120    if shell.is_empty() {
121        log::error!("$SHELL is not set, falling back to {fallback_shell}");
122        fallback_shell.to_owned()
123    } else {
124        shell.to_owned()
125    }
126}
127
128fn handle_rpc_messages_over_child_process_stdio(
129    mut remote_proxy_process: Child,
130    incoming_tx: UnboundedSender<Envelope>,
131    mut outgoing_rx: UnboundedReceiver<Envelope>,
132    mut connection_activity_tx: Sender<()>,
133    cx: &AsyncApp,
134) -> Task<Result<i32>> {
135    let mut child_stderr = remote_proxy_process.stderr.take().unwrap();
136    let mut child_stdout = remote_proxy_process.stdout.take().unwrap();
137    let mut child_stdin = remote_proxy_process.stdin.take().unwrap();
138
139    let mut stdin_buffer = Vec::new();
140    let mut stdout_buffer = Vec::new();
141    let mut stderr_buffer = Vec::new();
142    let mut stderr_offset = 0;
143
144    let stdin_task = cx.background_spawn(async move {
145        while let Some(outgoing) = outgoing_rx.next().await {
146            write_message(&mut child_stdin, &mut stdin_buffer, outgoing).await?;
147        }
148        anyhow::Ok(())
149    });
150
151    let stdout_task = cx.background_spawn({
152        let mut connection_activity_tx = connection_activity_tx.clone();
153        async move {
154            loop {
155                stdout_buffer.resize(MESSAGE_LEN_SIZE, 0);
156                let len = child_stdout.read(&mut stdout_buffer).await?;
157
158                if len == 0 {
159                    return anyhow::Ok(());
160                }
161
162                if len < MESSAGE_LEN_SIZE {
163                    child_stdout.read_exact(&mut stdout_buffer[len..]).await?;
164                }
165
166                let message_len = message_len_from_buffer(&stdout_buffer);
167                let envelope =
168                    read_message_with_len(&mut child_stdout, &mut stdout_buffer, message_len)
169                        .await?;
170                connection_activity_tx.try_send(()).ok();
171                incoming_tx.unbounded_send(envelope).ok();
172            }
173        }
174    });
175
176    let stderr_task: Task<anyhow::Result<()>> = cx.background_spawn(async move {
177        loop {
178            stderr_buffer.resize(stderr_offset + 1024, 0);
179
180            let len = child_stderr
181                .read(&mut stderr_buffer[stderr_offset..])
182                .await?;
183            if len == 0 {
184                return anyhow::Ok(());
185            }
186
187            stderr_offset += len;
188            let mut start_ix = 0;
189            while let Some(ix) = stderr_buffer[start_ix..stderr_offset]
190                .iter()
191                .position(|b| b == &b'\n')
192            {
193                let line_ix = start_ix + ix;
194                let content = &stderr_buffer[start_ix..line_ix];
195                start_ix = line_ix + 1;
196                if let Ok(record) = serde_json::from_slice::<LogRecord>(content) {
197                    record.log(log::logger())
198                } else {
199                    std::io::stderr()
200                        .write_fmt(format_args!(
201                            "(remote) {}\n",
202                            String::from_utf8_lossy(content)
203                        ))
204                        .ok();
205                }
206            }
207            stderr_buffer.drain(0..start_ix);
208            stderr_offset -= start_ix;
209
210            connection_activity_tx.try_send(()).ok();
211        }
212    });
213
214    cx.background_spawn(async move {
215        let result = futures::select! {
216            result = stdin_task.fuse() => {
217                result.context("stdin")
218            }
219            result = stdout_task.fuse() => {
220                result.context("stdout")
221            }
222            result = stderr_task.fuse() => {
223                result.context("stderr")
224            }
225        };
226        let exit_status = remote_proxy_process.status().await?;
227        let status = exit_status.code().unwrap_or_else(|| {
228            #[cfg(unix)]
229            let status = std::os::unix::process::ExitStatusExt::signal(&exit_status).unwrap_or(1);
230            #[cfg(not(unix))]
231            let status = 1;
232            status
233        });
234        match result {
235            Ok(_) => Ok(status),
236            Err(error) => Err(error),
237        }
238    })
239}
240
241#[cfg(any(debug_assertions, feature = "build-remote-server-binary"))]
242async fn build_remote_server_from_source(
243    platform: &crate::RemotePlatform,
244    delegate: &dyn crate::RemoteClientDelegate,
245    binary_exists_on_server: bool,
246    cx: &mut AsyncApp,
247) -> Result<Option<std::path::PathBuf>> {
248    use std::env::VarError;
249    use std::path::Path;
250    use util::command::{Command, Stdio, new_command};
251
252    if let Ok(path) = std::env::var("ZED_COPY_REMOTE_SERVER") {
253        let path = std::path::PathBuf::from(path);
254        if path.exists() {
255            return Ok(Some(path));
256        } else {
257            log::warn!(
258                "ZED_COPY_REMOTE_SERVER path does not exist, falling back to ZED_BUILD_REMOTE_SERVER: {}",
259                path.display()
260            );
261        }
262    }
263
264    // By default, we make building remote server from source opt-out and we do not force artifact compression
265    // for quicker builds.
266    let build_remote_server =
267        std::env::var("ZED_BUILD_REMOTE_SERVER").unwrap_or("nocompress".into());
268
269    if let "never" = &*build_remote_server {
270        return Ok(None);
271    } else if let "false" | "no" | "off" | "0" = &*build_remote_server {
272        if binary_exists_on_server {
273            return Ok(None);
274        }
275        log::warn!("ZED_BUILD_REMOTE_SERVER is disabled, but no server binary exists on the server")
276    }
277
278    async fn run_cmd(command: &mut Command) -> Result<()> {
279        let output = command
280            .kill_on_drop(true)
281            .stdout(Stdio::inherit())
282            .stderr(Stdio::inherit())
283            .output()
284            .await?;
285        anyhow::ensure!(
286            output.status.success(),
287            "Failed to run command: {command:?}: output: {}",
288            String::from_utf8_lossy(&output.stderr)
289        );
290        Ok(())
291    }
292
293    let use_musl = !build_remote_server.contains("nomusl");
294    let triple = format!(
295        "{}-{}",
296        platform.arch,
297        match platform.os {
298            RemoteOs::Linux =>
299                if use_musl {
300                    "unknown-linux-musl"
301                } else {
302                    "unknown-linux-gnu"
303                },
304            RemoteOs::MacOs => "apple-darwin",
305            RemoteOs::Windows if cfg!(windows) => "pc-windows-msvc",
306            RemoteOs::Windows => "pc-windows-gnu",
307        }
308    );
309    let mut rust_flags = match std::env::var("RUSTFLAGS") {
310        Ok(val) => val,
311        Err(VarError::NotPresent) => String::new(),
312        Err(e) => {
313            log::error!("Failed to get env var `RUSTFLAGS` value: {e}");
314            String::new()
315        }
316    };
317    if platform.os == RemoteOs::Linux && use_musl {
318        rust_flags.push_str(" -C target-feature=+crt-static");
319
320        if let Ok(path) = std::env::var("ZED_ZSTD_MUSL_LIB") {
321            rust_flags.push_str(&format!(" -C link-arg=-L{path}"));
322        }
323    }
324    if platform.arch.as_str() == std::env::consts::ARCH
325        && platform.os.as_str() == std::env::consts::OS
326    {
327        delegate.set_status(Some("Building remote server binary from source"), cx);
328        log::info!("building remote server binary from source");
329        run_cmd(
330            new_command("cargo")
331                .current_dir(concat!(env!("CARGO_MANIFEST_DIR"), "/../.."))
332                .args([
333                    "build",
334                    "--package",
335                    "remote_server",
336                    "--features",
337                    "debug-embed",
338                    "--target-dir",
339                    "target/remote_server",
340                    "--target",
341                    &triple,
342                ])
343                .env("RUSTFLAGS", &rust_flags),
344        )
345        .await?;
346    } else {
347        if which("zig", cx).await?.is_none() {
348            anyhow::bail!(if cfg!(not(windows)) {
349                "zig not found on $PATH, install zig (see https://ziglang.org/learn/getting-started or use zigup)"
350            } else {
351                "zig not found on $PATH, install zig (use `winget install -e --id zig.zig` or see https://ziglang.org/learn/getting-started or use zigup)"
352            });
353        }
354
355        let rustup = which("rustup", cx)
356            .await?
357            .context("rustup not found on $PATH, install rustup (see https://rustup.rs/)")?;
358        delegate.set_status(Some("Adding rustup target for cross-compilation"), cx);
359        log::info!("adding rustup target");
360        run_cmd(new_command(rustup).args(["target", "add"]).arg(&triple)).await?;
361
362        if which("cargo-zigbuild", cx).await?.is_none() {
363            delegate.set_status(Some("Installing cargo-zigbuild for cross-compilation"), cx);
364            log::info!("installing cargo-zigbuild");
365            run_cmd(new_command("cargo").args(["install", "--locked", "cargo-zigbuild"])).await?;
366        }
367
368        delegate.set_status(
369            Some(&format!(
370                "Building remote binary from source for {triple} with Zig"
371            )),
372            cx,
373        );
374        log::info!("building remote binary from source for {triple} with Zig");
375        run_cmd(
376            new_command("cargo")
377                .current_dir(concat!(env!("CARGO_MANIFEST_DIR"), "/../.."))
378                .args([
379                    "zigbuild",
380                    "--package",
381                    "remote_server",
382                    "--features",
383                    "debug-embed",
384                    "--target-dir",
385                    "target/remote_server",
386                    "--target",
387                    &triple,
388                ])
389                .env("RUSTFLAGS", &rust_flags),
390        )
391        .await?;
392    };
393    let bin_path = Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/../.."))
394        .join("target")
395        .join("remote_server")
396        .join(&triple)
397        .join("debug")
398        .join("remote_server")
399        .with_extension(if platform.os.is_windows() { "exe" } else { "" });
400
401    let path = if !build_remote_server.contains("nocompress") {
402        delegate.set_status(Some("Compressing binary"), cx);
403
404        #[cfg(not(target_os = "windows"))]
405        let archive_path = {
406            run_cmd(new_command("gzip").arg("-f").arg(&bin_path)).await?;
407            bin_path.with_extension("gz")
408        };
409
410        #[cfg(target_os = "windows")]
411        let archive_path = {
412            let zip_path = bin_path.with_extension("zip");
413            if smol::fs::metadata(&zip_path).await.is_ok() {
414                smol::fs::remove_file(&zip_path).await?;
415            }
416            let compress_command = format!(
417                "Compress-Archive -Path '{}' -DestinationPath '{}' -Force",
418                bin_path.display(),
419                zip_path.display(),
420            );
421            run_cmd(new_command("powershell.exe").args([
422                "-NoProfile",
423                "-Command",
424                &compress_command,
425            ]))
426            .await?;
427            zip_path
428        };
429
430        std::env::current_dir()?.join(archive_path)
431    } else {
432        bin_path
433    };
434
435    Ok(Some(path))
436}
437
438#[cfg(any(debug_assertions, feature = "build-remote-server-binary"))]
439async fn which(
440    binary_name: impl AsRef<str>,
441    cx: &mut AsyncApp,
442) -> Result<Option<std::path::PathBuf>> {
443    let binary_name = binary_name.as_ref().to_string();
444    let binary_name_cloned = binary_name.clone();
445    let res = cx
446        .background_spawn(async move { which::which(binary_name_cloned) })
447        .await;
448    match res {
449        Ok(path) => Ok(Some(path)),
450        Err(which::Error::CannotFindBinaryPath) => Ok(None),
451        Err(err) => Err(anyhow::anyhow!(
452            "Failed to run 'which' to find the binary '{binary_name}': {err}"
453        )),
454    }
455}
456
457#[cfg(test)]
458mod tests {
459    use super::*;
460
461    #[test]
462    fn test_parse_platform() {
463        let result = parse_platform("Linux x86_64\n").unwrap();
464        assert_eq!(result.os, RemoteOs::Linux);
465        assert_eq!(result.arch, RemoteArch::X86_64);
466
467        let result = parse_platform("Darwin arm64\n").unwrap();
468        assert_eq!(result.os, RemoteOs::MacOs);
469        assert_eq!(result.arch, RemoteArch::Aarch64);
470
471        let result = parse_platform("Linux x86_64").unwrap();
472        assert_eq!(result.os, RemoteOs::Linux);
473        assert_eq!(result.arch, RemoteArch::X86_64);
474
475        let result = parse_platform("some shell init output\nLinux aarch64\n").unwrap();
476        assert_eq!(result.os, RemoteOs::Linux);
477        assert_eq!(result.arch, RemoteArch::Aarch64);
478
479        let result = parse_platform("some shell init output\nLinux aarch64").unwrap();
480        assert_eq!(result.os, RemoteOs::Linux);
481        assert_eq!(result.arch, RemoteArch::Aarch64);
482
483        assert_eq!(
484            parse_platform("Linux armv8l\n").unwrap().arch,
485            RemoteArch::Aarch64
486        );
487        assert_eq!(
488            parse_platform("Linux aarch64\n").unwrap().arch,
489            RemoteArch::Aarch64
490        );
491        assert_eq!(
492            parse_platform("Linux x86_64\n").unwrap().arch,
493            RemoteArch::X86_64
494        );
495
496        let result = parse_platform(
497            r#"Linux x86_64 - What you're referring to as Linux, is in fact, GNU/Linux...\n"#,
498        )
499        .unwrap();
500        assert_eq!(result.os, RemoteOs::Linux);
501        assert_eq!(result.arch, RemoteArch::X86_64);
502
503        assert!(parse_platform("Windows x86_64\n").is_err());
504        assert!(parse_platform("Linux armv7l\n").is_err());
505    }
506
507    #[test]
508    fn test_parse_os_version() {
509        // Linux delegates to `util::parse_os_release` (tested there); confirm
510        // the dispatch is wired up.
511        let os_release = "ID=ubuntu\nVERSION_ID=\"24.04\"\n";
512        assert_eq!(
513            parse_os_version(RemoteOs::Linux, os_release),
514            Some("ubuntu 24.04".to_string())
515        );
516
517        // macOS `sw_vers -productVersion` prints a bare version, possibly after
518        // shell initialization noise.
519        assert_eq!(
520            parse_os_version(RemoteOs::MacOs, "15.6.1\n"),
521            Some("15.6.1".to_string())
522        );
523        assert_eq!(
524            parse_os_version(RemoteOs::MacOs, "shell noise\n26.0\n"),
525            Some("26.0".to_string())
526        );
527        assert_eq!(parse_os_version(RemoteOs::MacOs, ""), None);
528
529        // Windows `cmd.exe /c ver`, with the trailing revision dropped to match
530        // the `major.minor.build` format used by local Windows telemetry.
531        assert_eq!(
532            parse_os_version(
533                RemoteOs::Windows,
534                "Microsoft Windows [Version 10.0.19045.5011]\n"
535            ),
536            Some("10.0.19045".to_string())
537        );
538        // Localized output: only the version number is relied upon.
539        assert_eq!(
540            parse_os_version(
541                RemoteOs::Windows,
542                "Microsoft Windows [Versione 10.0.22631.1]"
543            ),
544            Some("10.0.22631".to_string())
545        );
546        assert_eq!(parse_os_version(RemoteOs::Windows, "no version here"), None);
547    }
548
549    #[test]
550    fn test_parse_shell() {
551        assert_eq!(parse_shell("/bin/bash\n", "sh"), "/bin/bash");
552        assert_eq!(parse_shell("/bin/zsh\n", "sh"), "/bin/zsh");
553
554        assert_eq!(parse_shell("/bin/bash", "sh"), "/bin/bash");
555        assert_eq!(
556            parse_shell("some shell init output\n/bin/bash\n", "sh"),
557            "/bin/bash"
558        );
559        assert_eq!(
560            parse_shell("some shell init output\n/bin/bash", "sh"),
561            "/bin/bash"
562        );
563        assert_eq!(parse_shell("", "sh"), "sh");
564        assert_eq!(parse_shell("\n", "sh"), "sh");
565    }
566}
567