#[cfg(not(unix))]
use crate::error::PlatformError;
use anyhow::{Context, Result};
use rch_common::{
ApiError, ApiResponse, ErrorCode, RequiredRuntime, WorkerCapabilities, WorkerConfig,
classify_command_detailed,
};
#[cfg(unix)]
use rch_common::{SshClient, SshOptions};
use std::path::{Path, PathBuf};
use crate::status_types::{
DaemonFullStatusResponse, SpeedScoreListResponseFromApi, SpeedScoreResponseFromApi,
SpeedScoreViewFromApi, WorkerCapabilitiesFromApi, WorkerCapabilitiesResponseFromApi,
WorkerStatusFromApi, extract_json_body,
};
use crate::ui::context::OutputContext;
use crate::ui::progress::MultiProgressManager;
use crate::ui::theme::{StatusIndicator, Theme};
use tracing::debug;
use super::helpers::{
classify_ssh_error, configured_socket_path, format_ssh_report, indent_lines,
major_version_mismatch, runtime_label, rust_version_mismatch, send_daemon_command,
ssh_error_code, urlencoding_encode,
};
use super::helpers::{config_dir, load_workers_from_config};
use super::types::{
WorkerActionResponse, WorkerBenchmarkResult, WorkerDiskInfo, WorkerInfo, WorkerProbeResult,
WorkerProbeSummary, WorkersCapabilitiesReport, WorkersListResponse, WorkersProbeResponse,
};
use crate::hook::required_runtime_for_kind;
use crate::toolchain::{detect_declared_components, detect_toolchain};
pub(super) fn has_any_capabilities(capabilities: &WorkerCapabilities) -> bool {
capabilities.rustc_version.is_some()
|| !capabilities.rustup_toolchains.is_empty()
|| !capabilities.rustup_components.is_empty()
|| capabilities.bun_version.is_some()
|| capabilities.node_version.is_some()
|| capabilities.npm_version.is_some()
|| !capabilities.tools_present.is_empty()
|| !capabilities.tools_absent.is_empty()
}
fn format_named_tool_matrix(capabilities: &WorkerCapabilities) -> String {
if capabilities.tools_present.is_empty() && capabilities.tools_absent.is_empty() {
return "none declared".to_string();
}
let mut parts = Vec::new();
if !capabilities.tools_present.is_empty() {
parts.push(format!(
"verified: {}",
capabilities.tools_present.join(", ")
));
}
if !capabilities.tools_absent.is_empty() {
parts.push(format!("failed: {}", capabilities.tools_absent.join(", ")));
}
parts.join("; ")
}
#[cfg(unix)]
async fn probe_connected_worker_capabilities(client: &mut SshClient) -> Result<WorkerCapabilities> {
let output = client
.execute("if command -v rch-wkr >/dev/null 2>&1; then rch-wkr capabilities; else ~/.local/bin/rch-wkr capabilities; fi")
.await
.context("worker capability command failed")?;
if !output.success() {
anyhow::bail!(
"worker capability command exited {}: {}",
output.exit_code,
output.stderr.trim()
);
}
serde_json::from_str(&output.stdout).context("worker capability JSON was invalid")
}
fn missing_declared_components(
capabilities: &WorkerCapabilities,
toolchain: Option<&str>,
declared_components: &[String],
) -> Vec<String> {
let Some(toolchain) = toolchain else {
return declared_components.to_vec();
};
declared_components
.iter()
.filter(|component| !capabilities.has_rustup_component(toolchain, component))
.cloned()
.collect()
}
fn format_rustup_component_matrix(capabilities: &WorkerCapabilities) -> String {
if capabilities.rustup_components.is_empty() {
"unknown".to_string()
} else {
capabilities.rustup_components.join(", ")
}
}
pub(super) async fn probe_local_capabilities() -> WorkerCapabilities {
fn rustc_version_command() -> tokio::process::Command {
let mut command = tokio::process::Command::new("rustc");
command.arg("--version");
command
}
fn bun_version_command() -> tokio::process::Command {
let mut command = tokio::process::Command::new("bun");
command.arg("--version");
command
}
fn node_version_command() -> tokio::process::Command {
let mut command = tokio::process::Command::new("node");
command.arg("--version");
command
}
fn npm_version_command() -> tokio::process::Command {
let mut command = tokio::process::Command::new("npm");
command.arg("--version");
command
}
async fn run_version(mut command: tokio::process::Command) -> Option<String> {
let output = command.output().await.ok()?;
if !output.status.success() {
return None;
}
let stdout = String::from_utf8_lossy(&output.stdout);
let trimmed = stdout.trim();
if trimmed.is_empty() {
None
} else {
Some(trimmed.to_string())
}
}
let (rustc, bun, node, npm) = tokio::join!(
run_version(rustc_version_command()),
run_version(bun_version_command()),
run_version(node_version_command()),
run_version(npm_version_command()),
);
let mut caps = WorkerCapabilities::new();
caps.rustc_version = rustc;
caps.bun_version = bun;
caps.node_version = node;
caps.npm_version = npm;
caps
}
pub(super) fn collect_local_capability_warnings(
workers: &[WorkerCapabilitiesFromApi],
local: &WorkerCapabilities,
) -> Vec<String> {
let mut warnings = Vec::new();
if let Some(local_rust) = local.rustc_version.as_ref() {
let missing: Vec<String> = workers
.iter()
.filter(|worker| !worker.capabilities.has_rust())
.map(|worker| worker.id.clone())
.collect();
if !missing.is_empty() {
warnings.push(format!(
"Workers missing Rust runtime (local: {}): {}",
local_rust,
missing.join(", ")
));
}
let mismatched: Vec<String> = workers
.iter()
.filter_map(|worker| {
let remote = worker.capabilities.rustc_version.as_ref()?;
if rust_version_mismatch(local_rust, remote) {
Some(format!("{} ({})", worker.id, remote))
} else {
None
}
})
.collect();
if !mismatched.is_empty() {
warnings.push(format!(
"Rust version mismatch vs local {}: {}",
local_rust,
mismatched.join(", ")
));
}
}
if let Some(local_bun) = local.bun_version.as_ref() {
let missing: Vec<String> = workers
.iter()
.filter(|worker| !worker.capabilities.has_bun())
.map(|worker| worker.id.clone())
.collect();
if !missing.is_empty() {
warnings.push(format!(
"Workers missing Bun runtime (local: {}): {}",
local_bun,
missing.join(", ")
));
}
let mismatched: Vec<String> = workers
.iter()
.filter_map(|worker| {
let remote = worker.capabilities.bun_version.as_ref()?;
if major_version_mismatch(local_bun, remote) {
Some(format!("{} ({})", worker.id, remote))
} else {
None
}
})
.collect();
if !mismatched.is_empty() {
warnings.push(format!(
"Bun major version mismatch vs local {}: {}",
local_bun,
mismatched.join(", ")
));
}
}
if let Some(local_node) = local.node_version.as_ref() {
let missing: Vec<String> = workers
.iter()
.filter(|worker| !worker.capabilities.has_node())
.map(|worker| worker.id.clone())
.collect();
if !missing.is_empty() {
warnings.push(format!(
"Workers missing Node runtime (local: {}): {}",
local_node,
missing.join(", ")
));
}
let mismatched: Vec<String> = workers
.iter()
.filter_map(|worker| {
let remote = worker.capabilities.node_version.as_ref()?;
if major_version_mismatch(local_node, remote) {
Some(format!("{} ({})", worker.id, remote))
} else {
None
}
})
.collect();
if !mismatched.is_empty() {
warnings.push(format!(
"Node major version mismatch vs local {}: {}",
local_node,
mismatched.join(", ")
));
}
}
if let Some(local_npm) = local.npm_version.as_ref() {
let missing: Vec<String> = workers
.iter()
.filter(|worker| worker.capabilities.npm_version.is_none())
.map(|worker| worker.id.clone())
.collect();
if !missing.is_empty() {
warnings.push(format!(
"Workers missing npm runtime (local: {}): {}",
local_npm,
missing.join(", ")
));
}
let mismatched: Vec<String> = workers
.iter()
.filter_map(|worker| {
let remote = worker.capabilities.npm_version.as_ref()?;
if major_version_mismatch(local_npm, remote) {
Some(format!("{} ({})", worker.id, remote))
} else {
None
}
})
.collect();
if !mismatched.is_empty() {
warnings.push(format!(
"npm major version mismatch vs local {}: {}",
local_npm,
mismatched.join(", ")
));
}
}
warnings
}
pub(super) fn collect_refresh_warnings(workers: &[WorkerCapabilitiesFromApi]) -> Vec<String> {
workers
.iter()
.filter_map(|worker| {
let refresh = worker.refresh.as_ref()?;
if !refresh.attempted || refresh.live {
return None;
}
let detail = refresh
.message
.as_deref()
.unwrap_or("daemon returned cached capability snapshot");
Some(format!(
"Worker {} capabilities are cached after refresh attempt: {}",
worker.id, detail
))
})
.collect()
}
fn workers_list_verbose_enabled(ctx: &OutputContext) -> bool {
ctx.is_verbose() && !ctx.is_json()
}
async fn query_worker_disk_status() -> Option<DaemonFullStatusResponse> {
match tokio::time::timeout(
std::time::Duration::from_secs(1),
send_daemon_command("GET /status\n"),
)
.await
{
Ok(Ok(response)) => {
extract_json_body(&response).and_then(|json| serde_json::from_str(json).ok())
}
Ok(Err(_)) | Err(_) => None,
}
}
fn valid_disk_free(value: Option<f64>) -> Option<f64> {
value.filter(|value| value.is_finite() && *value >= 0.0)
}
fn disk_info_from_daemon(worker: Option<&WorkerStatusFromApi>) -> WorkerDiskInfo {
let Some(worker) = worker else {
return WorkerDiskInfo::default();
};
let disk_free_gb = valid_disk_free(worker.pressure_disk_free_gb);
let disk_free_ratio = worker
.pressure_disk_free_ratio
.filter(|ratio| ratio.is_finite() && (0.0..=1.0).contains(ratio));
WorkerDiskInfo {
disk_free_gb,
disk_free_ratio,
build_disk_free_gb: valid_disk_free(worker.pressure_build_disk_free_gb),
disk_measurement_source: (disk_free_gb.is_some() || disk_free_ratio.is_some())
.then_some("daemon"),
disk_pressure_state: worker.pressure_state.clone(),
disk_pressure_reason: worker.pressure_reason_code.clone(),
disk_pressure_source: worker.pressure_state.as_ref().map(|_| "daemon"),
}
}
#[cfg(any(unix, test))]
fn disk_info_from_probe(
capabilities: Option<&WorkerCapabilities>,
daemon_worker: Option<&WorkerStatusFromApi>,
) -> WorkerDiskInfo {
let mut disk = disk_info_from_daemon(daemon_worker);
disk.disk_free_gb = capabilities.and_then(|caps| valid_disk_free(caps.disk_free_gb));
disk.disk_free_ratio = capabilities.and_then(|caps| {
let free = disk.disk_free_gb?;
let total = caps
.disk_total_gb
.filter(|total| total.is_finite() && *total > 0.0)?;
(free <= total).then_some(free / total)
});
disk.build_disk_free_gb =
capabilities.and_then(|caps| valid_disk_free(caps.build_disk_free_gb));
disk.disk_measurement_source = disk.disk_free_gb.map(|_| "probe");
disk
}
fn format_worker_disk(disk: &WorkerDiskInfo) -> String {
let free = disk.disk_free_gb.map_or_else(
|| "unknown".to_string(),
|free| format!("{free:.1} GiB free"),
);
let ratio = disk.disk_free_ratio.map_or_else(
|| "unknown free %".to_string(),
|ratio| format!("{:.1}% free", ratio * 100.0),
);
let measurement_source = disk.disk_measurement_source.unwrap_or("unknown source");
let build = disk
.build_disk_free_gb
.filter(|build| {
disk.disk_free_gb
.is_none_or(|free| (build - free).abs() >= 0.05)
})
.map_or_else(String::new, |build| {
format!("; build disk: {build:.1} GiB free")
});
let pressure = match disk.disk_pressure_state.as_deref() {
Some("warning") => "WARNING",
Some("critical") => "CRITICAL",
Some("telemetry_gap") | None => "unknown",
Some(state) => state,
};
let pressure_source = disk
.disk_pressure_source
.map_or_else(String::new, |source| format!(" (cached {source})"));
format!(
"Disk: {free} ({ratio}, {measurement_source}){build}; pressure: {pressure}{pressure_source}"
)
}
fn apply_live_worker_slots(
workers: &mut [WorkerConfig],
daemon_status: Option<&DaemonFullStatusResponse>,
) {
let Some(status) = daemon_status else {
return;
};
for worker in workers {
if let Some(live) = status
.workers
.iter()
.find(|live| live.id == worker.id.as_str())
{
worker.total_slots = live.total_slots;
}
}
}
fn render_worker_verbose_lines(
worker: &WorkerConfig,
daemon_status: Option<&DaemonFullStatusResponse>,
style: &Theme,
) -> Vec<String> {
let mut lines = Vec::new();
if let Some(status) = daemon_status {
if let Some(worker_status) = status.workers.iter().find(|w| w.id == worker.id.as_str()) {
let circuit_display = match worker_status.circuit_state.as_str() {
"closed" => style.success("Closed"),
"half_open" => style.warning("HalfOpen"),
"open" => style.error("Open"),
other => style.muted(other),
};
lines.push(format!(
" {} {} {}",
style.key("Circuit"),
style.muted(":"),
circuit_display
));
let slots_display = if worker_status.used_slots > 0 {
style.warning(&format!(
"{}/{}",
worker_status.used_slots, worker_status.total_slots
))
} else {
style.success(&format!(
"{}/{}",
worker_status.used_slots, worker_status.total_slots
))
};
lines.push(format!(
" {} {} {}",
style.key("In Use"),
style.muted(":"),
slots_display
));
let status_display = match worker_status.status.as_str() {
"healthy" => style.success("Healthy"),
"degraded" => style.warning("Degraded"),
"unreachable" => style.error("Unreachable"),
"draining" => style.warning("Draining"),
"drained" => style.info("Drained"),
"disabled" => style.muted("Disabled"),
other => style.muted(other),
};
lines.push(format!(
" {} {} {}",
style.key("Status"),
style.muted(":"),
status_display
));
if let Some(ref last_error) = worker_status.last_error {
lines.push(format!(
" {} {} {}",
style.key("LastErr"),
style.muted(":"),
style.error(last_error)
));
}
if let Some(recovery_secs) = worker_status.recovery_in_secs {
lines.push(format!(
" {} {} {}s",
style.key("Recover"),
style.muted(":"),
style.info(&recovery_secs.to_string())
));
}
} else {
lines.push(format!(
" {} {}",
style.muted("Live status:"),
style.muted("(not in daemon)")
));
}
} else {
lines.push(format!(
" {} {}",
style.muted("Live status:"),
style.muted("(daemon not running)")
));
}
lines.push(format!(
" {} {} {}",
style.key("SSH Key"),
style.muted(":"),
style.muted(&worker.identity_file)
));
lines
}
#[cfg(test)]
pub(super) fn summarize_capabilities(capabilities: &WorkerCapabilities) -> String {
let mut parts = Vec::new();
if let Some(rustc) = capabilities.rustc_version.as_ref() {
parts.push(format!("rustc {}", rustc));
}
if let Some(bun) = capabilities.bun_version.as_ref() {
parts.push(format!("bun {}", bun));
}
if let Some(node) = capabilities.node_version.as_ref() {
parts.push(format!("node {}", node));
}
if let Some(npm) = capabilities.npm_version.as_ref() {
parts.push(format!("npm {}", npm));
}
if parts.is_empty() {
"unknown".to_string()
} else {
parts.join(", ")
}
}
pub(super) async fn query_workers_capabilities(
refresh: bool,
) -> Result<WorkerCapabilitiesResponseFromApi> {
let command = if refresh {
"GET /workers/capabilities?refresh=true\n"
} else {
"GET /workers/capabilities\n"
};
let response = send_daemon_command(command).await?;
let json = extract_json_body(&response)
.ok_or_else(|| anyhow::anyhow!("Invalid response format from daemon"))?;
let capabilities: WorkerCapabilitiesResponseFromApi =
serde_json::from_str(json).context("Failed to parse worker capabilities response")?;
Ok(capabilities)
}
async fn query_speedscore_list() -> Result<SpeedScoreListResponseFromApi> {
let response = send_daemon_command("GET /speedscores\n").await?;
let json = extract_json_body(&response)
.ok_or_else(|| anyhow::anyhow!("Invalid response format from daemon"))?;
let scores: SpeedScoreListResponseFromApi =
serde_json::from_str(json).context("Failed to parse SpeedScore list response")?;
Ok(scores)
}
pub async fn workers_list(show_speedscore: bool, ctx: &OutputContext) -> Result<()> {
let mut workers = load_workers_from_config()?;
let style = ctx.theme();
let speedscores = if show_speedscore {
query_speedscore_list().await.ok()
} else {
None
};
let daemon_status = if workers.is_empty() {
None
} else {
query_worker_disk_status().await
};
apply_live_worker_slots(&mut workers, daemon_status.as_ref());
let mut worker_infos: Vec<WorkerInfo> = workers.iter().map(WorkerInfo::from).collect();
for info in &mut worker_infos {
let live = daemon_status
.as_ref()
.and_then(|status| status.workers.iter().find(|worker| worker.id == info.id));
info.disk = disk_info_from_daemon(live);
}
if ctx.is_json() {
if let Some(ref scores) = speedscores {
for info in &mut worker_infos {
if let Some(score_entry) = scores.workers.iter().find(|s| s.worker_id == info.id)
&& let Some(ref score) = score_entry.speedscore
{
info.speedscore = Some(score.total);
}
}
}
let response = WorkersListResponse {
count: workers.len(),
workers: worker_infos,
};
let _ = ctx.json(&ApiResponse::ok("workers list", response));
return Ok(());
}
if workers.is_empty() {
let config_path = config_dir()
.map(|d| d.join("workers.toml"))
.unwrap_or_else(|| PathBuf::from("~/.config/rch/workers.toml"));
println!(" {} No workers configured.", style.symbols.info);
println!();
println!(
" Create a workers config at: {}",
style.value(&config_path.display().to_string())
);
println!();
println!(
" Run {} to generate example configuration.",
style.highlight("rch config init")
);
return Ok(());
}
println!("{}", style.format_header("Configured Workers"));
println!();
for (worker, info) in workers.iter().zip(&worker_infos) {
println!(
" {} {} {}@{}",
style.symbols.bullet_filled,
style.highlight(worker.id.as_str()),
style.muted(&worker.user),
style.info(&worker.host)
);
let mut stats_line = format!(
" {} {} {} {} {} {}",
style.key("Slots"),
style.muted(":"),
style.value(&worker.total_slots.to_string()),
style.key("Priority"),
style.muted(":"),
style.value(&worker.priority.to_string())
);
if let Some(ref scores) = speedscores
&& let Some(score_entry) = scores
.workers
.iter()
.find(|s| s.worker_id == worker.id.as_str())
&& let Some(ref score) = score_entry.speedscore
{
let score_color = match score.total {
x if x >= 75.0 => style.success(&format!("{:.0}", x)),
x if x >= 45.0 => style.warning(&format!("{:.0}", x)),
x => style.error(&format!("{:.0}", x)),
};
stats_line.push_str(&format!(
" {} {} {}",
style.key("Score"),
style.muted(":"),
score_color
));
}
println!("{}", stats_line);
println!(" {}", format_worker_disk(&info.disk));
if !worker.tags.is_empty() {
println!(
" {} {} {}",
style.key("Tags"),
style.muted(":"),
style.muted(&worker.tags.join(", "))
);
}
if workers_list_verbose_enabled(ctx) {
debug!(target: "rch::verbose", worker = %worker.id, "rendering verbose worker details");
for line in render_worker_verbose_lines(worker, daemon_status.as_ref(), style) {
println!("{line}");
}
}
println!();
}
println!(
"{} {} worker(s)",
style.muted("Total:"),
style.highlight(&workers.len().to_string())
);
Ok(())
}
pub async fn workers_capabilities(
command: Option<String>,
refresh: bool,
ctx: &OutputContext,
) -> Result<()> {
let style = ctx.theme();
let response = query_workers_capabilities(refresh).await?;
let workers = response.workers;
let local_capabilities = probe_local_capabilities().await;
let local_has_any = has_any_capabilities(&local_capabilities);
let mut warnings = Vec::new();
let mut required_runtime = None;
if let Some(command) = command.as_deref() {
let details = classify_command_detailed(command);
if !details.classification.is_compilation {
warnings.push(format!(
"Command '{}' is not a compilation command: {}",
command, details.classification.reason
));
}
let runtime = required_runtime_for_kind(details.classification.kind);
if runtime != RequiredRuntime::None {
required_runtime = Some(runtime);
}
}
if let Some(runtime) = required_runtime {
let missing: Vec<String> = workers
.iter()
.filter(|worker| {
let caps = &worker.capabilities;
match runtime {
RequiredRuntime::Rust => !caps.has_rust(),
RequiredRuntime::Bun => !caps.has_bun(),
RequiredRuntime::Node => !caps.has_node(),
RequiredRuntime::Nix => !caps.has_nix(),
RequiredRuntime::Go => !caps.has_go(),
RequiredRuntime::Zig => !caps.has_zig(),
RequiredRuntime::None => false,
}
})
.map(|worker| worker.id.clone())
.collect();
if !missing.is_empty() {
warnings.push(format!(
"Workers missing required runtime {}: {}",
runtime_label(&runtime),
missing.join(", ")
));
}
}
if local_has_any {
warnings.extend(collect_local_capability_warnings(
&workers,
&local_capabilities,
));
}
warnings.extend(collect_refresh_warnings(&workers));
if ctx.is_json() {
let report = WorkersCapabilitiesReport {
workers,
local: Some(local_capabilities),
required_runtime,
warnings,
};
let _ = ctx.json(&ApiResponse::ok("workers capabilities", report));
return Ok(());
}
if workers.is_empty() {
println!(
"{} {}",
StatusIndicator::Warning.display(style),
style.warning("No workers configured")
);
return Ok(());
}
println!("{}", style.format_header("Worker Capabilities"));
println!();
let key_width = ["Rust", "Bun", "Node", "npm"]
.iter()
.map(|label| label.len())
.max()
.unwrap_or(4);
println!("{}", style.highlight("Local Capabilities"));
let render = |label: &str, value: Option<&String>| {
let (indicator, display) = if let Some(ver) = value {
(StatusIndicator::Success, style.value(ver))
} else {
(StatusIndicator::Warning, style.warning("unknown"))
};
let padded_label = format!("{label:width$}", width = key_width);
println!(
" {} {} {} {}",
indicator.display(style),
style.key(&padded_label),
style.muted(":"),
display
);
};
render("Rust", local_capabilities.rustc_version.as_ref());
render("Bun", local_capabilities.bun_version.as_ref());
render("Node", local_capabilities.node_version.as_ref());
render("npm", local_capabilities.npm_version.as_ref());
if !local_has_any {
println!(
" {} {}",
StatusIndicator::Warning.display(style),
style.warning("No local runtimes detected")
);
}
println!();
if let Some(runtime) = required_runtime.as_ref() {
println!(
"{} {}",
style.key("Required runtime:"),
style.value(runtime_label(runtime))
);
println!();
}
for worker in &workers {
println!(
" {} {} {}@{}",
style.symbols.bullet_filled,
style.highlight(&worker.id),
style.muted(&worker.user),
style.info(&worker.host)
);
let caps = &worker.capabilities;
render("Rust", caps.rustc_version.as_ref());
render("Bun", caps.bun_version.as_ref());
render("Node", caps.node_version.as_ref());
render("npm", caps.npm_version.as_ref());
if let Some(refresh) = worker.refresh.as_ref() {
let (indicator, label) = if refresh.live {
(StatusIndicator::Success, style.value("live refresh"))
} else {
(
StatusIndicator::Warning,
style.warning("cached after failed refresh"),
)
};
println!(
" {} {} {} {}",
indicator.display(style),
style.key("Refresh"),
style.muted(":"),
label
);
}
println!();
}
if !warnings.is_empty() {
println!("{}", style.format_header("Warnings"));
for warning in warnings {
println!(
" {} {}",
StatusIndicator::Warning.display(style),
style.warning(&warning)
);
}
}
Ok(())
}
#[cfg(not(unix))]
pub async fn workers_probe(
_worker_id: Option<String>,
_all: bool,
_ctx: &OutputContext,
) -> Result<()> {
Err(PlatformError::UnixOnly {
feature: "worker probe".to_string(),
})?
}
#[cfg(unix)]
pub async fn workers_probe(
worker_id: Option<String>,
all: bool,
ctx: &OutputContext,
) -> Result<()> {
let workers = load_workers_from_config()?;
let style = ctx.theme();
let project_root = std::env::current_dir().ok();
let declared_components = project_root
.as_deref()
.map(detect_declared_components)
.unwrap_or_default();
let declared_toolchain = project_root
.as_deref()
.and_then(|root| detect_toolchain(root).ok())
.map(|toolchain| toolchain.rustup_toolchain());
if workers.is_empty() {
if ctx.is_json() {
let _ = ctx.json(&ApiResponse::<WorkersProbeResponse>::ok(
"workers probe",
WorkersProbeResponse {
results: vec![],
summary: WorkerProbeSummary::default(),
},
));
}
return Ok(());
}
let targets: Vec<&WorkerConfig> = if all {
workers.iter().collect()
} else if let Some(id) = &worker_id {
workers.iter().filter(|w| w.id.as_str() == id).collect()
} else {
if ctx.is_json() {
let _ = ctx.json(&ApiResponse::<()>::err(
"workers probe",
ApiError::new(
ErrorCode::ConfigValidationError,
"Specify a worker ID or use --all",
),
));
} else {
println!(
"{} Specify a worker ID or use {} to probe all workers.",
StatusIndicator::Info.display(style),
style.highlight("--all")
);
}
return Ok(());
};
if targets.is_empty() {
if let Some(id) = worker_id {
if ctx.is_json() {
let _ = ctx.json(&ApiResponse::<()>::err(
"workers probe",
ApiError::new(
ErrorCode::ConfigInvalidWorker,
format!("Worker '{}' not found", id),
),
));
} else {
println!(
"{} Worker '{}' not found in configuration.",
StatusIndicator::Warning.display(style),
style.highlight(&id)
);
}
}
return Ok(());
}
let daemon_status = query_worker_disk_status().await;
let mut results = Vec::new();
if !ctx.is_json() {
println!(
"Probing {} worker(s)...\n",
style.highlight(&targets.len().to_string())
);
}
for worker in targets {
let daemon_worker = daemon_status.as_ref().and_then(|status| {
status
.workers
.iter()
.find(|live| live.id == worker.id.as_str())
});
if !ctx.is_json() {
print!(
" {} {}@{}... ",
style.highlight(worker.id.as_str()),
style.muted(&worker.user),
style.info(&worker.host)
);
}
let ssh_options = SshOptions::default();
let mut client = SshClient::new(worker.clone(), ssh_options.clone());
match client.connect().await {
Ok(()) => {
let start = std::time::Instant::now();
match client.health_check().await {
Ok(true) => {
let latency =
u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX);
let capability_result =
probe_connected_worker_capabilities(&mut client).await;
let capabilities = capability_result.as_ref().ok().cloned();
let disk = disk_info_from_probe(capabilities.as_ref(), daemon_worker);
let missing_components = capabilities.as_ref().map_or_else(
|| declared_components.clone(),
|capabilities| {
missing_declared_components(
capabilities,
declared_toolchain.as_deref(),
&declared_components,
)
},
);
let capability_error =
capability_result.as_ref().err().map(ToString::to_string);
let (status, error, error_code) = if let Some(error) = capability_error {
(
"capability_unknown".to_string(),
Some(error),
Some(ErrorCode::WorkerMissingToolchain.code_string()),
)
} else if !missing_components.is_empty() {
let toolchain = declared_toolchain.as_deref().unwrap_or("<unknown>");
(
"capability_missing".to_string(),
Some(format!(
"missing rustup component(s) for {toolchain}: {}",
missing_components.join(", ")
)),
Some(ErrorCode::WorkerMissingToolchain.code_string()),
)
} else {
("ok".to_string(), None, None)
};
results.push(WorkerProbeResult {
id: worker.id.as_str().to_string(),
host: worker.host.clone(),
status: status.clone(),
latency_ms: Some(latency),
error: error.clone(),
error_code: error_code.clone(),
capabilities: capabilities.clone(),
missing_components: missing_components.clone(),
disk: disk.clone(),
});
if !ctx.is_json() {
if status == "ok" {
println!(
"{} ({}ms) {}",
StatusIndicator::Success.with_label(style, "OK"),
style.muted(&latency.to_string()),
format_worker_disk(&disk)
);
} else {
println!(
"{} ({}ms) [{}] {}",
StatusIndicator::Warning
.with_label(style, "CAPABILITY_MISSING"),
style.muted(&latency.to_string()),
style.highlight(error_code.as_deref().unwrap_or("RCH-E205")),
format_worker_disk(&disk)
);
if let Some(error) = error.as_deref() {
println!(" {}", style.warning(error));
}
}
if let Some(capabilities) = capabilities.as_ref() {
println!(
" {} {}",
style.key("Rustup components:"),
style.value(&format_rustup_component_matrix(capabilities))
);
println!(
" {} {}",
style.key("Named tools:"),
style.value(&format_named_tool_matrix(capabilities))
);
if !capabilities.probe_warnings.is_empty() {
println!(
" {} {}",
style.key("Probe warnings:"),
style.warning(&capabilities.probe_warnings.join("; "))
);
}
}
}
}
Ok(false) => {
let code = rch_common::ErrorCode::WorkerHealthCheckFailed.code_string();
results.push(WorkerProbeResult {
id: worker.id.as_str().to_string(),
host: worker.host.clone(),
status: "unhealthy".to_string(),
latency_ms: None,
error: Some("Health check failed".to_string()),
error_code: Some(code),
capabilities: None,
missing_components: Vec::new(),
disk: disk_info_from_probe(None, daemon_worker),
});
if !ctx.is_json() {
println!(
"{} {}",
StatusIndicator::Error.with_label(style, "Health check failed"),
format_worker_disk(&disk_info_from_probe(None, daemon_worker))
);
}
}
Err(e) => {
let ssh_error = classify_ssh_error(worker, &e, ssh_options.connect_timeout);
let code = ssh_error_code(&ssh_error).code_string();
let report = format_ssh_report(ssh_error);
results.push(WorkerProbeResult {
id: worker.id.as_str().to_string(),
host: worker.host.clone(),
status: "error".to_string(),
latency_ms: None,
error: Some(report.clone()),
error_code: Some(code.clone()),
capabilities: None,
missing_components: Vec::new(),
disk: disk_info_from_probe(None, daemon_worker),
});
if !ctx.is_json() {
println!(
"{} Health check failed [{}] {}:\n{}",
StatusIndicator::Error.display(style),
style.highlight(&code),
format_worker_disk(&disk_info_from_probe(None, daemon_worker)),
indent_lines(&report, " ")
);
}
}
}
let _ = client.disconnect().await;
}
Err(e) => {
let ssh_error = classify_ssh_error(worker, &e, ssh_options.connect_timeout);
let code = ssh_error_code(&ssh_error).code_string();
let report = format_ssh_report(ssh_error);
results.push(WorkerProbeResult {
id: worker.id.as_str().to_string(),
host: worker.host.clone(),
status: "connection_failed".to_string(),
latency_ms: None,
error: Some(report.clone()),
error_code: Some(code.clone()),
capabilities: None,
missing_components: Vec::new(),
disk: disk_info_from_probe(None, daemon_worker),
});
if !ctx.is_json() {
println!(
"{} Connection failed [{}] {}:\n{}",
StatusIndicator::Error.display(style),
style.highlight(&code),
format_worker_disk(&disk_info_from_probe(None, daemon_worker)),
indent_lines(&report, " ")
);
}
}
}
}
let summary = summarize_probe_results(&results);
if ctx.is_json() {
let _ = ctx.json(&ApiResponse::ok(
"workers probe",
WorkersProbeResponse { results, summary },
));
} else {
println!("\n{}", format_probe_summary_line(&summary, style));
}
Ok(())
}
pub(crate) fn summarize_probe_results(results: &[WorkerProbeResult]) -> WorkerProbeSummary {
let mut summary = WorkerProbeSummary {
total: results.len(),
..WorkerProbeSummary::default()
};
for result in results {
match result.status.as_str() {
"ok" | "healthy" => summary.healthy += 1,
"unhealthy" => summary.unhealthy += 1,
"capability_missing" | "capability_unknown" => summary.capability_missing += 1,
_ => summary.failed += 1,
}
if let Some(code) = &result.error_code {
*summary.by_error_code.entry(code.clone()).or_insert(0) += 1;
} else if result.error.is_some() {
*summary
.by_error_code
.entry("other".to_string())
.or_insert(0) += 1;
}
}
summary
}
fn format_probe_summary_line(
summary: &WorkerProbeSummary,
style: &crate::ui::theme::Theme,
) -> String {
let mut parts: Vec<String> = Vec::new();
parts.push(format!("{} healthy", summary.healthy));
if summary.capability_missing > 0 {
parts.push(format!(
"{} capability missing/unknown",
summary.capability_missing
));
}
for (code, count) in &summary.by_error_code {
parts.push(format!("{} {}", count, code));
}
format!(
"{} worker(s) probed: {}.",
style.highlight(&summary.total.to_string()),
parts.join(", ")
)
}
#[cfg(not(unix))]
pub async fn workers_benchmark_filtered(
_worker_id: Option<&str>,
_force: bool,
_ctx: &OutputContext,
) -> Result<()> {
Err(PlatformError::UnixOnly {
feature: "worker benchmark".to_string(),
})?
}
#[cfg(unix)]
pub async fn workers_benchmark_filtered(
worker_id: Option<&str>,
force: bool,
ctx: &OutputContext,
) -> Result<()> {
let all_workers = load_workers_from_config()?;
let style = ctx.theme();
let workers: Vec<_> = if let Some(target) = worker_id {
let found: Vec<_> = all_workers
.iter()
.filter(|w| w.id.as_str() == target)
.cloned()
.collect();
if found.is_empty() {
let configured: Vec<String> = all_workers.iter().map(|w| w.id.to_string()).collect();
anyhow::bail!(
"unknown worker id: {target:?}; configured workers: {}",
if configured.is_empty() {
"(none)".to_string()
} else {
configured.join(", ")
}
);
}
found
} else {
all_workers
};
tracing::info!(
target: "rch::workers::benchmark",
worker_filter = ?worker_id,
worker_count = workers.len(),
force,
"workers.benchmark.start",
);
if force {
tracing::debug!(
target: "rch::workers::benchmark",
"workers.benchmark.force=true (no-op today; reserved for future recency gate)",
);
}
if workers.is_empty() {
if ctx.is_json() {
ctx.json(&ApiResponse::<Vec<WorkerBenchmarkResult>>::ok(
"workers benchmark",
vec![],
))?;
}
return Ok(());
}
let mut results = Vec::new();
let mp = MultiProgressManager::new(ctx);
if !ctx.is_json() && !mp.is_visible() {
println!(
"Running benchmarks on {} worker(s)...\n",
style.highlight(&workers.len().to_string())
);
}
for worker in &workers {
let spinner = if !ctx.is_json() {
let pb = mp.add_spinner(worker.id.as_str(), "Connecting...");
Some(pb)
} else {
None
};
let ssh_options = SshOptions::default();
let mut client = SshClient::new(worker.clone(), ssh_options.clone());
match client.connect().await {
Ok(()) => {
if let Some(ref pb) = spinner {
pb.set_message("Running benchmark...");
}
let bench_id = uuid::Uuid::new_v4();
let bench_dir = format!("rch_bench_{}", bench_id);
let benchmark_cmd = format!(
r###"#
cd /tmp && \
mkdir -p {0} && \
cd {0} && \
echo 'fn main() {{ println!("hello"); }}' > main.rs && \
(time rustc main.rs -o hello) 2>&1 | grep real || echo 'rustc not found'; \
cd .. && rm -rf {0}
"###,
bench_dir
);
let start = std::time::Instant::now();
let result = client.execute(&benchmark_cmd).await;
let duration = start.elapsed();
match result {
Ok(r) if r.success() => {
let duration_ms = u64::try_from(duration.as_millis()).unwrap_or(u64::MAX);
results.push(WorkerBenchmarkResult {
id: worker.id.as_str().to_string(),
host: worker.host.clone(),
status: "ok".to_string(),
duration_ms: Some(duration_ms),
error: None,
});
if let Some(ref pb) = spinner {
pb.finish_with_message(format!("✓ {}ms", duration_ms));
}
}
Ok(r) => {
results.push(WorkerBenchmarkResult {
id: worker.id.as_str().to_string(),
host: worker.host.clone(),
status: "failed".to_string(),
duration_ms: None,
error: Some(format!("exit code {}", r.exit_code)),
});
if let Some(ref pb) = spinner {
pb.finish_with_message(format!("✗ Failed (exit={})", r.exit_code));
}
}
Err(e) => {
let ssh_error = classify_ssh_error(worker, &e, ssh_options.command_timeout);
let report = format_ssh_report(ssh_error);
results.push(WorkerBenchmarkResult {
id: worker.id.as_str().to_string(),
host: worker.host.clone(),
status: "error".to_string(),
duration_ms: None,
error: Some(report.clone()),
});
if let Some(ref pb) = spinner {
pb.finish_with_message(format!(
"✗ {}",
report.lines().next().unwrap_or("Error")
));
}
}
}
let _ = client.disconnect().await;
}
Err(e) => {
let ssh_error = classify_ssh_error(worker, &e, ssh_options.connect_timeout);
let report = format_ssh_report(ssh_error);
results.push(WorkerBenchmarkResult {
id: worker.id.as_str().to_string(),
host: worker.host.clone(),
status: "connection_failed".to_string(),
duration_ms: None,
error: Some(report.clone()),
});
if let Some(ref pb) = spinner {
pb.finish_with_message(format!(
"✗ Connection failed: {}",
report.lines().next().unwrap_or("Error")
));
}
}
}
}
if ctx.is_json() {
ctx.json(&ApiResponse::ok("workers benchmark", results))?;
} else {
println!(
"\n{} For accurate speed scores, use longer benchmark runs.",
StatusIndicator::Info.display(style)
);
}
Ok(())
}
pub async fn workers_compare(worker_ids: &[String], ctx: &OutputContext) -> Result<()> {
if worker_ids.len() < 2 {
anyhow::bail!(
"workers compare requires at least 2 worker IDs; got {}",
worker_ids.len()
);
}
let all_workers = load_workers_from_config()?;
let known: std::collections::BTreeSet<&str> =
all_workers.iter().map(|w| w.id.as_str()).collect();
let unknown: Vec<&str> = worker_ids
.iter()
.map(String::as_str)
.filter(|id| !known.contains(id))
.collect();
if !unknown.is_empty() {
let configured: Vec<String> = all_workers.iter().map(|w| w.id.to_string()).collect();
anyhow::bail!(
"unknown worker id(s): {}; configured workers: {}",
unknown.join(", "),
if configured.is_empty() {
"(none)".to_string()
} else {
configured.join(", ")
}
);
}
let mut entries: Vec<(String, Option<SpeedScoreViewFromApi>)> =
Vec::with_capacity(worker_ids.len());
for id in worker_ids {
let payload = format!("GET /speedscore/{}\n", id);
let response = send_daemon_command(&payload).await?;
let json = extract_json_body(&response).ok_or_else(|| {
anyhow::anyhow!("invalid response format from daemon for /speedscore/{id}")
})?;
let parsed: SpeedScoreResponseFromApi = serde_json::from_str(json)
.with_context(|| format!("failed to parse /speedscore/{id} response from daemon"))?;
entries.push((id.clone(), parsed.speedscore));
}
tracing::info!(
target: "rch::workers::compare",
worker_count = entries.len(),
recorded_count = entries.iter().filter(|(_, s)| s.is_some()).count(),
"workers.compare.fetched",
);
fn finite_metric(
score: &SpeedScoreViewFromApi,
pick: fn(&SpeedScoreViewFromApi) -> f64,
) -> Option<f64> {
let value = pick(score);
value.is_finite().then_some(value)
}
fn leader_for<F>(entries: &[(String, Option<SpeedScoreViewFromApi>)], pick: F) -> Option<usize>
where
F: Fn(&SpeedScoreViewFromApi) -> f64,
{
let mut best: Option<(usize, f64)> = None;
for (i, (_, sc)) in entries.iter().enumerate() {
if let Some(s) = sc {
let v = pick(s);
if !v.is_finite() {
continue;
}
match best {
None => best = Some((i, v)),
Some((_, bv)) if v > bv => best = Some((i, v)),
_ => {}
}
}
}
best.map(|(i, _)| i)
}
if ctx.is_json() {
let row = |label: &'static str, pick: fn(&SpeedScoreViewFromApi) -> f64| {
serde_json::json!({
"metric": label,
"leader_index": leader_for(&entries, pick),
"values": entries
.iter()
.map(|(_, s)| s.as_ref().and_then(|score| finite_metric(score, pick)))
.collect::<Vec<_>>(),
})
};
ctx.json(&ApiResponse::ok(
"workers compare",
serde_json::json!({
"workers": entries.iter().map(|(id, _)| id).collect::<Vec<_>>(),
"rows": [
row("speedscore", |s| s.total),
row("cpu", |s| s.cpu_score),
row("memory", |s| s.memory_score),
row("disk", |s| s.disk_score),
row("network", |s| s.network_score),
row("compile", |s| s.compilation_score),
],
"recommendation_leader_index": leader_for(&entries, |s| s.total),
}),
))?;
return Ok(());
}
let style = ctx.theme();
let col_width = entries
.iter()
.map(|(id, _)| id.len())
.max()
.unwrap_or(8)
.max(8);
print!(" {:<12}", "Metric");
for (id, _) in &entries {
print!(" {:>width$}", id, width = col_width);
}
println!();
print!(" {:-<12}", "");
for _ in &entries {
print!(" {:->width$}", "", width = col_width);
}
println!();
let row = |label: &str, pick: fn(&SpeedScoreViewFromApi) -> f64| {
let leader = leader_for(&entries, pick);
print!(" {:<12}", label);
for (i, (_, sc)) in entries.iter().enumerate() {
let cell = match sc {
Some(s) => match finite_metric(s, pick) {
Some(val) => {
if Some(i) == leader {
format!("*{val:>w$.1}", w = col_width - 1)
} else {
format!("{val:>w$.1}", w = col_width)
}
}
None => format!("{:>w$}", "-", w = col_width),
},
None => format!("{:>w$}", "-", w = col_width),
};
print!(" {cell}");
}
println!();
};
row("SpeedScore", |s| s.total);
row("CPU", |s| s.cpu_score);
row("Memory", |s| s.memory_score);
row("Disk", |s| s.disk_score);
row("Network", |s| s.network_score);
row("Compile", |s| s.compilation_score);
println!();
match leader_for(&entries, |s| s.total) {
Some(idx) => {
if let Some((id, Some(score))) = entries.get(idx) {
println!(
"{} {} leads with SpeedScore {:.1}",
style.format_success("Recommendation:"),
style.highlight(id),
score.total
);
} else {
println!(
"{}",
style.format_warning(
"No recorded SpeedScore for any named worker; \
run `rch workers benchmark` first."
)
);
}
}
None => {
println!(
"{}",
style.format_warning(
"No recorded SpeedScore for any named worker; \
run `rch workers benchmark` first."
)
);
}
}
Ok(())
}
fn confirm_or_explain_non_interactive(prompt: &str) -> Result<bool> {
use std::io::IsTerminal;
if !std::io::stdin().is_terminal() {
anyhow::bail!(
"'{prompt}' needs an interactive confirmation but stdin is not a terminal; \
re-run with -y/--yes to confirm non-interactively (or --json for machine output)"
);
}
Ok(dialoguer::Confirm::new()
.with_prompt(prompt)
.default(false)
.interact()?)
}
pub async fn workers_drain(worker_id: &str, skip_confirm: bool, ctx: &OutputContext) -> Result<()> {
let style = ctx.theme();
let socket_path_str = configured_socket_path()?;
if !Path::new(&socket_path_str).exists() {
if ctx.is_json() {
let _ = ctx.json(&ApiResponse::<()>::err(
"workers drain",
ApiError::new(ErrorCode::InternalDaemonNotRunning, "Daemon is not running"),
));
} else {
println!(
"{} Daemon is not running. Start it with {}",
StatusIndicator::Error.display(style),
style.highlight("rch daemon start")
);
println!(
"\n{} Draining requires the daemon to track worker state.",
StatusIndicator::Info.display(style)
);
}
return Ok(());
}
if !skip_confirm && !ctx.is_json() {
println!(
"{} This will stop routing new jobs to worker {}.",
StatusIndicator::Warning.display(style),
style.highlight(worker_id)
);
println!(
" {} Active builds will be allowed to complete.",
StatusIndicator::Info.display(style)
);
let confirmed = confirm_or_explain_non_interactive("Drain this worker?")?;
if !confirmed {
println!("{} Aborted.", StatusIndicator::Info.display(style));
return Ok(());
}
}
match send_daemon_command(&format!(
"POST /workers/{}/drain\n",
urlencoding_encode(worker_id)
))
.await
{
Ok(response) => {
if response.contains("error") || response.contains("Error") {
if ctx.is_json() {
let _ = ctx.json(&ApiResponse::ok(
"workers drain",
WorkerActionResponse {
worker_id: worker_id.to_string(),
action: "drain".to_string(),
success: false,
message: Some(response),
},
));
} else {
println!(
"{} Failed to drain worker: {}",
StatusIndicator::Error.display(style),
style.muted(&response)
);
}
} else if ctx.is_json() {
let _ = ctx.json(&ApiResponse::ok(
"workers drain",
WorkerActionResponse {
worker_id: worker_id.to_string(),
action: "drain".to_string(),
success: true,
message: Some("Worker is now draining".to_string()),
},
));
} else {
println!(
"{} Worker {} is now draining.",
StatusIndicator::Success.display(style),
style.highlight(worker_id)
);
println!(
" {} No new jobs will be sent. Existing jobs will complete.",
StatusIndicator::Info.display(style)
);
}
}
Err(e) => {
if ctx.is_json() {
let _ = ctx.json(&ApiResponse::<()>::err(
"workers drain",
ApiError::new(ErrorCode::InternalStateError, e.to_string()),
));
} else {
println!(
"{} Failed to communicate with daemon: {}",
StatusIndicator::Error.display(style),
style.muted(&e.to_string())
);
println!(
"\n{} Drain/enable commands require the daemon to be running.",
StatusIndicator::Info.display(style)
);
}
}
}
Ok(())
}
pub async fn workers_enable(worker_id: &str, ctx: &OutputContext) -> Result<()> {
let style = ctx.theme();
let socket_path_str = configured_socket_path()?;
if !Path::new(&socket_path_str).exists() {
if ctx.is_json() {
let _ = ctx.json(&ApiResponse::<()>::err(
"workers enable",
ApiError::new(ErrorCode::InternalDaemonNotRunning, "Daemon is not running"),
));
} else {
println!(
"{} Daemon is not running. Start it with {}",
StatusIndicator::Error.display(style),
style.highlight("rch daemon start")
);
}
return Ok(());
}
match send_daemon_command(&format!(
"POST /workers/{}/enable\n",
urlencoding_encode(worker_id)
))
.await
{
Ok(response) => {
if response.contains("error") || response.contains("Error") {
if ctx.is_json() {
let _ = ctx.json(&ApiResponse::ok(
"workers enable",
WorkerActionResponse {
worker_id: worker_id.to_string(),
action: "enable".to_string(),
success: false,
message: Some(response),
},
));
} else {
println!(
"{} Failed to enable worker: {}",
StatusIndicator::Error.display(style),
style.muted(&response)
);
}
} else if ctx.is_json() {
let _ = ctx.json(&ApiResponse::ok(
"workers enable",
WorkerActionResponse {
worker_id: worker_id.to_string(),
action: "enable".to_string(),
success: true,
message: Some("Worker is now enabled".to_string()),
},
));
} else {
println!(
"{} Worker {} is now enabled.",
StatusIndicator::Success.display(style),
style.highlight(worker_id)
);
}
}
Err(e) => {
if ctx.is_json() {
let _ = ctx.json(&ApiResponse::<()>::err(
"workers enable",
ApiError::new(ErrorCode::InternalStateError, e.to_string()),
));
} else {
println!(
"{} Failed to communicate with daemon: {}",
StatusIndicator::Error.display(style),
style.muted(&e.to_string())
);
}
}
}
Ok(())
}
pub async fn workers_disable(
worker_id: &str,
reason: Option<String>,
drain_first: bool,
skip_confirm: bool,
ctx: &OutputContext,
) -> Result<()> {
let style = ctx.theme();
let socket_path_str = configured_socket_path()?;
if !Path::new(&socket_path_str).exists() {
if ctx.is_json() {
let _ = ctx.json(&ApiResponse::<()>::err(
"workers disable",
ApiError::new(ErrorCode::InternalDaemonNotRunning, "Daemon is not running"),
));
} else {
println!(
"{} Daemon is not running. Start it with {}",
StatusIndicator::Error.display(style),
style.highlight("rch daemon start")
);
}
return Ok(());
}
if !skip_confirm && !ctx.is_json() {
println!(
"{} This will mark worker {} as offline.",
StatusIndicator::Warning.display(style),
style.highlight(worker_id)
);
if drain_first {
println!(
" {} Active builds will complete before disabling.",
StatusIndicator::Info.display(style)
);
} else {
println!(
" {} The worker will be immediately excluded from job assignment.",
StatusIndicator::Info.display(style)
);
}
let confirmed = confirm_or_explain_non_interactive("Disable this worker?")?;
if !confirmed {
println!("{} Aborted.", StatusIndicator::Info.display(style));
return Ok(());
}
}
let mut url = format!("POST /workers/{}/disable", urlencoding_encode(worker_id));
let mut query_parts = Vec::new();
if let Some(ref r) = reason {
query_parts.push(format!("reason={}", urlencoding_encode(r)));
}
if drain_first {
query_parts.push("drain=true".to_string());
}
if !query_parts.is_empty() {
url = format!("{}?{}", url, query_parts.join("&"));
}
url.push('\n');
match send_daemon_command(&url).await {
Ok(response) => {
if response.contains("error") || response.contains("Error") {
if ctx.is_json() {
let _ = ctx.json(&ApiResponse::ok(
"workers disable",
WorkerActionResponse {
worker_id: worker_id.to_string(),
action: "disable".to_string(),
success: false,
message: Some(response),
},
));
} else {
println!(
"{} Failed to disable worker: {}",
StatusIndicator::Error.display(style),
style.muted(&response)
);
}
} else if ctx.is_json() {
let _ = ctx.json(&ApiResponse::ok(
"workers disable",
WorkerActionResponse {
worker_id: worker_id.to_string(),
action: "disable".to_string(),
success: true,
message: Some(if drain_first {
"Worker is draining before disable".to_string()
} else {
"Worker is now disabled".to_string()
}),
},
));
} else if drain_first {
println!(
"{} Worker {} is draining before disable.",
StatusIndicator::Success.display(style),
style.highlight(worker_id)
);
println!(
" {} Existing jobs will complete, then worker will be disabled.",
StatusIndicator::Info.display(style)
);
if let Some(ref r) = reason {
println!(
" {} Reason: {}",
StatusIndicator::Info.display(style),
style.muted(r)
);
}
} else {
println!(
"{} Worker {} is now disabled.",
StatusIndicator::Success.display(style),
style.highlight(worker_id)
);
println!(
" {} No jobs will be sent to this worker.",
StatusIndicator::Info.display(style)
);
if let Some(ref r) = reason {
println!(
" {} Reason: {}",
StatusIndicator::Info.display(style),
style.muted(r)
);
}
}
}
Err(e) => {
if ctx.is_json() {
let _ = ctx.json(&ApiResponse::<()>::err(
"workers disable",
ApiError::new(ErrorCode::InternalStateError, e.to_string()),
));
} else {
println!(
"{} Failed to communicate with daemon: {}",
StatusIndicator::Error.display(style),
style.muted(&e.to_string())
);
}
}
}
Ok(())
}
#[cfg(test)]
mod probe_summary_tests {
use super::*;
use crate::ui::context::OutputConfig;
use crate::ui::writer::SharedOutputBuffer;
use rch_common::WorkerId;
fn mk(status: &str, error_code: Option<&str>, error: Option<&str>) -> WorkerProbeResult {
WorkerProbeResult {
id: "w".to_string(),
host: "h".to_string(),
status: status.to_string(),
latency_ms: None,
error: error.map(String::from),
error_code: error_code.map(String::from),
capabilities: None,
missing_components: Vec::new(),
disk: WorkerDiskInfo::default(),
}
}
#[test]
fn summarize_counts_healthy_and_error_codes() {
let results = vec![
mk("ok", None, None),
mk("ok", None, None),
mk("connection_failed", Some("RCH-E100"), Some("...")),
mk("connection_failed", Some("RCH-E108"), Some("...")),
mk("connection_failed", Some("RCH-E108"), Some("...")),
mk("unhealthy", Some("RCH-E202"), Some("Health check failed")),
];
let s = summarize_probe_results(&results);
assert_eq!(s.total, 6);
assert_eq!(s.healthy, 2);
assert_eq!(s.unhealthy, 1);
assert_eq!(s.failed, 3);
assert_eq!(s.capability_missing, 0);
assert_eq!(s.by_error_code.get("RCH-E100"), Some(&1));
assert_eq!(s.by_error_code.get("RCH-E108"), Some(&2));
assert_eq!(s.by_error_code.get("RCH-E202"), Some(&1));
}
#[test]
fn summarize_buckets_uncategorized_errors_as_other() {
let results = vec![mk("error", None, Some("something weird"))];
let s = summarize_probe_results(&results);
assert_eq!(s.by_error_code.get("other"), Some(&1));
}
#[test]
fn summarize_separates_component_capability_gaps_from_connectivity_failures() {
let results = vec![mk(
"capability_missing",
Some("RCH-E205"),
Some("missing clippy"),
)];
let summary = summarize_probe_results(&results);
assert_eq!(summary.capability_missing, 1);
assert_eq!(summary.failed, 0);
assert_eq!(summary.by_error_code.get("RCH-E205"), Some(&1));
}
#[test]
fn declared_components_use_exact_toolchain_scoped_inventory() {
let capabilities = WorkerCapabilities {
rustup_components: vec![
"stable-x86_64-unknown-linux-gnu:clippy".to_string(),
"nightly-2026-07-05-x86_64-unknown-linux-gnu:rustfmt".to_string(),
],
..WorkerCapabilities::default()
};
assert_eq!(
missing_declared_components(
&capabilities,
Some("nightly-2026-07-05"),
&["clippy".to_string(), "rustfmt".to_string()],
),
vec!["clippy"]
);
assert_eq!(
format_rustup_component_matrix(&capabilities),
"stable-x86_64-unknown-linux-gnu:clippy, nightly-2026-07-05-x86_64-unknown-linux-gnu:rustfmt"
);
}
#[test]
fn summarize_empty_input() {
let s = summarize_probe_results(&[]);
assert_eq!(s.total, 0);
assert_eq!(s.healthy, 0);
assert!(s.by_error_code.is_empty());
}
fn make_worker() -> WorkerConfig {
WorkerConfig {
id: WorkerId::new("builder-1"),
host: "127.0.0.1".to_string(),
user: "ubuntu".to_string(),
identity_file: "~/.ssh/rch_test".to_string(),
total_slots: 8,
priority: 100,
tags: vec!["rust".to_string()],
tools: Vec::new(),
}
}
fn make_daemon_status() -> DaemonFullStatusResponse {
serde_json::from_value(serde_json::json!({
"daemon": {
"pid": 123,
"uptime_secs": 45,
"version": "test",
"socket_path": "/tmp/rch.sock",
"started_at": "2026-01-01T00:00:00Z",
"workers_total": 1,
"workers_healthy": 1,
"slots_total": 8,
"slots_available": 6
},
"workers": [{
"id": "builder-1",
"host": "127.0.0.1",
"user": "ubuntu",
"status": "degraded",
"circuit_state": "open",
"used_slots": 2,
"total_slots": 8,
"speed_score": 90.0,
"last_error": "ssh timeout",
"recovery_in_secs": 30
}],
"active_builds": [],
"recent_builds": [],
"issues": [],
"alerts": [],
"stats": {
"total_builds": 0,
"success_count": 0,
"failure_count": 0,
"remote_count": 0,
"local_count": 0,
"avg_duration_ms": 0
}
}))
.unwrap()
}
fn make_context(config: OutputConfig) -> OutputContext {
let stdout = SharedOutputBuffer::new().as_writer(true);
let stderr = SharedOutputBuffer::new().as_writer(true);
OutputContext::with_writers(config, stdout, stderr)
}
#[test]
fn disk_visibility_preserves_zero_and_marks_daemon_pressure() {
let mut status = make_daemon_status();
let worker = &mut status.workers[0];
worker.pressure_disk_free_gb = Some(0.0);
worker.pressure_disk_free_ratio = Some(0.0);
worker.pressure_state = Some("critical".to_string());
worker.pressure_reason_code = Some("disk_free_critical".to_string());
let disk = disk_info_from_daemon(Some(worker));
assert_eq!(
format_worker_disk(&disk),
"Disk: 0.0 GiB free (0.0% free, daemon); pressure: CRITICAL (cached daemon)"
);
assert_eq!(
disk.disk_pressure_reason.as_deref(),
Some("disk_free_critical")
);
worker.pressure_disk_free_gb = Some(25.0);
worker.pressure_disk_free_ratio = Some(0.125);
worker.pressure_state = Some("warning".to_string());
let disk = disk_info_from_daemon(Some(worker));
assert!(format_worker_disk(&disk).contains("25.0 GiB free (12.5% free"));
assert!(format_worker_disk(&disk).contains("pressure: WARNING"));
worker.pressure_state = Some("telemetry_gap".to_string());
worker.pressure_disk_free_gb = None;
worker.pressure_disk_free_ratio = None;
let disk = disk_info_from_daemon(Some(worker));
assert!(format_worker_disk(&disk).starts_with("Disk: unknown"));
assert!(format_worker_disk(&disk).contains("pressure: unknown"));
}
#[test]
fn disk_visibility_shows_the_build_disk_when_tmpfs_is_the_tightest_mount() {
let mut status = make_daemon_status();
let worker = &mut status.workers[0];
worker.pressure_disk_free_gb = Some(13.9);
worker.pressure_disk_free_ratio = Some(0.895);
worker.pressure_build_disk_free_gb = Some(1700.0);
worker.pressure_state = Some("healthy".to_string());
let disk = disk_info_from_daemon(Some(worker));
assert_eq!(disk.build_disk_free_gb, Some(1700.0));
assert_eq!(
format_worker_disk(&disk),
"Disk: 13.9 GiB free (89.5% free, daemon); build disk: 1700.0 GiB free; \
pressure: healthy (cached daemon)"
);
worker.pressure_build_disk_free_gb = Some(13.9);
let disk = disk_info_from_daemon(Some(worker));
assert!(!format_worker_disk(&disk).contains("build disk"));
let capabilities = WorkerCapabilities {
disk_free_gb: Some(13.9),
disk_total_gb: Some(15.5),
build_disk_free_gb: Some(1700.0),
build_disk_total_gb: Some(1900.0),
..Default::default()
};
let disk = disk_info_from_probe(Some(&capabilities), None);
assert_eq!(disk.build_disk_free_gb, Some(1700.0));
assert!(format_worker_disk(&disk).contains("; build disk: 1700.0 GiB free;"));
}
#[test]
fn disk_visibility_probe_uses_fresh_measurements_and_cached_pressure() {
let mut status = make_daemon_status();
status.workers[0].pressure_disk_free_gb = Some(5.0);
status.workers[0].pressure_disk_free_ratio = Some(0.025);
status.workers[0].pressure_state = Some("critical".to_string());
let capabilities = WorkerCapabilities {
disk_free_gb: Some(25.0),
disk_total_gb: Some(200.0),
..Default::default()
};
let disk = disk_info_from_probe(Some(&capabilities), Some(&status.workers[0]));
assert_eq!(
format_worker_disk(&disk),
"Disk: 25.0 GiB free (12.5% free, probe); pressure: CRITICAL (cached daemon)"
);
let failed = disk_info_from_probe(None, Some(&status.workers[0]));
assert!(failed.disk_free_gb.is_none());
assert!(failed.disk_free_ratio.is_none());
assert!(failed.disk_measurement_source.is_none());
assert_eq!(failed.disk_pressure_state.as_deref(), Some("critical"));
let without_daemon = disk_info_from_probe(Some(&capabilities), None);
assert!(format_worker_disk(&without_daemon).contains("25.0 GiB free"));
assert!(format_worker_disk(&without_daemon).ends_with("pressure: unknown"));
assert!(format_worker_disk(&disk_info_from_daemon(None)).starts_with("Disk: unknown"));
}
#[test]
fn disk_visibility_rejects_invalid_measurements_without_inventing_zero() {
let mut capabilities = WorkerCapabilities {
disk_free_gb: Some(25.0),
..Default::default()
};
for total in [
None,
Some(0.0),
Some(-1.0),
Some(f64::NAN),
Some(f64::INFINITY),
Some(10.0),
] {
capabilities.disk_total_gb = total;
let disk = disk_info_from_probe(Some(&capabilities), None);
assert_eq!(disk.disk_free_gb, Some(25.0));
assert!(disk.disk_free_ratio.is_none(), "invalid total {total:?}");
}
capabilities.disk_total_gb = Some(200.0);
for free in [None, Some(-1.0), Some(f64::NAN), Some(f64::INFINITY)] {
capabilities.disk_free_gb = free;
let disk = disk_info_from_probe(Some(&capabilities), None);
assert!(disk.disk_free_gb.is_none());
assert!(disk.disk_free_ratio.is_none());
}
capabilities.disk_free_gb = Some(0.0);
let disk = disk_info_from_probe(Some(&capabilities), None);
assert_eq!(disk.disk_free_gb, Some(0.0));
assert_eq!(disk.disk_free_ratio, Some(0.0));
let mut status = make_daemon_status();
status.workers[0].pressure_disk_free_gb = Some(f64::NAN);
for ratio in [f64::NAN, f64::INFINITY, -0.1, 1.1] {
status.workers[0].pressure_disk_free_ratio = Some(ratio);
let disk = disk_info_from_daemon(Some(&status.workers[0]));
assert!(disk.disk_free_gb.is_none());
assert!(disk.disk_free_ratio.is_none());
}
}
#[test]
fn disk_visibility_list_and_probe_json_toon_preserve_numbers_and_nulls() {
use crate::ui::context::OutputFormat;
let mut status = make_daemon_status();
status.workers[0].pressure_state = Some("warning".to_string());
for free in [Some(25.0), Some(0.0), None] {
let known = free.is_some();
status.workers[0].pressure_disk_free_gb = free;
status.workers[0].pressure_disk_free_ratio = free.map(|free| free / 200.0);
let disk = disk_info_from_daemon(known.then_some(&status.workers[0]));
let mut info = WorkerInfo::from(&make_worker());
info.disk = disk;
let mut probe = mk("ok", None, None);
probe.latency_ms = Some(42);
let capabilities = WorkerCapabilities {
disk_free_gb: free,
disk_total_gb: Some(200.0),
..Default::default()
};
probe.disk =
disk_info_from_probe(Some(&capabilities), known.then_some(&status.workers[0]));
let payloads = [
serde_json::to_value(WorkersListResponse {
workers: vec![info],
count: 1,
})
.unwrap(),
serde_json::to_value(WorkersProbeResponse {
summary: summarize_probe_results(std::slice::from_ref(&probe)),
results: vec![probe],
})
.unwrap(),
];
for (payload, collection) in payloads.iter().zip(["workers", "results"]) {
for format in [OutputFormat::Json, OutputFormat::Toon] {
let stdout = SharedOutputBuffer::new();
let stderr = SharedOutputBuffer::new();
let ctx = OutputContext::with_writers(
OutputConfig {
json: true,
format,
..Default::default()
},
stdout.as_writer(true),
stderr.as_writer(true),
);
ctx.json(&ApiResponse::ok("workers", payload)).unwrap();
let output = stdout.to_string_lossy();
let json = match format {
OutputFormat::Json => output,
OutputFormat::Toon => toon_rust::toon_to_json(&output).unwrap(),
};
let decoded: serde_json::Value = serde_json::from_str(&json).unwrap();
let worker = &decoded["data"][collection][0];
if known {
assert_eq!(worker["disk_free_gb"].as_f64(), free);
assert_eq!(
worker["disk_free_ratio"].as_f64(),
free.map(|free| free / 200.0)
);
assert_eq!(worker["disk_pressure_state"], "warning");
assert_eq!(worker["disk_pressure_source"], "daemon");
assert_eq!(
worker["disk_measurement_source"],
if collection == "workers" {
"daemon"
} else {
"probe"
}
);
} else {
for key in [
"disk_free_gb",
"disk_free_ratio",
"disk_pressure_state",
"disk_measurement_source",
"disk_pressure_source",
"disk_pressure_reason",
] {
assert_eq!(worker.get(key), Some(&serde_json::Value::Null));
}
}
if collection == "results" {
assert_eq!(worker["latency_ms"].as_f64(), Some(42.0));
if format == OutputFormat::Json {
assert_eq!(worker["latency_ms"].as_u64(), Some(42));
}
}
assert!(stderr.to_string_lossy().is_empty());
}
}
}
}
#[test]
fn workers_list_uses_live_capacity_including_zero_for_all_formats() {
let mut status = make_daemon_status();
for effective in [3, 0] {
status.workers[0].total_slots = effective;
let mut workers = vec![make_worker()];
apply_live_worker_slots(&mut workers, Some(&status));
assert_eq!(workers[0].total_slots, effective);
let json = serde_json::to_value(WorkerInfo::from(&workers[0])).unwrap();
assert_eq!(json["total_slots"], effective);
assert_eq!(workers[0].priority, 100);
assert_eq!(workers[0].host, "127.0.0.1");
}
}
#[test]
fn workers_list_retains_configured_capacity_without_matching_live_status() {
let mut workers = vec![make_worker()];
apply_live_worker_slots(&mut workers, None);
assert_eq!(workers[0].total_slots, 8);
let mut status = make_daemon_status();
status.workers[0].id = "different-worker".to_string();
status.workers[0].total_slots = 0;
apply_live_worker_slots(&mut workers, Some(&status));
assert_eq!(workers[0].total_slots, 8);
}
#[test]
fn test_workers_list_verbose_shows_extra_columns() {
let worker = make_worker();
let status = make_daemon_status();
let style = crate::ui::theme::Style::new(false, true, false);
let output = render_worker_verbose_lines(&worker, Some(&status), &style).join("\n");
assert!(output.contains("Circuit"));
assert!(output.contains("In Use"));
assert!(output.contains("Status"));
assert!(output.contains("LastErr"));
assert!(output.contains("Recover"));
assert!(output.contains("SSH Key"));
assert!(output.contains("ssh timeout"));
assert!(output.contains("~/.ssh/rch_test"));
}
#[test]
fn test_workers_list_normal_hides_extra_columns() {
let normal_ctx = make_context(OutputConfig::default());
assert!(!workers_list_verbose_enabled(&normal_ctx));
let verbose_ctx = make_context(OutputConfig {
verbose: true,
..Default::default()
});
assert!(workers_list_verbose_enabled(&verbose_ctx));
let verbose_json_ctx = make_context(OutputConfig {
json: true,
verbose: true,
..Default::default()
});
assert!(!workers_list_verbose_enabled(&verbose_json_ctx));
}
}