//! The device host that is this machine, as T3 Code's `LocalDeviceHost`:
//! expo-device-hub runs as a supervised child on a loopback port, and the
//! agent-device daemon runs in HTTP mode under the app's own state folder.
//! Both start lazily, once the user has agreed to them.
//!
//! The hub binds loopback only. It has unauthenticated routes that run
//! commands, so it's never exposed beyond this machine.
use std::{
io::{BufRead as _, BufReader, Read as _, Write as _},
os::unix::process::CommandExt as _,
path::{Path, PathBuf},
process::{Child, Command, Stdio},
sync::{Arc, Mutex},
thread,
time::{Duration, Instant},
};
use anyhow::{Context as _, Result, anyhow, bail};
use serde::{Deserialize, Serialize};
use super::toolchain::{self, Tool};
use super::{HostStatus, Platform, PlatformAvailability};
const HUB_READY_TIMEOUT: Duration = Duration::from_secs(30);
const DAEMON_READY_TIMEOUT: Duration = Duration::from_secs(30);
/// A hub that ran this long before dying restarts at once; one that dies
/// sooner waits, doubling up to the maximum, so a crash on boot can't spin.
const HUB_STABLE_UPTIME: Duration = Duration::from_secs(60);
const HUB_RESTART_MAX: Duration = Duration::from_secs(30);
/// Where the Android SDK is, and which of its parts are there.
#[derive(Clone, Debug, Default)]
pub struct AndroidSdk {
pub root: Option<PathBuf>,
adb: bool,
emulator: bool,
avdmanager: bool,
legacy_avdmanager: bool,
}
/// `ANDROID_HOME`, the standard locations, then the folder above an `adb`
/// on PATH.
pub fn android_sdk() -> AndroidSdk {
let var = |name: &str| std::env::var(name).ok().map(|value| value.trim().to_string()).filter(|value| !value.is_empty());
let explicit = var("ANDROID_HOME").or_else(|| var("ANDROID_SDK_ROOT"));
let home = std::env::var_os("HOME").map(PathBuf::from).unwrap_or_default();
let mut candidates: Vec<PathBuf> = match &explicit {
Some(root) => vec![PathBuf::from(root)],
None => vec![home.join("Library/Android/sdk"), home.join("Android/Sdk")],
};
if explicit.is_none() {
for dir in std::env::var_os("PATH").map(|paths| std::env::split_paths(&paths).collect::<Vec<_>>()).unwrap_or_default() {
if let Ok(adb) = std::fs::canonicalize(dir.join("adb"))
&& let Some(root) = adb.parent().and_then(Path::parent)
{
candidates.push(root.to_path_buf());
}
}
}
for root in candidates {
let adb = root.join("platform-tools/adb").is_file();
let emulator = root.join("emulator/emulator").is_file();
if explicit.is_some() || adb || emulator {
let avdmanager = root.join("cmdline-tools/latest/bin/avdmanager").is_file();
let legacy_avdmanager = !avdmanager && root.join("tools/bin/avdmanager").is_file();
return AndroidSdk { root: Some(root), adb, emulator, avdmanager, legacy_avdmanager };
}
}
AndroidSdk::default()
}
/// Why a platform can't run here, or `None` when it can.
pub fn platform_reason(platform: Platform, sdk: &AndroidSdk) -> Option<String> {
match platform {
Platform::Ios => {
if !cfg!(target_os = "macos") {
Some("iOS Simulators need macOS with Xcode.".into())
} else if crate::browser::find_program("xcrun").is_none() {
Some("Xcode command line tools were not found.".into())
} else {
None
}
}
Platform::Android => {
let Some(root) = &sdk.root else {
return Some(
"Android SDK was not found. Install it with Android Studio or set ANDROID_HOME to your SDK directory."
.into(),
);
};
let root = root.display();
if !sdk.adb {
Some(format!("Android SDK Platform-Tools are missing from {root}. Install them in Android Studio's SDK Manager."))
} else if !sdk.emulator {
Some(format!("Android Emulator is missing from {root}. Install it in Android Studio's SDK Manager."))
} else if !sdk.avdmanager && sdk.legacy_avdmanager {
Some(format!(
"The Android SDK command-line tools in {root} appear to be an older, unsupported version. Install Android SDK Command-line Tools (latest) in Android Studio's SDK Manager under SDK Tools."
))
} else if !sdk.avdmanager {
Some(format!("Android SDK Command-line Tools (latest) are missing from {root}. Install them in Android Studio's SDK Manager."))
} else {
None
}
}
}
}
pub fn availability(sdk: &AndroidSdk) -> Vec<PlatformAvailability> {
[Platform::Ios, Platform::Android]
.into_iter()
.map(|platform| {
let reason = platform_reason(platform, sdk);
PlatformAvailability { platform, available: reason.is_none(), reason }
})
.collect()
}
/// The environment the hub and the platform tools run with: the SDK's
/// tools on PATH and its root in `ANDROID_HOME`.
fn apply_env(command: &mut Command, sdk: &AndroidSdk) {
let mut dirs: Vec<PathBuf> = Vec::new();
if let Some(root) = &sdk.root {
command.env("ANDROID_HOME", root);
dirs.push(root.join("platform-tools"));
dirs.push(root.join("emulator"));
}
// npm-installed tools run with the Node the app found.
if let Some(node_dir) = toolchain::node().as_deref().and_then(Path::parent) {
dirs.push(node_dir.to_path_buf());
}
let current = std::env::var_os("PATH").unwrap_or_default();
let path = std::env::join_paths(dirs.into_iter().chain(std::env::split_paths(¤t))).unwrap_or(current);
command.env("PATH", path).env("FORCE_COLOR", "0").env("NO_COLOR", "1");
// SSH sessions may lack the folder the emulator publishes its gRPC token in.
if std::env::var_os("XDG_RUNTIME_DIR").is_none() {
let dir = PathBuf::from(format!("/run/user/{}", unsafe { libc::getuid() }));
if dir.is_dir() {
command.env("XDG_RUNTIME_DIR", dir);
}
}
}
/// What a finished command printed.
#[derive(Debug, Default)]
pub struct Output {
pub stdout: String,
pub stderr: String,
/// The exit code; 124 for a timeout, 127 when it couldn't start.
pub code: i32,
}
impl Output {
pub fn ok(self, operation: &str) -> Result<String> {
if self.code == 0 {
Ok(self.stdout)
} else {
let detail = self.stderr.trim();
let detail = if detail.is_empty() { self.stdout.trim() } else { detail };
let detail: String = detail.chars().take(400).collect();
Err(anyhow!(
"Device {operation} failed: The device command failed (exit code {}).{}",
self.code,
if detail.is_empty() { String::new() } else { format!("\n{detail}") }
))
}
}
}
/// Runs a host command (`adb`, `emulator`, `xcrun`, or Node) with the SDK's
/// environment, waiting at most `timeout`.
pub fn run(program: &str, args: &[&str], timeout: Duration, stdin: Option<&str>) -> Output {
let sdk = android_sdk();
let program: PathBuf = match (program, &sdk.root) {
("emulator", Some(root)) => root.join("emulator/emulator"),
("adb", Some(root)) if root.join("platform-tools/adb").is_file() => root.join("platform-tools/adb"),
_ => PathBuf::from(program),
};
let mut command = Command::new(&program);
command.args(args).stdout(Stdio::piped()).stderr(Stdio::piped());
command.stdin(if stdin.is_some() { Stdio::piped() } else { Stdio::null() });
apply_env(&mut command, &sdk);
let mut child = match command.spawn() {
Ok(child) => child,
Err(err) => return Output { stderr: err.to_string(), code: 127, ..Output::default() },
};
if let (Some(text), Some(mut pipe)) = (stdin, child.stdin.take()) {
let text = text.to_string();
thread::spawn(move || {
let _ = pipe.write_all(text.as_bytes());
});
}
let read = |pipe: Option<Box<dyn std::io::Read + Send>>| {
thread::spawn(move || {
let mut text = Vec::new();
if let Some(mut pipe) = pipe {
let _ = pipe.read_to_end(&mut text);
}
String::from_utf8_lossy(&text).into_owned()
})
};
let stdout = read(child.stdout.take().map(|pipe| Box::new(pipe) as Box<dyn std::io::Read + Send>));
let stderr = read(child.stderr.take().map(|pipe| Box::new(pipe) as Box<dyn std::io::Read + Send>));
let deadline = Instant::now() + timeout;
let code = loop {
match child.try_wait() {
Ok(Some(status)) => break status.code().unwrap_or(1),
Ok(None) if Instant::now() < deadline => thread::sleep(Duration::from_millis(20)),
_ => {
let _ = child.kill();
let _ = child.wait();
break 124;
}
}
};
Output { stdout: stdout.join().unwrap_or_default(), stderr: stderr.join().unwrap_or_default(), code }
}
/// Paths of helpers serve-sim bundles, when the hub has them.
#[derive(Clone, Debug, Default)]
pub struct Helpers {
pub ax_settings: Option<PathBuf>,
pub serve_sim_cli: Option<PathBuf>,
}
/// A running hub's endpoints.
#[derive(Clone, Debug)]
pub struct Ready {
/// The hub's loopback origin, `http://127.0.0.1:<port>`.
pub origin: String,
pub node: PathBuf,
pub helpers: Helpers,
}
/// The agent-device daemon's endpoint.
#[derive(Clone, Debug)]
pub struct AgentEndpoint {
pub base_url: String,
pub token: String,
}
struct HubProcess {
child: Arc<Mutex<Child>>,
pid: u32,
origin: String,
started: Instant,
}
impl HubProcess {
fn alive(&self) -> bool {
self.child.lock().map(|mut child| matches!(child.try_wait(), Ok(None))).unwrap_or(false)
}
fn kill(&self) {
if let Ok(mut child) = self.child.lock() {
unsafe {
libc::kill(self.pid as i32, libc::SIGTERM);
}
let deadline = Instant::now() + Duration::from_secs(3);
while matches!(child.try_wait(), Ok(None)) && Instant::now() < deadline {
thread::sleep(Duration::from_millis(50));
}
let _ = child.kill();
let _ = child.wait();
}
}
}
#[derive(Default)]
struct Running {
hub: Option<HubProcess>,
node: Option<PathBuf>,
agent: Option<AgentEndpoint>,
restart_delay: Duration,
}
/// Written next to the daemon's state, so a hub left behind by an app that
/// died without cleaning up is stopped on the next start.
#[derive(Serialize, Deserialize)]
struct HubStateFile {
pid: u32,
port: u16,
#[serde(rename = "entryPath")]
entry_path: String,
}
#[derive(Deserialize)]
struct DaemonFile {
#[serde(rename = "httpPort")]
http_port: u16,
token: String,
pid: Option<i32>,
version: Option<String>,
}
/// The state folder agent-device and the hub record live in.
pub fn state_dir() -> Option<PathBuf> {
Some(toolchain::base_dir()?.join("device").join("agent-device"))
}
/// The phase callback ensure calls report progress through.
pub type OnPhase<'a> = &'a dyn Fn(HostStatus, Option<String>);
pub struct LocalHost {
running: Arc<Mutex<Running>>,
/// Starts and stops one at a time.
lock: Mutex<()>,
}
impl LocalHost {
pub fn new() -> Self {
Self { running: Arc::default(), lock: Mutex::new(()) }
}
fn lock_running(&self) -> std::sync::MutexGuard<'_, Running> {
self.running.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
}
/// The hub's endpoints if it runs, without starting anything.
pub fn current(&self) -> Option<Ready> {
let running = self.lock_running();
let hub = running.hub.as_ref().filter(|hub| hub.alive())?;
Some(Ready { origin: hub.origin.clone(), node: running.node.clone()?, helpers: helpers() })
}
/// Whether the hub runs, and agent-device's version if its daemon does.
pub fn running_versions(&self) -> (bool, Option<String>) {
let hub = self.current().is_some();
let agent = self.lock_running().agent.is_some().then(read_daemon_file).flatten();
let agent = agent.filter(|file| file.pid.is_none_or(pid_alive)).and_then(|file| file.version);
(hub, agent)
}
/// Installs the hub on first use and starts it; a running hub returns at once.
pub fn ensure_ready(&self, on_phase: OnPhase) -> Result<Ready> {
let _guard = self.lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
self.ensure_hub(on_phase)
}
fn ensure_hub(&self, on_phase: OnPhase) -> Result<Ready> {
if let Some(ready) = self.current() {
return Ok(ready);
}
let node = toolchain::node().context(
"Local device support needs Node.js 22 or newer. Install it, then restart Vectrum Code.",
)?;
if !toolchain::is_installed(Tool::Hub) {
on_phase(HostStatus::Installing, Some(toolchain::install_message(Tool::Hub)));
}
let entry = toolchain::ensure(Tool::Hub).context("installing device support")?;
on_phase(HostStatus::Starting, None);
let hub = spawn_hub(&entry, &node)?;
let ready = Ready { origin: hub.origin.clone(), node: node.clone(), helpers: helpers() };
let pid = hub.pid;
{
let mut running = self.lock_running();
running.hub = Some(hub);
running.node = Some(node);
running.restart_delay = Duration::ZERO;
}
toolchain::prune(Tool::Hub);
supervise(self.running.clone(), pid, entry);
Ok(ready)
}
/// Installs and starts agent-device once agent access is granted.
pub fn ensure_agent_ready(&self, on_phase: OnPhase) -> Result<(Ready, AgentEndpoint)> {
let _guard = self.lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
let ready = self.ensure_hub(on_phase)?;
if let Some(agent) = self.lock_running().agent.clone() {
return Ok((ready, agent));
}
if !toolchain::is_installed(Tool::Agent) {
on_phase(HostStatus::Installing, Some(toolchain::install_message(Tool::Agent)));
}
let entry = toolchain::ensure(Tool::Agent).context("installing agent tools")?;
on_phase(HostStatus::Starting, None);
let agent = start_daemon(&entry, &ready.node)?;
toolchain::prune(Tool::Agent);
self.lock_running().agent = Some(agent.clone());
Ok((ready, agent))
}
/// Stops only agent-device; watching through the hub carries on.
pub fn stop_agent(&self) {
let _guard = self.lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
let node = self.lock_running().node.clone();
stop_daemon(node.as_deref());
self.lock_running().agent = None;
}
/// Stops the helpers. Simulators and emulators keep running; the user owns those.
pub fn stop(&self) {
let _guard = self.lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
let (hub, node) = {
let mut running = self.lock_running();
running.agent = None;
(running.hub.take(), running.node.clone())
};
if let Some(hub) = hub {
hub.kill();
}
if let Some(dir) = state_dir() {
let _ = std::fs::remove_file(dir.join("hub.json"));
}
stop_daemon(node.as_deref());
}
}
/// Restarts the hub when it dies under us, unless it was replaced or stopped.
fn supervise(running: Arc<Mutex<Running>>, pid: u32, entry: PathBuf) {
let _ = thread::Builder::new().name("device-hub-supervisor".into()).spawn(move || {
loop {
thread::sleep(Duration::from_millis(500));
let delay = {
let mut guard = running.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
let Some(hub) = guard.hub.as_ref().filter(|hub| hub.pid == pid) else { return };
if hub.alive() {
continue;
}
let uptime = hub.started.elapsed();
let delay = if uptime >= HUB_STABLE_UPTIME { Duration::ZERO } else { guard.restart_delay };
guard.restart_delay = if uptime >= HUB_STABLE_UPTIME {
Duration::ZERO
} else if guard.restart_delay.is_zero() {
Duration::from_secs(1)
} else {
(guard.restart_delay * 2).min(HUB_RESTART_MAX)
};
delay
};
thread::sleep(delay);
let mut guard = running.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
if guard.hub.as_ref().is_none_or(|hub| hub.pid != pid) {
return;
}
let Some(node) = guard.node.clone() else { return };
match spawn_hub(&entry, &node) {
Ok(hub) => {
let next = hub.pid;
guard.hub = Some(hub);
drop(guard);
// Hand over to a supervisor for the replacement.
return supervise(running.clone(), next, entry);
}
Err(_) => {
guard.hub = None;
return;
}
}
}
});
}
fn helpers() -> Helpers {
let Some(dist) = toolchain::hub_package_dir().map(|dir| dir.join("vendor/serve-sim/dist")) else {
return Helpers::default();
};
let existing = |path: PathBuf| path.exists().then_some(path);
Helpers {
ax_settings: existing(dist.join("simax/serve-sim-ax-settings")),
serve_sim_cli: existing(dist.join("serve-sim.js")),
}
}
fn pid_alive(pid: i32) -> bool {
unsafe { libc::kill(pid, 0) == 0 }
}
/// Stops a hub a previous run left behind, if that pid is still that hub.
fn reap_stale_hub(state: &Path) {
let file = state.join("hub.json");
let Some(previous) = std::fs::read_to_string(&file).ok().and_then(|text| serde_json::from_str::<HubStateFile>(&text).ok())
else {
return;
};
let cmdline = std::fs::read(format!("/proc/{}/cmdline", previous.pid)).unwrap_or_default();
if pid_alive(previous.pid as i32) && String::from_utf8_lossy(&cmdline).contains(&previous.entry_path) {
unsafe {
libc::kill(previous.pid as i32, libc::SIGTERM);
}
}
let _ = std::fs::remove_file(file);
}
fn free_port() -> Result<u16> {
let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).context("reserving a port for the device hub")?;
Ok(listener.local_addr()?.port())
}
fn spawn_hub(entry: &Path, node: &Path) -> Result<HubProcess> {
let state = state_dir().context("no data folder")?;
std::fs::create_dir_all(&state)?;
reap_stale_hub(&state);
let port = free_port()?;
let origin = format!("http://127.0.0.1:{port}");
let mut command = Command::new(node);
command
.arg(entry)
.args(["--port", &port.to_string(), "--host", "127.0.0.1", "--hide-sidebar", "--hide-boot-device"])
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
apply_env(&mut command, &android_sdk());
// The hub goes with the app if the app dies without stopping it.
unsafe {
command.pre_exec(|| {
libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGTERM);
Ok(())
});
}
let mut child = command.spawn().context("starting the device hub")?;
let pid = child.id();
// Its output goes to a log beside its state, for diagnosing a hub that won't start.
let log = Arc::new(Mutex::new(std::fs::File::create(state.join("hub.log")).ok()));
for pipe in [child.stdout.take().map(|p| Box::new(p) as Box<dyn std::io::Read + Send>), child.stderr.take().map(|p| Box::new(p) as Box<dyn std::io::Read + Send>)].into_iter().flatten() {
let log = log.clone();
thread::spawn(move || {
for line in BufReader::new(pipe).lines().map_while(Result::ok) {
if let Ok(mut log) = log.lock()
&& let Some(file) = log.as_mut()
{
let _ = writeln!(file, "{line}");
}
}
});
}
let hub = HubProcess { child: Arc::new(Mutex::new(child)), pid, origin: origin.clone(), started: Instant::now() };
let deadline = Instant::now() + HUB_READY_TIMEOUT;
let agent = ureq::AgentBuilder::new().timeout(Duration::from_secs(2)).build();
loop {
if agent.get(&format!("{origin}/readyz")).call().is_ok() {
break;
}
if !hub.alive() || Instant::now() > deadline {
hub.kill();
let tail = std::fs::read_to_string(state.join("hub.log")).unwrap_or_default();
let tail: Vec<&str> = tail.lines().rev().take(5).collect();
bail!(
"The device hub did not start.{}",
if tail.is_empty() { String::new() } else { format!("\n{}", tail.into_iter().rev().collect::<Vec<_>>().join("\n")) }
);
}
thread::sleep(Duration::from_millis(150));
}
let record = HubStateFile { pid, port, entry_path: entry.display().to_string() };
let _ = std::fs::write(state.join("hub.json"), serde_json::to_string(&record).unwrap_or_default());
Ok(hub)
}
fn read_daemon_file() -> Option<DaemonFile> {
let text = std::fs::read_to_string(state_dir()?.join("daemon.json")).ok()?;
serde_json::from_str(&text).ok()
}
fn daemon_command(node: &Path, entry: &Path, state: &Path) -> Command {
let mut command = Command::new(node);
command
.arg(entry)
.env("AGENT_DEVICE_STATE_DIR", state)
.env("AGENT_DEVICE_DAEMON_SERVER_MODE", "http")
// The daemon idles out after five minutes by default; the app owns
// its lifetime and stops it itself.
.env("AGENT_DEVICE_DAEMON_IDLE_TIMEOUT_MS", "0")
.env("AGENT_DEVICE_NO_UPDATE_NOTIFIER", "1")
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null());
apply_env(&mut command, &android_sdk());
command
}
/// The first agent-device command in a state folder starts the daemon and
/// waits for it; `devices` is the cheapest. Its daemon.json says where it is.
fn start_daemon(entry: &Path, node: &Path) -> Result<AgentEndpoint> {
let state = state_dir().context("no data folder")?;
std::fs::create_dir_all(&state)?;
let endpoint = |file: DaemonFile| AgentEndpoint {
base_url: format!("http://127.0.0.1:{}", file.http_port),
token: file.token,
};
if let Some(existing) = read_daemon_file() {
let health = ureq::AgentBuilder::new()
.timeout(Duration::from_secs(2))
.build()
.get(&format!("http://127.0.0.1:{}/health", existing.http_port))
.call();
if health.is_ok_and(|response| response.status() == 200) {
return Ok(endpoint(existing));
}
let _ = std::fs::remove_file(state.join("daemon.json"));
}
let mut command = daemon_command(node, entry, &state);
command.args(["devices", "--json"]);
if let Ok(mut child) = command.spawn() {
let deadline = Instant::now() + DAEMON_READY_TIMEOUT;
while matches!(child.try_wait(), Ok(None)) && Instant::now() < deadline {
thread::sleep(Duration::from_millis(100));
}
let _ = child.kill();
let _ = child.wait();
}
let deadline = Instant::now() + DAEMON_READY_TIMEOUT;
loop {
if let Some(file) = read_daemon_file() {
return Ok(endpoint(file));
}
if Instant::now() > deadline {
bail!("Agent tools did not start within {} ms.", DAEMON_READY_TIMEOUT.as_millis());
}
thread::sleep(Duration::from_millis(100));
}
}
fn stop_daemon(node: Option<&Path>) {
let (Some(node), Some(entry), Some(state)) = (node, toolchain::entry_path(Tool::Agent), state_dir()) else { return };
if !entry.is_file() || read_daemon_file().is_none() {
return;
}
let mut command = daemon_command(node, &entry, &state);
command.args(["daemon", "stop", "--state-dir"]).arg(&state);
if let Ok(mut child) = command.spawn() {
let deadline = Instant::now() + Duration::from_secs(10);
while matches!(child.try_wait(), Ok(None)) && Instant::now() < deadline {
thread::sleep(Duration::from_millis(100));
}
let _ = child.kill();
let _ = child.wait();
}
}