mod load;
use crate::hook::{
cargo_job_count_for_command, estimate_cores_for_command, extract_project_name_with_policy,
preferred_workers, query_daemon_dry_run, required_runtime_for_kind,
};
use crate::status_types::{
DaemonFullStatusResponse, IssueFromApi, SelfTestHistoryResponseFromApi,
SelfTestResultRecordFromApi, SelfTestRunResponseFromApi, SelfTestStatusResponseFromApi,
WorkerStatusFromApi, extract_json_body,
};
use crate::toolchain::detect_toolchain;
use crate::ui::context::OutputContext;
use crate::ui::progress::Spinner;
use crate::ui::theme::{StatusIndicator, Theme};
use anyhow::{Context, Result};
use rch_common::fleet_smoke_profile::{
ProfileMode, ScenarioAction, SmokeProfileEvent, SmokeProfileInputs, SmokeProfilePlan,
SmokeScenario, plan_smoke_profile,
};
use rch_common::{
ApiResponse, PlacementPlan, RequestedWorkerFacts, RequestedWorkerOutcome, RequiredRuntime,
WorkerConfig, evaluate_requested_worker, normalize_project_path_with_policy, resolve_placement,
};
use std::path::Path;
use tracing::debug;
use self::load::run_load_storm;
use super::config::collect_value_sources;
use super::helpers::{humanize_duration, runtime_label, urlencoding_encode};
use super::types::{
DiagnoseDaemonStatus, DiagnoseDecision, DiagnoseResponse, DiagnoseThreshold,
DiagnoseWorkerSelection, DryRunPipelineStep, DryRunSummary,
};
use super::workers::{
collect_local_capability_warnings, has_any_capabilities, probe_local_capabilities,
query_workers_capabilities,
};
use super::{load_workers_from_config, query_daemon_health, send_daemon_command};
pub(super) fn build_diagnose_decision(
classification: &rch_common::Classification,
threshold: f64,
) -> DiagnoseDecision {
let would_intercept = classification.is_compilation && classification.confidence >= threshold;
let reason = if !classification.is_compilation {
"Command not classified as compilation".to_string()
} else if classification.confidence < threshold {
format!(
"Confidence {:.2} below threshold {:.2}",
classification.confidence, threshold
)
} else {
format!(
"Compilation command with confidence {:.2} >= threshold {:.2}",
classification.confidence, threshold
)
};
DiagnoseDecision {
would_intercept,
reason,
}
}
pub(super) fn build_diagnose_decision_with_config(
classification: &rch_common::Classification,
config: &rch_common::RchConfig,
) -> DiagnoseDecision {
if classification.is_compilation
&& let Some(policy) = crate::hook::config_local_policy(config, classification.kind)
{
return DiagnoseDecision {
would_intercept: false,
reason: format!("Would run locally: {}", policy.reason()),
};
}
let threshold = if config.general.force_remote {
0.0
} else {
config.compilation.confidence_threshold
};
build_diagnose_decision(classification, threshold)
}
pub(super) fn build_diagnose_slot_estimate(
kind: Option<rch_common::CompilationKind>,
command: &str,
config: &rch_common::CompilationConfig,
) -> (u32, Option<u32>) {
(
estimate_cores_for_command(kind, command, config),
cargo_job_count_for_command(command),
)
}
pub(super) fn build_dry_run_summary(
would_intercept: bool,
reason: &str,
worker_selection: &Option<DiagnoseWorkerSelection>,
daemon_reachable: bool,
) -> DryRunSummary {
if !would_intercept {
return DryRunSummary {
would_offload: false,
reason: reason.to_string(),
pipeline_steps: vec![DryRunPipelineStep {
step: 1,
name: "Local execution".to_string(),
description: "Command would run locally (not offloaded)".to_string(),
skipped: false,
skip_reason: None,
estimated_duration_ms: None,
}],
transfer_estimate: None,
total_estimated_ms: None,
};
}
let mut steps = Vec::new();
steps.push(DryRunPipelineStep {
step: 1,
name: "Classification".to_string(),
description: "Command classified as compilation".to_string(),
skipped: false,
skip_reason: None,
estimated_duration_ms: Some(1),
});
steps.push(DryRunPipelineStep {
step: 2,
name: "Daemon query".to_string(),
description: "Request worker selection from daemon".to_string(),
skipped: !daemon_reachable,
skip_reason: if !daemon_reachable {
Some("Daemon not reachable".to_string())
} else {
None
},
estimated_duration_ms: Some(5),
});
let worker_selected = worker_selection
.as_ref()
.is_some_and(|s| s.worker.is_some());
let no_worker_reason = if daemon_reachable {
worker_selection
.as_ref()
.map(|s| s.reason.to_string())
.unwrap_or_else(|| "No worker available".to_string())
} else {
"Daemon not reachable".to_string()
};
steps.push(DryRunPipelineStep {
step: 3,
name: "Worker selection".to_string(),
description: if worker_selected {
format!(
"Worker {} selected",
worker_selection
.as_ref()
.and_then(|s| s.worker.as_ref())
.map(|w| w.id.as_str())
.unwrap_or("unknown")
)
} else {
"Select best available worker".to_string()
},
skipped: !worker_selected,
skip_reason: if !worker_selected {
Some(no_worker_reason.clone())
} else {
None
},
estimated_duration_ms: Some(2),
});
steps.push(DryRunPipelineStep {
step: 4,
name: "Transfer".to_string(),
description: "Sync project files to worker via rsync+zstd".to_string(),
skipped: !worker_selected,
skip_reason: if !worker_selected {
Some(no_worker_reason.clone())
} else {
None
},
estimated_duration_ms: None, });
steps.push(DryRunPipelineStep {
step: 5,
name: "Remote execution".to_string(),
description: "Execute compilation command on worker".to_string(),
skipped: !worker_selected,
skip_reason: if !worker_selected {
Some(no_worker_reason.clone())
} else {
None
},
estimated_duration_ms: None, });
steps.push(DryRunPipelineStep {
step: 6,
name: "Artifact retrieval".to_string(),
description: "Retrieve build artifacts from remote worker".to_string(),
skipped: !worker_selected,
skip_reason: if !worker_selected {
Some(no_worker_reason.clone())
} else {
None
},
estimated_duration_ms: None, });
let reason = if worker_selected {
"compilation command meets threshold, worker available".to_string()
} else if !daemon_reachable {
"compilation command meets threshold, but daemon is not reachable".to_string()
} else {
format!("compilation command meets threshold, but no worker selected: {no_worker_reason}")
};
DryRunSummary {
would_offload: worker_selected,
reason,
pipeline_steps: steps,
transfer_estimate: None, total_estimated_ms: None, }
}
fn summarize_capabilities(caps: &rch_common::WorkerCapabilities) -> String {
let mut parts = Vec::new();
if caps.has_rust() {
parts.push(format!(
"rust {}",
caps.rustc_version.as_deref().unwrap_or("?")
));
}
if caps.has_bun() {
parts.push(format!(
"bun {}",
caps.bun_version.as_deref().unwrap_or("?")
));
}
if caps.has_node() {
parts.push(format!(
"node {}",
caps.node_version.as_deref().unwrap_or("?")
));
}
if parts.is_empty() {
"none detected".to_string()
} else {
parts.join(", ")
}
}
fn build_diagnose_placement(
worker_selection: &Option<DiagnoseWorkerSelection>,
role: rch_common::BoxRole,
) -> PlacementPlan {
refine_diagnose_placement(
rch_common::placement::resolve_placement_with_role(|key| std::env::var(key).ok(), role),
worker_selection,
)
}
fn refine_diagnose_placement(
mut plan: PlacementPlan,
worker_selection: &Option<DiagnoseWorkerSelection>,
) -> PlacementPlan {
let effective_worker = worker_selection
.as_ref()
.and_then(|s| s.worker.as_ref())
.map(|w| w.id.to_string());
plan = plan.with_effective_worker(effective_worker.clone());
if let Some(requested) = plan.requested_worker.clone() {
let requested_workers = requested
.split(',')
.map(str::trim)
.filter(|worker| !worker.is_empty())
.collect::<Vec<_>>();
let effective_requested = effective_worker
.as_deref()
.filter(|effective| requested_workers.contains(effective));
let outcome = if let Some(effective) = effective_requested {
evaluate_requested_worker(&RequestedWorkerFacts::admissible(effective))
} else if let Some(diagnostics) = worker_selection
.as_ref()
.and_then(|selection| selection.diagnostics.as_ref())
{
requested_workers
.iter()
.map(|requested| {
diagnostics
.workers
.iter()
.find(|worker| worker.worker_id.as_str() == *requested)
.map_or_else(
|| {
evaluate_requested_worker(&RequestedWorkerFacts {
requested: Some((*requested).to_string()),
exists: false,
..RequestedWorkerFacts::none()
})
},
|worker| {
evaluate_requested_worker(&RequestedWorkerFacts::from_diagnostic(
(*requested).to_string(),
worker,
))
},
)
})
.find(|outcome| outcome.status.is_refusal())
.unwrap_or_else(RequestedWorkerOutcome::requested)
} else {
RequestedWorkerOutcome::requested()
};
plan = plan.with_requested_worker_outcome(outcome);
}
plan
}
pub async fn diagnose(command: &str, dry_run: bool, ctx: &OutputContext) -> Result<()> {
use rch_common::classify_command_detailed;
let style = ctx.theme();
let loaded = crate::config::load_config_with_sources()?;
let config = loaded.config;
let topology_policy = config.path_topology.to_policy();
let details = classify_command_detailed(command);
let threshold = config.compilation.confidence_threshold;
let value_sources = collect_value_sources(&config, &loaded.sources);
let threshold_source = value_sources
.iter()
.find(|s| s.key == "compilation.confidence_threshold")
.map(|s| s.source.clone())
.unwrap_or_else(|| "default".to_string());
let mut decision = build_diagnose_decision_with_config(&details.classification, &config);
if decision.would_intercept
&& let Ok(cwd) = std::env::current_dir()
&& let Some(reason) = crate::hook::project_topology_local_reason(&topology_policy, &cwd)
{
decision = DiagnoseDecision {
would_intercept: false,
reason: format!("Would run locally: {reason}"),
};
}
let would_intercept = decision.would_intercept;
debug!(
"Diagnose input='{}' normalized='{}'",
details.original, details.normalized
);
debug!(
"Classification confidence={:.2} reason='{}'",
details.classification.confidence, details.classification.reason
);
debug!(
"Confidence threshold={:.2} source='{}'",
threshold, threshold_source
);
for tier in &details.tiers {
debug!(
"Tier {} {} decision={:?} reason='{}'",
tier.tier, tier.name, tier.decision, tier.reason
);
}
let required_runtime = required_runtime_for_kind(details.classification.kind);
let socket_path = config.general.socket_path.clone();
let socket_exists = Path::new(&socket_path).exists();
let capabilities_future = probe_local_capabilities();
let daemon_health_future = async {
if socket_exists {
query_daemon_health(&socket_path).await.ok()
} else {
None
}
};
let (local_capabilities, daemon_health) =
tokio::join!(capabilities_future, daemon_health_future);
let local_has_any = has_any_capabilities(&local_capabilities);
let mut capabilities_warnings = Vec::new();
let mut daemon_status = DiagnoseDaemonStatus {
socket_path: socket_path.clone(),
socket_exists,
reachable: false,
status: None,
version: None,
uptime_seconds: None,
error: None,
};
if let Some(health) = daemon_health {
daemon_status.reachable = true;
daemon_status.status = Some(health.status);
daemon_status.version = Some(health.version);
daemon_status.uptime_seconds = Some(health.uptime_seconds);
debug!(
"Daemon health ok status='{}' version='{}' uptime={}s",
daemon_status.status.as_deref().unwrap_or("unknown"),
daemon_status.version.as_deref().unwrap_or("unknown"),
daemon_status.uptime_seconds.unwrap_or(0)
);
} else if socket_exists {
daemon_status.error = Some("health check failed".to_string());
debug!("Daemon health check failed");
} else {
daemon_status.error = Some("daemon socket not found".to_string());
debug!("Daemon socket not found: {}", socket_path);
}
let mut worker_selection = None;
if would_intercept && daemon_status.reachable {
let (estimated_cores, cargo_jobs) =
build_diagnose_slot_estimate(details.classification.kind, command, &config.compilation);
let project_root = std::env::current_dir().ok();
let normalized_project_root = project_root.as_ref().and_then(|path| {
normalize_project_path_with_policy(path, &topology_policy)
.map(|normalized| normalized.canonical_path().to_path_buf())
.ok()
});
let project = extract_project_name_with_policy(&topology_policy);
let toolchain = project_root
.as_ref()
.or(normalized_project_root.as_ref())
.and_then(|root| detect_toolchain(root).ok());
let preferred_workers = preferred_workers();
match query_daemon_dry_run(
&socket_path,
&project,
estimated_cores,
config.compilation.disk_headroom_gib,
command,
toolchain.as_ref(),
required_runtime,
&preferred_workers,
)
.await
{
Ok(response) => {
worker_selection = Some(DiagnoseWorkerSelection {
estimated_cores,
cargo_jobs,
worker: response.worker.clone(),
reason: response.reason.clone(),
diagnostics: response.diagnostics.clone(),
});
if let Some(worker) = response.worker.as_ref() {
debug!(
"Worker selected id='{}' slots_free={} speed_score={:.2} reason={:?}",
worker.id, worker.slots_available, worker.speed_score, response.reason
);
} else {
debug!("No worker selected reason={:?}", response.reason);
}
}
Err(err) => {
daemon_status.error = Some(format!("selection request failed: {}", err));
debug!("Worker selection request failed: {}", err);
}
}
}
if details.classification.is_compilation {
if daemon_status.reachable {
match query_workers_capabilities(false).await {
Ok(response) => {
if required_runtime != RequiredRuntime::None {
let missing: Vec<String> = response
.workers
.iter()
.filter(|worker| {
let caps = &worker.capabilities;
match &required_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() {
capabilities_warnings.push(format!(
"Workers missing required runtime {}: {}",
runtime_label(&required_runtime),
missing.join(", ")
));
}
}
if local_has_any {
capabilities_warnings.extend(collect_local_capability_warnings(
&response.workers,
&local_capabilities,
));
}
}
Err(err) => {
capabilities_warnings.push(format!("Worker capabilities unavailable: {}", err));
}
}
} else if required_runtime != RequiredRuntime::None {
capabilities_warnings
.push("Worker capabilities unavailable (daemon not reachable)".to_string());
}
}
let placement = build_diagnose_placement(&worker_selection, config.general.role);
let dry_run_summary = if dry_run {
Some(build_dry_run_summary(
would_intercept,
&decision.reason,
&worker_selection,
daemon_status.reachable,
))
} else {
None
};
let speed_gate = would_intercept.then(|| {
crate::hook::diagnose_speed_gate(
&config,
&extract_project_name_with_policy(&topology_policy),
details.classification.kind,
)
});
if ctx.is_json() {
let response = DiagnoseResponse {
classification: details.classification.clone(),
tiers: details.tiers.clone(),
command: details.original.clone(),
normalized_command: details.normalized.clone(),
decision: DiagnoseDecision {
would_intercept: decision.would_intercept,
reason: decision.reason.clone(),
},
threshold: DiagnoseThreshold {
value: threshold,
source: threshold_source.clone(),
},
daemon: daemon_status,
required_runtime,
local_capabilities: local_has_any.then(|| local_capabilities.clone()),
capabilities_warnings: capabilities_warnings.clone(),
worker_selection,
placement: placement.clone(),
speed_gate: speed_gate.clone(),
dry_run: dry_run_summary.clone(),
};
ctx.json(&ApiResponse::ok("diagnose", response))?;
return Ok(());
}
if dry_run {
println!("{}", style.format_header("RCH Dry Run"));
println!();
if let Some(ref summary) = dry_run_summary {
let offload_label = if summary.would_offload {
style.format_success("YES")
} else {
style.format_warning("NO")
};
println!(
"{} {} {}",
style.key("Would offload:"),
offload_label,
style.muted(&format!("({})", summary.reason))
);
println!();
println!("{}", style.highlight("Pipeline Steps"));
for step in &summary.pipeline_steps {
let status = if step.skipped {
style.muted("[SKIP]")
} else {
style.success("[RUN]")
};
println!(
" {} {} {}",
status,
style.value(&format!("{}.", step.step)),
style.highlight(&step.name)
);
println!(" {}", style.muted(&step.description));
if let Some(ref reason) = step.skip_reason {
println!(" {} {}", style.warning("Skip:"), style.muted(reason));
}
if let Some(ms) = step.estimated_duration_ms {
println!(" {} ~{}ms", style.muted("Est:"), ms);
}
}
if let Some(ref transfer) = summary.transfer_estimate {
println!();
println!("{}", style.highlight("Transfer Estimate"));
println!(
" {} {} ({} files)",
style.key("Size:"),
style.value(&transfer.human_size),
transfer.files
);
println!(" {} ~{}ms", style.key("Time:"), transfer.estimated_time_ms);
if transfer.would_skip {
println!(
" {} {}",
style.warning("Would skip:"),
transfer
.skip_reason
.as_deref()
.unwrap_or("threshold exceeded")
);
}
}
if let Some(total_ms) = summary.total_estimated_ms {
println!();
println!("{} ~{}ms", style.key("Total estimated:"), total_ms);
}
}
println!();
println!(
" {} This is a dry run. No network calls were made.",
style.muted("ℹ")
);
return Ok(());
}
println!("{}", style.format_header("RCH Diagnose"));
println!();
println!("{}", style.highlight("Command Analysis"));
println!(
" {} {}",
style.key("Input:"),
style.value(details.original.trim())
);
if details.normalized != details.original.trim() {
println!(
" {} {}",
style.key("Normalized:"),
style.value(details.normalized.trim())
);
}
println!(" {} {}", style.key("Tool:"), style.value("Bash"));
println!();
println!("{}", style.highlight("Classification"));
let kind_label = details
.classification
.kind
.map(|k| format!("{:?}", k))
.unwrap_or_else(|| "none".to_string());
println!(" {} {}", style.key("Kind:"), style.value(&kind_label));
println!(
" {} {} {}",
style.key("Confidence:"),
style.value(&format!("{:.2}", details.classification.confidence)),
style.muted(&format!("({})", details.classification.reason))
);
println!(
" {} {} {}",
style.key("Threshold:"),
style.value(&format!("{:.2}", threshold)),
style.muted(&format!("# from {}", threshold_source))
);
let decision_label = if would_intercept {
style.format_success("WOULD INTERCEPT")
} else {
style.format_warning("WOULD NOT INTERCEPT")
};
println!(" {} {}", style.key("Decision:"), decision_label);
println!(
" {} {}",
style.key("Reason:"),
style.value(&decision.reason)
);
if let Some(gate) = speed_gate.as_ref().filter(|gate| gate.enabled) {
println!(
" {} {} {}",
style.key("Speed gate:"),
style.value(gate.decision.as_deref().unwrap_or("unknown")),
style.muted(&format!(
"# {} (local_slots {}, free memory {}%, needs {}%)",
gate.reason.as_deref().unwrap_or(""),
gate.local_slots,
gate.free_memory_pct
.map_or_else(|| "?".to_string(), |pct| pct.to_string()),
gate.min_free_memory_pct
))
);
}
println!();
println!("{}", style.highlight("Runtime Capabilities"));
println!(
" {} {}",
style.key("Required runtime:"),
style.value(runtime_label(&required_runtime))
);
println!(
" {} {}",
style.key("Local runtimes:"),
style.value(&summarize_capabilities(&local_capabilities))
);
if capabilities_warnings.is_empty() {
println!(" {} {}", style.key("Warnings:"), style.value("none"));
} else {
for warning in &capabilities_warnings {
println!(
" {} {}",
StatusIndicator::Warning.display(style),
style.warning(warning)
);
}
}
println!();
println!("{}", style.highlight("Tier Decisions"));
for tier in &details.tiers {
let decision = match tier.decision {
rch_common::TierDecision::Pass => style.format_success("PASS"),
rch_common::TierDecision::Reject => style.format_warning("REJECT"),
};
println!(
" {} {} {} {}",
style.key(&format!("Tier {}:", tier.tier)),
style.value(&tier.name),
style.muted("→"),
decision
);
println!(" {} {}", style.muted("reason:"), tier.reason);
}
println!();
println!("{}", style.highlight("Daemon Status"));
println!(
" {} {}",
style.key("Socket:"),
style.value(&daemon_status.socket_path)
);
println!(
" {} {}",
style.key("Socket exists:"),
style.value(&daemon_status.socket_exists.to_string())
);
println!(
" {} {}",
style.key("Reachable:"),
style.value(&daemon_status.reachable.to_string())
);
if let Some(status) = &daemon_status.status {
println!(" {} {}", style.key("Status:"), style.value(status));
}
if let Some(version) = &daemon_status.version {
println!(" {} {}", style.key("Version:"), style.value(version));
}
if let Some(uptime) = daemon_status.uptime_seconds {
println!(
" {} {}s",
style.key("Uptime:"),
style.value(&uptime.to_string())
);
}
if let Some(error) = &daemon_status.error {
println!(" {} {}", style.key("Error:"), style.value(error));
}
println!();
println!("{}", style.highlight("Transfer Configuration"));
let config_exclude_count = config.transfer.exclude_patterns.len();
let project_root = std::env::current_dir().ok();
let rchignore_count = project_root
.as_ref()
.and_then(|root| crate::transfer::parse_rchignore(&root.join(".rchignore")).ok())
.map(|patterns| patterns.len())
.unwrap_or(0);
let effective_count = config_exclude_count + rchignore_count;
println!(
" {} {} {}",
style.key("Exclude patterns:"),
style.value(&effective_count.to_string()),
style.muted(&format!(
"({} from config, {} from .rchignore)",
config_exclude_count, rchignore_count
))
);
if rchignore_count > 0 {
println!(
" {} {}",
style.key(".rchignore:"),
style.format_success("detected")
);
} else {
println!(
" {} {}",
style.key(".rchignore:"),
style.muted("not found")
);
}
println!(
" {} {}",
style.key("Compression:"),
style.value(&format!("zstd level {}", config.transfer.compression_level))
);
println!(
" {} {}",
style.key("Remote base:"),
style.value(&config.transfer.remote_base)
);
println!();
println!("{}", style.highlight("Worker Selection (simulated)"));
if let Some(selection) = &worker_selection {
match &selection.worker {
Some(worker) => {
println!(
" {} {}",
style.key("Selected:"),
style.value(&worker.id.to_string())
);
println!(
" {} {}@{}",
style.key("Host:"),
style.value(&worker.user),
style.value(&worker.host)
);
println!(
" {} {}",
style.key("Slots free:"),
style.value(&worker.slots_available.to_string())
);
println!(" {} {:.1}", style.key("Speed score:"), worker.speed_score);
println!(
" {} {}",
style.key("Reason:"),
style.value(&selection.reason.to_string())
);
}
None => {
println!(
" {} {}",
style.key("Result:"),
style.value("No worker selected")
);
println!(
" {} {}",
style.key("Reason:"),
style.value(&selection.reason.to_string())
);
}
}
} else if !would_intercept {
println!(
" {} {}",
style.key("Skipped:"),
style.value("Command would not be intercepted")
);
} else if !daemon_status.reachable {
println!(
" {} {}",
style.key("Skipped:"),
style.value("Daemon not reachable")
);
} else {
println!(
" {} {}",
style.key("Skipped:"),
style.value("Selection unavailable")
);
}
println!();
println!("{}", style.highlight("Placement Controls"));
println!(
" {} {}",
style.key("Requested worker:"),
style.value(placement.requested_worker.as_deref().unwrap_or("(none)"))
);
println!(
" {} {}",
style.key("Effective worker:"),
style.value(placement.effective_worker.as_deref().unwrap_or("(none)"))
);
if let Some(profile) = placement.requested_profile.as_deref() {
println!(
" {} {}",
style.key("Requested profile:"),
style.value(profile)
);
}
println!(
" {} {}",
style.key("Strict remote:"),
style.value(placement.strict_remote_policy.as_str())
);
println!(
" {} {}",
style.key("Queue policy:"),
style.value(placement.queue_policy.as_str())
);
println!(
" {} {}",
style.key("Visibility:"),
style.value(placement.visibility_mode.as_str())
);
if let Some(ms) = placement.wait_timeout_ms {
println!(
" {} {}ms",
style.key("Wait timeout:"),
style.value(&ms.to_string())
);
}
println!(
" {} {}",
style.key("Target dir:"),
style.value(placement.target_dir_policy.as_str())
);
let outcome = &placement.requested_worker_outcome;
if outcome.status.is_refusal() {
println!(
" {} {} {}",
style.key("Requested worker:"),
style.format_warning(&format!("REFUSED ({})", outcome.status.as_str())),
style.muted(
&outcome
.reason_code
.as_deref()
.map(|c| format!("[{c}]"))
.unwrap_or_default()
)
);
if let Some(action) = outcome.next_action.as_deref() {
println!(" {} {}", style.muted("next:"), style.value(action));
}
} else {
println!(
" {} {}",
style.key("Requested-worker outcome:"),
style.value(outcome.status.as_str())
);
}
for diag in &placement.diagnostics {
println!(
" {} {} {}",
StatusIndicator::Warning.display(style),
style.muted(&format!("[{}]", diag.control)),
style.warning(&diag.message)
);
}
Ok(())
}
pub(super) fn merge_job_required_tools(
mut tools: Vec<String>,
config_tools: &[String],
job: bool,
) -> Vec<String> {
if job {
for tool in config_tools {
let tool = tool.trim();
if !tool.is_empty() && !tools.iter().any(|existing| existing == tool) {
tools.push(tool.to_string());
}
}
}
tools
}
pub(super) fn format_admit_required_capabilities(
req: &rch_common::admit_preflight::RequiredCapabilities,
) -> Vec<String> {
let mut needs: Vec<String> = Vec::new();
if req.needs_cargo {
needs.push("cargo".to_string());
}
if req.needs_zig {
needs.push("zig".to_string());
}
if req.needs_bun {
needs.push("bun".to_string());
}
if req.needs_nix {
needs.push("nix".to_string());
}
if req.needs_go {
needs.push("go".to_string());
}
if req.needs_node {
needs.push("node".to_string());
}
if let Some(os) = &req.needs_os {
needs.push(format!("os:{os}"));
}
for t in &req.needs_targets {
needs.push(format!("target:{t}"));
}
for tc in &req.needs_toolchains {
needs.push(format!("toolchain:{tc}"));
}
for tool in &req.needs_tools {
needs.push(format!("tool:{tool}"));
}
needs
}
pub async fn admit(
command: &str,
job: bool,
require_tool: Vec<String>,
ctx: &OutputContext,
) -> Result<()> {
use rch_common::admit_preflight::preflight;
let require_tool = if job && let Ok(config) = crate::config::load_config() {
merge_job_required_tools(require_tool, &config.jobs.required_tools, true)
} else {
require_tool
};
let proof_policy = !require_tool.is_empty()
|| ["RCH_REQUIRE_REMOTE", "RCH_FORCE_REMOTE"]
.iter()
.any(|key| {
std::env::var(key).is_ok_and(|v| {
matches!(
v.trim().to_ascii_lowercase().as_str(),
"1" | "true" | "yes" | "on"
)
})
});
let pf = preflight(command, proof_policy);
let pf = if job {
pf.as_job(require_tool)
} else {
pf.with_required_tools(require_tool)
};
if ctx.is_json() {
ctx.json(&ApiResponse::ok("admit", &pf))?;
return Ok(());
}
let style = ctx.theme();
println!("{}", style.format_header("RCH Admit"));
println!();
println!("{} {}", style.key("Recommendation:"), {
let r = pf.base_recommendation.as_str();
match pf.base_recommendation {
rch_common::admit_preflight::AdmitRecommendation::Offload => style.format_success(r),
rch_common::admit_preflight::AdmitRecommendation::Local => style.format_warning(r),
_ => style.value(r).to_string(),
}
});
if pf.job_mode {
println!(
"{} {}",
style.key("Mode:"),
style.value("job (classifier bypassed)")
);
}
println!("{} {}", style.key("Compilation:"), pf.is_compilation);
if let Some(family) = &pf.family {
println!("{} {}", style.key("Family:"), style.value(family));
}
if pf.compound.len() > 1 {
println!(
"{} {} parts",
style.key("Compound:"),
style.value(&pf.compound.len().to_string())
);
}
let needs = format_admit_required_capabilities(&pf.required);
println!(
"{} {}",
style.key("Required capabilities:"),
if needs.is_empty() {
style.muted("none").to_string()
} else {
style.value(&needs.join(", ")).to_string()
}
);
if pf.proof_policy {
println!("{} {}", style.key("Proof policy:"), style.value("strict"));
}
println!();
println!("{}", style.muted(&pf.detail));
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub async fn self_test(
action: Option<crate::SelfTestAction>,
worker: Option<String>,
all: bool,
project: Option<std::path::PathBuf>,
timeout: u64,
debug: bool,
scheduled: bool,
smoke: bool,
soak: bool,
load: bool,
dry_run: bool,
ctx: &OutputContext,
) -> Result<()> {
if smoke || load {
return self_test_smoke(worker, all, timeout, dry_run, soak, load, ctx).await;
}
match action {
Some(crate::SelfTestAction::Status) => self_test_status(ctx).await,
Some(crate::SelfTestAction::History { limit }) => self_test_history(limit, ctx).await,
None => self_test_run(worker, all, project, timeout, debug, scheduled, ctx).await,
}
}
async fn self_test_status(ctx: &OutputContext) -> Result<()> {
let response = send_daemon_command("GET /self-test/status\n").await?;
let json = extract_json_body(&response).ok_or_else(|| anyhow::anyhow!("Invalid response"))?;
let status: SelfTestStatusResponseFromApi = serde_json::from_str(json)?;
ctx.json(&status)?;
if ctx.is_json() {
return Ok(());
}
let style = ctx.style();
println!("{}", style.format_header("Self-Test Status"));
println!(
" {} {} {}",
style.key("Scheduled"),
style.muted(":"),
if status.enabled {
style.success("Enabled")
} else {
style.warning("Disabled")
}
);
if let Some(schedule) = status.schedule.as_ref() {
println!(
" {} {} {}",
style.key("Schedule"),
style.muted(":"),
style.info(schedule)
);
}
if let Some(interval) = status.interval.as_ref() {
println!(
" {} {} {}",
style.key("Interval"),
style.muted(":"),
style.info(interval)
);
}
if let Some(last) = status.last_run.as_ref() {
println!(
" {} {} {} ({} passed, {} failed)",
style.key("Last run"),
style.muted(":"),
style.info(&last.completed_at),
last.workers_passed,
last.workers_failed
);
}
if let Some(next) = status.next_run.as_ref() {
println!(
" {} {} {}",
style.key("Next run"),
style.muted(":"),
style.info(next)
);
}
Ok(())
}
async fn self_test_history(limit: usize, ctx: &OutputContext) -> Result<()> {
let command = format!("GET /self-test/history?limit={}\n", limit);
let response = send_daemon_command(&command).await?;
let json = extract_json_body(&response).ok_or_else(|| anyhow::anyhow!("Invalid response"))?;
let history: SelfTestHistoryResponseFromApi = serde_json::from_str(json)?;
ctx.json(&history)?;
if ctx.is_json() {
return Ok(());
}
let style = ctx.style();
println!("{}", style.format_header("Self-Test History"));
if history.runs.is_empty() {
println!(" {}", style.muted("No self-test runs recorded."));
return Ok(());
}
let rows: Vec<Vec<String>> = history
.runs
.iter()
.map(|run| {
vec![
run.id.to_string(),
run.run_type.clone(),
run.completed_at.clone(),
format!("{}ms", run.duration_ms),
run.workers_passed.to_string(),
run.workers_failed.to_string(),
]
})
.collect();
ctx.table(
&["ID", "Type", "Completed", "Duration", "Passed", "Failed"],
&rows,
);
for run in &history.runs {
if run.workers_failed == 0 {
continue;
}
println!(
"\n {} {}",
style.key("Failures for run"),
style.highlight(&run.id.to_string())
);
for result in history
.results
.iter()
.filter(|r| r.run_id == run.id && !r.passed)
{
let error = result
.error
.clone()
.unwrap_or_else(|| "unknown".to_string());
println!(
" {} {}: {}",
StatusIndicator::Error.display(style),
style.highlight(&result.worker_id),
style.error(&error)
);
}
}
Ok(())
}
fn default_self_test_worker<'a>(
workers: &'a [WorkerConfig],
disabled: &rch_common::bypass_record::AdminDisableStore,
) -> Option<&'a WorkerConfig> {
workers
.iter()
.find(|worker| disabled.get(worker.id.as_str()).is_none())
.or_else(|| workers.first())
}
async fn self_test_run(
worker: Option<String>,
all: bool,
project: Option<std::path::PathBuf>,
timeout: u64,
debug: bool,
scheduled: bool,
ctx: &OutputContext,
) -> Result<()> {
let mut worker_ids = Vec::new();
if scheduled {
} else if all {
} else if let Some(worker) = worker {
worker_ids.push(worker);
} else {
let workers = load_workers_from_config()?;
let disabled = rch_common::bypass_record::AdminDisableStore::load(
rch_common::bypass_record::default_admin_disable_path(),
);
let first = default_self_test_worker(&workers, &disabled)
.ok_or_else(|| anyhow::anyhow!("No workers configured"))?;
worker_ids.push(first.id.to_string());
}
let mut query = Vec::new();
for id in &worker_ids {
query.push(format!("worker={}", urlencoding_encode(id)));
}
if all {
query.push("all=true".to_string());
}
if scheduled {
query.push("scheduled=true".to_string());
}
if let Some(path) = project.as_ref() {
query.push(format!(
"project={}",
urlencoding_encode(&path.display().to_string())
));
}
if timeout > 0 {
query.push(format!("timeout={}", timeout));
}
if debug {
query.push("debug=true".to_string());
}
let command = if query.is_empty() {
"POST /self-test/run\n".to_string()
} else {
format!("POST /self-test/run?{}\n", query.join("&"))
};
let spinner = if !ctx.is_json() {
let target_desc = if all {
"all workers".to_string()
} else if worker_ids.is_empty() {
"default worker".to_string()
} else {
worker_ids.join(", ")
};
Some(Spinner::new(
ctx,
&format!("Running self-test on {}...", target_desc),
))
} else {
None
};
let response = send_daemon_command(&command).await;
let response = match response {
Ok(r) => {
if let Some(ref s) = spinner {
s.finish_and_clear();
}
r
}
Err(e) => {
if let Some(s) = spinner {
s.finish_error(&format!("Failed: {}", e));
}
return Err(e);
}
};
let json = extract_json_body(&response).ok_or_else(|| anyhow::anyhow!("Invalid response"))?;
let run: SelfTestRunResponseFromApi = serde_json::from_str(json)?;
ctx.json(&run)?;
if ctx.is_json() {
return Ok(());
}
let style = ctx.style();
println!("{}", style.format_header("Self-Test Result"));
println!(
" {} {} {}",
style.key("Run"),
style.muted(":"),
style.info(&run.run.completed_at)
);
println!(
" {} {} {} passed, {} failed",
style.key("Workers"),
style.muted(":"),
style.success(&run.run.workers_passed.to_string()),
style.error(&run.run.workers_failed.to_string())
);
let mut timed_out = 0usize;
for result in &run.results {
let status = if result.passed {
StatusIndicator::Success.display(style)
} else {
StatusIndicator::Error.display(style)
};
let is_timeout = result
.error
.as_deref()
.is_some_and(|e| e.contains("[RCH-E203]") || e.to_lowercase().contains("timed out"));
if is_timeout {
timed_out += 1;
}
let detail = if result.passed {
format!(
"remote={}ms local={}ms",
result.remote_time_ms.unwrap_or(0),
result.local_time_ms.unwrap_or(0)
)
} else {
result
.error
.clone()
.unwrap_or_else(|| "unknown error".to_string())
};
println!(
" {} {}: {}",
status,
style.highlight(&result.worker_id),
detail
);
if ctx.is_verbose() {
debug!(target: "rch::verbose", worker_id = %result.worker_id, "rendering verbose self-test result details");
for line in render_self_test_result_verbose_lines(result, style) {
println!("{line}");
}
}
}
let ok = run.run.workers_passed;
let failed_other = run.run.workers_failed.saturating_sub(timed_out);
println!(
"\n{} {} ok, {} timed out, {} failed.",
style.muted("Summary:"),
style.success(&ok.to_string()),
if timed_out > 0 {
style.warning(&timed_out.to_string())
} else {
style.muted("0")
},
if failed_other > 0 {
style.error(&failed_other.to_string())
} else {
style.muted("0")
}
);
Ok(())
}
const SMOKE_BEAD_ID: &str = "bd-session-history-remediation-ocv9i.16.6";
const SOAK_PASSES: usize = 3;
fn smoke_disk_roots() -> Vec<String> {
let base = crate::config::load_config_with_sources()
.ok()
.map(|loaded| loaded.config.transfer.remote_base)
.unwrap_or_else(rch_common::types::default_remote_base);
let mut roots = vec![base.clone()];
if let Some(parent) = std::path::Path::new(&base)
.parent()
.and_then(std::path::Path::to_str)
&& !parent.is_empty()
&& parent != "/"
&& parent != base
{
roots.push(parent.to_string());
}
roots
}
fn smoke_fleet_consistency(
report: &rch_common::fleet_status::FleetStatusReport,
) -> (bool, Option<String>) {
if report.capacity_collapsed() {
return (false, Some("fleet_capacity_collapsed".to_string()));
}
if !report.absence_alerts.is_empty() {
return (false, Some("fleet_workers_absent".to_string()));
}
(true, None)
}
async fn fetch_fleet_status_report() -> Result<rch_common::fleet_status::FleetStatusReport> {
let response = send_daemon_command("GET /status\n").await?;
let json = extract_json_body(&response)
.ok_or_else(|| anyhow::anyhow!("Invalid response format from daemon"))?;
let status: DaemonFullStatusResponse =
serde_json::from_str(json).context("Failed to parse daemon status response")?;
Ok(build_fleet_status_report(&status))
}
fn smoke_canary_verdict(result: &SelfTestResultRecordFromApi) -> (bool, Option<String>) {
if result.passed && result.remote_time_ms.is_some() {
(true, None)
} else {
(
false,
Some(
result
.error
.clone()
.unwrap_or_else(|| "remote_build_failed".to_string()),
),
)
}
}
fn smoke_artifact_verdict(result: &SelfTestResultRecordFromApi) -> (bool, Option<String>) {
if result.remote_hash.is_some() {
(true, None)
} else {
(
false,
Some(
result
.error
.clone()
.unwrap_or_else(|| "artifact_not_retrieved".to_string()),
),
)
}
}
fn smoke_artifact_byte_identical(result: &SelfTestResultRecordFromApi) -> bool {
match (&result.local_hash, &result.remote_hash) {
(Some(local), Some(remote)) => !local.is_empty() && local == remote,
_ => false,
}
}
fn smoke_cancel_verdict(status: Option<&str>) -> (bool, Option<String>) {
match status {
Some("error") => (true, None),
Some(other) => (false, Some(format!("unexpected_cancel_status:{other}"))),
None => (false, Some("cancel_response_malformed".to_string())),
}
}
async fn run_smoke_canary_scenarios(
run_id: &str,
worker_id: &str,
timeout_secs: u64,
remote_base: Option<&str>,
) -> Vec<SmokeProfileEvent> {
let mut events = vec![
SmokeProfileEvent::started(
run_id,
SMOKE_BEAD_ID,
Some(worker_id.to_string()),
SmokeScenario::CargoCanary,
),
SmokeProfileEvent::started(
run_id,
SMOKE_BEAD_ID,
Some(worker_id.to_string()),
SmokeScenario::ArtifactRetrieval,
),
];
let command = format!(
"POST /self-test/run?worker={}&timeout={}&debug=true&retries=0\n",
urlencoding_encode(worker_id),
timeout_secs
);
let canary_fp = Some("self-test canary (cargo build, debug)".to_string());
let artifact_fp = Some("rsync artifacts from worker + hash verify".to_string());
let started = std::time::Instant::now();
let result = send_daemon_command(&command)
.await
.ok()
.and_then(|response| {
let json = extract_json_body(&response)?;
let run: SelfTestRunResponseFromApi = serde_json::from_str(json).ok()?;
run.results.into_iter().find(|r| r.worker_id == worker_id)
});
let elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);
let Some(result) = result else {
for scenario in [SmokeScenario::CargoCanary, SmokeScenario::ArtifactRetrieval] {
events.push(SmokeProfileEvent::outcome(
run_id,
SMOKE_BEAD_ID,
Some(worker_id.to_string()),
scenario,
false,
Some("self_test_result_unavailable".to_string()),
if scenario == SmokeScenario::CargoCanary {
canary_fp.clone()
} else {
artifact_fp.clone()
},
elapsed_ms,
));
}
return events;
};
let (canary_passed, canary_reason) = smoke_canary_verdict(&result);
events.push(SmokeProfileEvent::outcome(
run_id,
SMOKE_BEAD_ID,
Some(worker_id.to_string()),
SmokeScenario::CargoCanary,
canary_passed,
canary_reason,
canary_fp,
result.remote_time_ms.unwrap_or(elapsed_ms),
));
let (artifact_passed, artifact_reason) = smoke_artifact_verdict(&result);
let mut artifact_event = SmokeProfileEvent::outcome(
run_id,
SMOKE_BEAD_ID,
Some(worker_id.to_string()),
SmokeScenario::ArtifactRetrieval,
artifact_passed,
artifact_reason,
artifact_fp,
elapsed_ms,
);
artifact_event.remote_target_dir = remote_base.map(str::to_string);
artifact_event.artifact_summary = Some(if artifact_passed {
let byte_identical = smoke_artifact_byte_identical(&result);
match result.remote_hash.as_deref() {
Some(hash) => format!(
"retrieved: hash {}… ({}ms remote){}",
&hash[..hash.len().min(12)],
result.remote_time_ms.unwrap_or(0),
if byte_identical {
"; bit-identical to local"
} else {
"; differs from local build (non-deterministic/rustc drift)"
}
),
None => "retrieved".to_string(),
}
} else {
"artifact not retrieved".to_string()
});
events.push(artifact_event);
events
}
async fn run_smoke_queue_attach_cancel(run_id: &str) -> SmokeProfileEvent {
const SENTINEL_BUILD_ID: u64 = u64::MAX;
let started = std::time::Instant::now();
let (passed, reason) = smoke_queue_attach_cancel_probe(SENTINEL_BUILD_ID).await;
let ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);
SmokeProfileEvent::outcome(
run_id,
SMOKE_BEAD_ID,
None,
SmokeScenario::QueueAttachCancel,
passed,
reason,
Some("GET /status; POST /builds/<sentinel>/cancel".to_string()),
ms,
)
}
async fn smoke_queue_attach_cancel_probe(sentinel_build_id: u64) -> (bool, Option<String>) {
match send_daemon_command("GET /status\n").await {
Ok(response) => {
let parsed = extract_json_body(&response)
.and_then(|json| serde_json::from_str::<DaemonFullStatusResponse>(json).ok());
if parsed.is_none() {
return (false, Some("queue_snapshot_unreadable".to_string()));
}
}
Err(_) => return (false, Some("queue_snapshot_unavailable".to_string())),
}
let cancel_cmd = format!("POST /builds/{sentinel_build_id}/cancel?force=false\n");
match send_daemon_command(&cancel_cmd).await {
Ok(response) => {
let status = extract_json_body(&response)
.and_then(|json| serde_json::from_str::<serde_json::Value>(json).ok())
.and_then(|v| v.get("status").and_then(|s| s.as_str()).map(str::to_string));
smoke_cancel_verdict(status.as_deref())
}
Err(_) => (false, Some("cancel_unavailable".to_string())),
}
}
fn build_smoke_inputs(
workers_configured: bool,
daemon_reachable: bool,
dry_run: bool,
soak: bool,
selected_worker: Option<String>,
) -> SmokeProfileInputs {
SmokeProfileInputs {
workers_configured,
remote_execution_available: daemon_reachable && workers_configured,
dry_run,
mode: if soak {
ProfileMode::Soak
} else {
ProfileMode::Smoke
},
selected_worker,
}
}
fn smoke_planned_events(
plan: &SmokeProfilePlan,
run_id: &str,
representative_worker: Option<&str>,
) -> Vec<SmokeProfileEvent> {
plan.scenarios
.iter()
.map(|planned| {
SmokeProfileEvent::planned(
run_id,
SMOKE_BEAD_ID,
representative_worker.map(str::to_string),
planned,
)
})
.collect()
}
fn write_smoke_jsonl(run_id: &str, events: &[SmokeProfileEvent]) -> Option<String> {
let dir = dirs::cache_dir()?.join("rch").join("smoke");
std::fs::create_dir_all(&dir).ok()?;
let path = dir.join(format!("{run_id}.jsonl"));
let mut buf = String::new();
for event in events {
let line = serde_json::to_string(event).ok()?;
buf.push_str(&line);
buf.push('\n');
}
std::fs::write(&path, buf).ok()?;
Some(path.display().to_string())
}
async fn self_test_smoke(
worker: Option<String>,
all: bool,
timeout: u64,
dry_run: bool,
soak: bool,
load: bool,
ctx: &OutputContext,
) -> Result<()> {
let workers = load_workers_from_config().unwrap_or_default();
let workers_configured = !workers.is_empty();
let probe_start = std::time::Instant::now();
let daemon_reachable = send_daemon_command("GET /health\n").await.is_ok();
let probe_ms = u64::try_from(probe_start.elapsed().as_millis()).unwrap_or(u64::MAX);
let selected_worker = if all { None } else { worker };
let mut inputs = build_smoke_inputs(
workers_configured,
daemon_reachable,
dry_run,
soak,
selected_worker.clone(),
);
if load {
inputs.mode = ProfileMode::Load;
}
let plan = plan_smoke_profile(&inputs);
let run_id = uuid::Uuid::new_v4().to_string();
if load {
return run_load_storm(
&run_id,
selected_worker.as_deref(),
timeout,
dry_run,
daemon_reachable,
ctx,
)
.await;
}
let mut events = smoke_planned_events(&plan, &run_id, selected_worker.as_deref());
if !dry_run
&& let Some(planned) = plan
.scenarios
.iter()
.find(|p| p.scenario == SmokeScenario::DaemonReachable)
&& planned.action.is_executed()
{
events.push(SmokeProfileEvent::outcome(
run_id.clone(),
SMOKE_BEAD_ID,
None,
SmokeScenario::DaemonReachable,
daemon_reachable,
Some(
rch_common::IncidentReasonCode::DaemonSocketRefused
.code()
.to_string(),
),
Some("GET /health".to_string()),
probe_ms,
));
}
if !dry_run
&& let Some(planned) = plan
.scenarios
.iter()
.find(|p| p.scenario == SmokeScenario::DesiredVsLiveFleet)
&& planned.action.is_executed()
{
let started = std::time::Instant::now();
let (passed, reason) = match fetch_fleet_status_report().await {
Ok(report) => smoke_fleet_consistency(&report),
Err(_) => (false, Some("fleet_status_unavailable".to_string())),
};
let ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);
events.push(SmokeProfileEvent::outcome(
run_id.clone(),
SMOKE_BEAD_ID,
None,
SmokeScenario::DesiredVsLiveFleet,
passed,
reason,
Some("GET /status".to_string()),
ms,
));
}
if !dry_run
&& plan.scenarios.iter().any(|p| {
p.scenario == SmokeScenario::WorkerCapabilitiesExactUserPath && p.action.is_executed()
})
{
let disk_roots = smoke_disk_roots();
let targets: Vec<&WorkerConfig> = match selected_worker.as_deref() {
Some(id) => workers.iter().filter(|w| w.id.0 == id).collect(),
None => workers.iter().collect(),
};
let passes = if soak { SOAK_PASSES } else { 1 };
for _ in 0..passes {
for worker in &targets {
let worker_events = crate::fleet::run_smoke_worker_scenarios(
&run_id,
SMOKE_BEAD_ID,
worker,
&disk_roots,
)
.await;
events.extend(worker_events);
}
}
}
if !dry_run
&& plan
.scenarios
.iter()
.any(|p| p.scenario == SmokeScenario::CargoCanary && p.action.is_executed())
{
let canary_worker = match selected_worker.as_deref() {
Some(id) => workers.iter().find(|w| w.id.0 == id),
None => workers.first(),
};
if let Some(worker) = canary_worker {
let remote_base = smoke_disk_roots().into_iter().next();
let canary_events =
run_smoke_canary_scenarios(&run_id, &worker.id.0, timeout, remote_base.as_deref())
.await;
events.extend(canary_events);
}
}
if !dry_run
&& plan
.scenarios
.iter()
.any(|p| p.scenario == SmokeScenario::QueueAttachCancel && p.action.is_executed())
{
events.push(run_smoke_queue_attach_cancel(&run_id).await);
}
if !dry_run
&& let Some(planned) = plan
.scenarios
.iter()
.find(|p| p.scenario == SmokeScenario::ProofModeRefusal)
&& matches!(planned.action, ScenarioAction::ExpectRefusal { .. })
{
let policy = resolve_placement(|k| (k == "RCH_REQUIRE_REMOTE").then(|| "1".to_string()))
.strict_remote_policy;
let held = policy.fail_closed();
events.push(SmokeProfileEvent::refused(
run_id.clone(),
SMOKE_BEAD_ID,
held,
Some(
rch_common::IncidentReasonCode::ProofRefusal
.code()
.to_string(),
),
0,
));
}
let log_path = write_smoke_jsonl(&run_id, &events);
let payload = serde_json::json!({
"run_id": run_id,
"bead_id": SMOKE_BEAD_ID,
"mode": plan.mode,
"dry_run": dry_run,
"workers_configured": workers_configured,
"remote_execution_available": inputs.remote_execution_available,
"daemon_reachable": daemon_reachable,
"overall_skipped": plan.overall_skipped,
"plan": plan,
"events": events,
"log_path": log_path,
});
ctx.json(&payload)?;
if ctx.is_json() {
return Ok(());
}
render_smoke_plan(&plan, daemon_reachable, log_path.as_deref(), ctx);
Ok(())
}
fn render_smoke_plan(
plan: &SmokeProfilePlan,
daemon_reachable: bool,
log_path: Option<&str>,
ctx: &OutputContext,
) {
let style = ctx.style();
println!("{}", style.format_header("Real-Fleet Smoke Profile (plan)"));
let rows: Vec<Vec<String>> = plan
.scenarios
.iter()
.map(|p| {
vec![
p.scenario.as_str().to_string(),
p.action.status_token().to_string(),
p.action.reason_code().unwrap_or("").to_string(),
]
})
.collect();
ctx.table(&["Scenario", "Action", "Reason"], &rows);
let count = |pred: fn(&ScenarioAction) -> bool| -> usize {
plan.scenarios.iter().filter(|p| pred(&p.action)).count()
};
let run = count(|a| matches!(a, ScenarioAction::Run));
let dry = count(|a| matches!(a, ScenarioAction::DryRun));
let skip = count(|a| matches!(a, ScenarioAction::Skip { .. }));
let refuse = count(|a| matches!(a, ScenarioAction::ExpectRefusal { .. }));
println!(
"\n{} {} run, {} dry-run, {} skip, {} expect-refusal{}.",
style.muted("Summary:"),
style.success(&run.to_string()),
dry,
skip,
refuse,
if plan.overall_skipped {
format!(" ({})", style.warning("real-fleet validation skipped"))
} else {
String::new()
}
);
println!(
" {} {}",
style.key("Daemon reachable"),
if daemon_reachable {
style.success("yes")
} else {
style.warning("no")
}
);
if let Some(path) = log_path {
println!(" {} {}", style.key("JSONL trace"), style.info(path));
}
println!(
" {}",
style.muted(
"Per-worker scenario execution is the operator real-fleet procedure; see docs/runbooks."
)
);
}
fn render_self_test_result_verbose_lines(
result: &SelfTestResultRecordFromApi,
style: &Theme,
) -> Vec<String> {
let mut lines = Vec::new();
if result.passed {
if let (Some(remote), Some(local)) = (result.remote_time_ms, result.local_time_ms) {
let speedup = if remote > 0 {
local as f64 / remote as f64
} else {
1.0
};
lines.push(format!(
" {} {:.1}x speedup",
style.muted("→"),
speedup
));
if let (Some(local_hash), Some(remote_hash)) = (&result.local_hash, &result.remote_hash)
{
let hash_match = if local_hash == remote_hash {
style.success("match")
} else {
style.error("MISMATCH")
};
lines.push(format!(" {} hash {}", style.muted("→"), hash_match));
}
}
} else if let Some(ref err) = result.error
&& err.len() > 50
{
lines.push(format!(
" {} {}",
style.muted("error:"),
style.error(err)
));
}
lines
}
fn local_build_status_hints(
observation: Option<&crate::local_builds::LocalBuildObservation>,
) -> Vec<crate::status_types::RemediationHint> {
let Some(observation) = observation else {
return Vec::new();
};
let check = crate::doctor::local_build_check(observation);
if check.status != crate::commands::DoctorCheckStatus::Warning {
return Vec::new();
}
let message = match check.details {
Some(details) => format!("{}; {details}", check.message),
None => check.message,
};
vec![crate::status_types::RemediationHint {
reason_code: "dispatcher_local_builds".to_string(),
severity: "warning".to_string(),
message,
suggested_action: check
.suggestion
.unwrap_or_else(|| "Check local process and cache access".to_string()),
worker_id: None,
}]
}
fn now_unix_ms_for_status() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX))
.unwrap_or(0)
}
fn local_fallback_issue(now_ms: u64) -> Option<IssueFromApi> {
let ledger = crate::hook::configured_incident_ledger();
local_fallback_issue_from(&ledger.read_all(), now_ms)
}
fn local_fallback_issue_from(
events: &[rch_common::IncidentEvent],
now_ms: u64,
) -> Option<IssueFromApi> {
const WINDOW_MS: u64 = 24 * 60 * 60 * 1000;
let cutoff = now_ms.saturating_sub(WINDOW_MS);
let mut by_reason: std::collections::BTreeMap<&str, usize> = std::collections::BTreeMap::new();
for event in events {
if event.reason_code == rch_common::IncidentReasonCode::LocalFallback
&& event.occurred_at_unix_ms >= cutoff
{
let reason = event
.details
.get("reason")
.map_or("unknown", String::as_str);
*by_reason.entry(reason).or_default() += 1;
}
}
let total: usize = by_reason.values().sum();
if total == 0 {
return None;
}
let mut ranked: Vec<(&str, usize)> = by_reason.into_iter().collect();
ranked.sort_by(|a, b| b.1.cmp(&a.1).then(a.0.cmp(b.0)));
let top = ranked
.iter()
.take(3)
.map(|(reason, count)| format!("{count}× {reason}"))
.collect::<Vec<_>>()
.join("; ");
Some(IssueFromApi {
severity: "warning".to_string(),
summary: format!("{total} local fallback build(s) on this host in the last 24h ({top})"),
remediation: Some(
"Reasons are recorded as RCH-I011 in the incident ledger (~/.local/state/rch/incidents.jsonl); `rch diagnose -- <cmd>` explains a single decision".to_string(),
),
})
}
fn status_error_with_local_builds(
error: anyhow::Error,
hints: &[crate::status_types::RemediationHint],
) -> anyhow::Error {
if hints.is_empty() {
return error;
}
let warning = hints
.iter()
.map(|hint| hint.message.as_str())
.collect::<Vec<_>>()
.join("; ");
let code = if error.downcast_ref::<serde_json::Error>().is_some() {
rch_common::ErrorCode::InternalSerdeError
} else {
rch_common::ErrorCode::InternalDaemonSocket
};
let api_error = rch_common::ApiError::from_code(code)
.with_details(format!("{error:#}"))
.with_context("local_build_warning", warning);
error.context(api_error)
}
fn status_view_with_local_builds(
report: &impl serde::Serialize,
hints: &[crate::status_types::RemediationHint],
) -> Result<serde_json::Value> {
let mut value = serde_json::to_value(report)?;
if !hints.is_empty() {
value["local_build_hints"] = serde_json::to_value(hints)?;
}
Ok(value)
}
pub async fn status_overview(
workers: bool,
jobs: bool,
fleet: bool,
remediation: bool,
ctx: &OutputContext,
) -> Result<()> {
use crate::status_types::{
CliStatusResponse, RemediationHint, RepoConvergenceStatusFromApi, STATUS_SCHEMA_VERSION,
SystemPosture, critical_pressure_worker_count, generate_convergence_remediations,
generate_worker_remediations, remote_admissible_worker_count,
};
let observation = crate::config::load_config()
.ok()
.and_then(|config| crate::local_builds::observe(config.general.role));
let local_hints = local_build_status_hints(observation.as_ref());
for hint in &local_hints {
ctx.warning(&hint.message);
}
let status_result: Result<DaemonFullStatusResponse> = async {
let response = send_daemon_command("GET /status\n").await?;
let json = extract_json_body(&response)
.ok_or_else(|| anyhow::anyhow!("Invalid response format from daemon"))?;
serde_json::from_str(json).context("Failed to parse daemon status response")
}
.await;
let fallback_issue = local_fallback_issue(now_unix_ms_for_status());
let mut status = status_result.map_err(|error| {
let mut hints = local_hints.clone();
if let Some(issue) = fallback_issue.as_ref() {
hints.push(crate::status_types::RemediationHint {
reason_code: "local_fallbacks".to_string(),
severity: issue.severity.clone(),
message: issue.summary.clone(),
suggested_action: issue.remediation.clone().unwrap_or_default(),
worker_id: None,
});
}
status_error_with_local_builds(error, &hints)
})?;
if let Some(issue) = fallback_issue {
status.issues.push(issue);
}
if fleet {
let report = build_fleet_status_report(&status);
if ctx.is_json() {
ctx.json(&ApiResponse::ok(
"status-fleet",
status_view_with_local_builds(&report, &local_hints)?,
))?;
} else {
crate::status_display::render_fleet_status(&report, ctx.style());
}
return Ok(());
}
if remediation {
let view = status
.remediation
.clone()
.unwrap_or_else(|| build_remediation_view_from_status(&status));
if ctx.is_json() {
ctx.json(&ApiResponse::ok(
"status-remediation",
status_view_with_local_builds(&view, &local_hints)?,
))?;
} else {
crate::status_display::render_remediation_view(&view, ctx.style());
}
return Ok(());
}
let convergence = match send_daemon_command("GET /repo-convergence/status\n").await {
Ok(conv_response) => extract_json_body(&conv_response)
.and_then(|j| serde_json::from_str::<RepoConvergenceStatusFromApi>(j).ok()),
Err(_) => None,
};
let posture = SystemPosture::from_status(&status);
let mut remediation_hints: Vec<RemediationHint> = generate_worker_remediations(&status.workers);
if ctx.is_json() {
remediation_hints.extend(local_hints);
}
if let Some(ref conv) = convergence {
remediation_hints.extend(generate_convergence_remediations(conv));
}
match &posture {
SystemPosture::LocalOnly => {
if status.daemon.workers_total == 0 {
remediation_hints.push(RemediationHint {
reason_code: "no_workers_configured".into(),
severity: "critical".into(),
message: "No workers configured; all builds run locally".into(),
suggested_action: "rch workers init or edit ~/.config/rch/workers.toml".into(),
worker_id: None,
});
} else if critical_pressure_worker_count(&status.workers) > 0
&& remote_admissible_worker_count(&status.workers) == 0
{
remediation_hints.push(RemediationHint {
reason_code: "all_workers_pressure_critical".into(),
severity: "critical".into(),
message: format!(
"No admissible remote workers; {} worker(s) are blocked by critical storage pressure",
critical_pressure_worker_count(&status.workers)
),
suggested_action:
"Use the pressure_critical worker hints for read-only df/du attribution; free space only with explicit operator approval, then run rch workers capabilities --refresh"
.into(),
worker_id: None,
});
} else {
remediation_hints.push(RemediationHint {
reason_code: "all_workers_down".into(),
severity: "critical".into(),
message: "All workers unreachable; builds falling back to local".into(),
suggested_action: "rch doctor --fix".into(),
worker_id: None,
});
}
}
SystemPosture::Degraded => {
let pressure_blocked = critical_pressure_worker_count(&status.workers);
remediation_hints.push(RemediationHint {
reason_code: "partial_capacity".into(),
severity: "warning".into(),
message: if pressure_blocked > 0 {
format!(
"Operating at reduced capacity: {}/{} workers healthy, {} pressure-blocked",
status.daemon.workers_healthy,
status.daemon.workers_total,
pressure_blocked
)
} else {
format!(
"Operating at reduced capacity: {}/{} workers healthy",
status.daemon.workers_healthy, status.daemon.workers_total
)
},
suggested_action: if pressure_blocked > 0 {
"Use rch status --workers --json pressure hints before retrying heavy builds"
.into()
} else {
"rch workers probe --all".into()
},
worker_id: None,
});
}
SystemPosture::RemoteReady => {}
}
if ctx.is_json() {
let cli_status = CliStatusResponse {
schema_version: STATUS_SCHEMA_VERSION.to_string(),
posture_description: posture.description().to_string(),
posture,
daemon: status,
convergence,
remediation_hints,
};
ctx.json(&ApiResponse::ok("status", &cli_status))?;
return Ok(());
}
let (show_workers, show_jobs) = status_overview_section_flags(workers, jobs, ctx);
crate::status_display::render_full_status(
&status,
show_workers,
show_jobs,
convergence.as_ref(),
&remediation_hints,
&posture,
ctx.style(),
);
if ctx.is_verbose() {
debug!(target: "rch::verbose", "rendering verbose status details");
for line in render_status_verbose_detail_lines(&status, ctx.theme()) {
println!("{line}");
}
}
Ok(())
}
fn build_fleet_status_report(
status: &DaemonFullStatusResponse,
) -> rch_common::fleet_status::FleetStatusReport {
let desired = super::helpers::load_workers_from_config().unwrap_or_default();
build_fleet_status_report_with_workers(status, &desired)
}
fn build_fleet_status_report_with_workers(
status: &DaemonFullStatusResponse,
desired: &[rch_common::WorkerConfig],
) -> rch_common::fleet_status::FleetStatusReport {
use chrono::{DateTime, Utc};
use rch_common::fleet_diff::WorkerObservation;
use rch_common::fleet_status::{
DEFAULT_ABSENCE_THRESHOLD_SECS, FleetWorkerSignal, compute_fleet_status,
};
use std::collections::BTreeSet;
let desired_ids: BTreeSet<String> = desired.iter().map(|w| w.id.to_string()).collect();
let live_ids: BTreeSet<String> = status.workers.iter().map(|w| w.id.clone()).collect();
let now = Utc::now();
let absent_secs_for = |id: &str| -> Option<u64> {
status
.alerts
.iter()
.filter(|a| a.worker_id.as_deref() == Some(id) && a.kind.contains("offline"))
.filter_map(|a| DateTime::parse_from_rfc3339(&a.first_seen).ok())
.map(|fs| {
u64::try_from((now - fs.with_timezone(&Utc)).num_seconds().max(0)).unwrap_or(0)
})
.max()
};
let mut signals: Vec<FleetWorkerSignal> =
Vec::with_capacity(desired.len() + status.workers.len());
for w in &status.workers {
let reachable = w.status != "unreachable";
let admin_disabled = w.status == "disabled";
let temporarily_bypassed = w.bypass.is_some();
let observation = WorkerObservation {
worker_id: w.id.clone(),
configured: desired_ids.contains(&w.id),
in_daemon_pool: true,
reachable,
admin_disabled,
temporarily_bypassed,
facts_known: w.pressure_state.as_deref() != Some("telemetry_gap"),
command_admissible: true,
};
let disk_pressure = matches!(
w.pressure_state.as_deref(),
Some("critical") | Some("warning")
);
let slots_saturated = w.used_slots >= w.total_slots;
signals.push(FleetWorkerSignal {
observation,
disk_pressure,
slots_saturated,
absent_secs: absent_secs_for(&w.id),
});
}
for cfg in desired {
let id = cfg.id.to_string();
if live_ids.contains(&id) {
continue;
}
let observation = WorkerObservation {
worker_id: id.clone(),
configured: true,
in_daemon_pool: false,
reachable: false,
admin_disabled: false,
temporarily_bypassed: false,
facts_known: true,
command_admissible: true,
};
signals.push(FleetWorkerSignal {
observation,
disk_pressure: false,
slots_saturated: false,
absent_secs: absent_secs_for(&id),
});
}
compute_fleet_status(&signals, DEFAULT_ABSENCE_THRESHOLD_SECS)
}
fn build_remediation_view_from_status(
status: &DaemonFullStatusResponse,
) -> rch_common::remediation_view::RemediationView {
use rch_common::bypass_record::BypassState;
use rch_common::fleet_diff::WorkerObservation;
use rch_common::fleet_status::DEFAULT_ABSENCE_THRESHOLD_SECS;
use rch_common::remediation_view::{
DiskLevel, JobsInput, MAX_VIEW_INCIDENTS, ProofQueueInput, RemediationIncidentLine,
RemediationWorkerRow, assemble, build_inputs,
};
let rows: Vec<RemediationWorkerRow> = status
.workers
.iter()
.map(|w| {
let facts_known = w.pressure_state.as_deref() != Some("telemetry_gap");
RemediationWorkerRow {
observation: WorkerObservation {
worker_id: w.id.clone(),
configured: true,
in_daemon_pool: true,
reachable: w.status != "unreachable",
admin_disabled: w.status == "disabled",
temporarily_bypassed: w.bypass.is_some(),
facts_known,
command_admissible: true,
},
disk_level: DiskLevel::from_pressure_state(
w.pressure_state.as_deref().unwrap_or(""),
),
reclaiming: false,
free_ratio: w.pressure_disk_free_ratio,
slots_used: w.used_slots,
slots_total: w.total_slots,
telemetry_known: facts_known,
telemetry_fresh: w.pressure_telemetry_fresh.unwrap_or(false),
telemetry_age_secs: w.pressure_telemetry_age_secs,
recovered_pending_canary: w
.bypass
.as_ref()
.is_some_and(|b| b.state == BypassState::RecoveredPendingCanary),
absent_secs: None,
}
})
.collect();
let jobs = JobsInput {
active: status.active_builds.len(),
queued: status.queued_builds.len(),
stuck: 0,
};
let incidents: Vec<RemediationIncidentLine> = status
.alerts
.iter()
.filter(|a| a.state == "active")
.take(MAX_VIEW_INCIDENTS)
.map(|a| {
RemediationIncidentLine::new(
a.kind.clone(),
"alert",
a.worker_id.clone(),
0,
&a.message,
)
})
.collect();
let inputs = build_inputs(
&rows,
jobs,
ProofQueueInput::default(),
incidents,
DEFAULT_ABSENCE_THRESHOLD_SECS,
);
assemble(&inputs, 0)
}
fn status_overview_section_flags(workers: bool, jobs: bool, ctx: &OutputContext) -> (bool, bool) {
(workers || ctx.is_verbose(), jobs || ctx.is_verbose())
}
fn render_status_verbose_detail_lines(
status: &DaemonFullStatusResponse,
style: &Theme,
) -> Vec<String> {
let mut lines = vec![
style.format_header("Verbose Details"),
String::new(),
format!(
" {} {} {}",
style.key("Socket"),
style.muted(":"),
style.value(&status.daemon.socket_path)
),
format!(
" {} {} {}",
style.key("Started"),
style.muted(":"),
style.value(&status.daemon.started_at)
),
];
if !status.alerts.is_empty() {
lines.push(String::new());
lines.push(format!(" {}", style.key("Active Alerts:")));
for alert in &status.alerts {
let severity_style = match alert.severity.as_str() {
"critical" | "error" => style.error(&alert.severity),
"warning" => style.warning(&alert.severity),
_ => style.info(&alert.severity),
};
lines.push(format!(
" {} [{}] {}",
severity_style,
style.muted(&alert.created_at),
style.value(&alert.message)
));
}
}
if !status.issues.is_empty() {
lines.push(String::new());
lines.push(format!(" {}", style.key("Known Issues:")));
for issue in &status.issues {
lines.push(format!(
" {} {} - {}",
style.warning("⚠"),
style.key(&issue.summary),
style.muted(issue.remediation.as_deref().unwrap_or(""))
));
}
}
lines
}
const CHECK_HOOK_NOT_INSTALLED_ISSUE: &str = "Claude Code hook not installed";
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
enum CheckIssueSeverity {
Info,
Warning,
Error,
}
fn check_issue_severity(severity: &str) -> CheckIssueSeverity {
if severity.eq_ignore_ascii_case("error") || severity.eq_ignore_ascii_case("critical") {
CheckIssueSeverity::Error
} else if severity.eq_ignore_ascii_case("info") {
CheckIssueSeverity::Info
} else {
CheckIssueSeverity::Warning
}
}
fn derive_check_outcome(
total_count: usize,
healthy_count: usize,
workers: &[WorkerStatusFromApi],
daemon_issues: &[IssueFromApi],
hook_installed: bool,
) -> (String, i32, Vec<String>) {
let mut ranked_issues: Vec<(CheckIssueSeverity, String)> = workers
.iter()
.filter(|worker| worker.status != "healthy")
.map(|worker| {
(
CheckIssueSeverity::Warning,
format!("Worker {} is {}", worker.id, worker.status),
)
})
.collect();
ranked_issues.extend(
daemon_issues
.iter()
.map(|issue| (check_issue_severity(&issue.severity), issue.summary.clone())),
);
let (mut status, mut exit_code) = if total_count == 0 {
ranked_issues.insert(
0,
(
CheckIssueSeverity::Error,
"No workers configured".to_string(),
),
);
("not_ready".to_string(), 2)
} else if healthy_count == total_count {
("ready".to_string(), 0)
} else if healthy_count > 0 {
if !workers.iter().any(|worker| worker.status != "healthy") {
let not_healthy = total_count.saturating_sub(healthy_count);
let worker_word = if not_healthy == 1 {
"worker"
} else {
"workers"
};
let verb = if not_healthy == 1 { "is" } else { "are" };
ranked_issues.push((CheckIssueSeverity::Warning, format!(
"{not_healthy} configured {worker_word} {verb} not healthy ({healthy_count}/{total_count} healthy)"
)));
}
("degraded".to_string(), 1)
} else {
let cause = if workers.len() == total_count
&& workers.iter().all(|worker| worker.status == "unreachable")
{
"All workers are unreachable"
} else {
"No healthy workers available"
};
ranked_issues.insert(0, (CheckIssueSeverity::Error, cause.to_string()));
("not_ready".to_string(), 2)
};
match daemon_issues
.iter()
.map(|issue| check_issue_severity(&issue.severity))
.max()
{
Some(CheckIssueSeverity::Error) => {
status = "not_ready".to_string();
exit_code = 2;
}
Some(CheckIssueSeverity::Warning) if status == "ready" => {
status = "degraded".to_string();
exit_code = 1;
}
_ => {}
}
if !hook_installed {
ranked_issues.push((
CheckIssueSeverity::Error,
CHECK_HOOK_NOT_INSTALLED_ISSUE.to_string(),
));
status = "not_ready".to_string();
exit_code = 2;
}
ranked_issues.sort_by_key(|(severity, _)| std::cmp::Reverse(*severity));
let issues = ranked_issues
.into_iter()
.map(|(_, summary)| summary)
.collect();
(status, exit_code, issues)
}
pub async fn check(ctx: &OutputContext) -> Result<()> {
#[derive(Debug, Clone, serde::Serialize, schemars::JsonSchema)]
struct CheckResponse {
status: String,
exit_code: i32,
daemon: Option<DaemonCheckInfo>,
workers: WorkersCheckInfo,
hook: HookCheckInfo,
issues: Vec<String>,
}
#[derive(Debug, Clone, serde::Serialize, schemars::JsonSchema)]
struct DaemonCheckInfo {
running: bool,
pid: Option<u32>,
uptime_secs: Option<u64>,
}
#[derive(Debug, Clone, serde::Serialize, schemars::JsonSchema)]
struct WorkersCheckInfo {
total: usize,
healthy: usize,
unhealthy: Vec<String>,
}
#[derive(Debug, Clone, serde::Serialize, schemars::JsonSchema)]
struct HookCheckInfo {
installed: bool,
}
let daemon_result = send_daemon_command("GET /status\n").await;
let hook_installed = {
use crate::agent::{AgentKind, HookStatus, check_hook_status};
matches!(
check_hook_status(AgentKind::ClaudeCode),
Ok(HookStatus::Installed)
)
};
let (status, exit_code, daemon_info, workers_info, issues) = match daemon_result {
Ok(response) => {
match extract_json_body(&response) {
Some(json) => {
match serde_json::from_str::<DaemonFullStatusResponse>(json) {
Ok(daemon_status) => {
let healthy_count = daemon_status.daemon.workers_healthy;
let total_count = daemon_status.daemon.workers_total;
let unhealthy: Vec<String> = daemon_status
.workers
.iter()
.filter(|w| w.status != "healthy")
.map(|w| w.id.clone())
.collect();
let daemon_info = DaemonCheckInfo {
running: true,
pid: Some(daemon_status.daemon.pid),
uptime_secs: Some(daemon_status.daemon.uptime_secs),
};
let workers_info = WorkersCheckInfo {
total: total_count,
healthy: healthy_count,
unhealthy: unhealthy.clone(),
};
let (status, exit_code, issues) = derive_check_outcome(
total_count,
healthy_count,
&daemon_status.workers,
&daemon_status.issues,
hook_installed,
);
(status, exit_code, daemon_info, workers_info, issues)
}
Err(_) => {
let daemon_info = DaemonCheckInfo {
running: true,
pid: None,
uptime_secs: None,
};
let workers_info = WorkersCheckInfo {
total: 0,
healthy: 0,
unhealthy: vec![],
};
(
"not_ready".to_string(),
2,
daemon_info,
workers_info,
vec!["Daemon returned invalid response".to_string()],
)
}
}
}
None => {
let daemon_info = DaemonCheckInfo {
running: true,
pid: None,
uptime_secs: None,
};
let workers_info = WorkersCheckInfo {
total: 0,
healthy: 0,
unhealthy: vec![],
};
(
"not_ready".to_string(),
2,
daemon_info,
workers_info,
vec!["Daemon returned invalid response format".to_string()],
)
}
}
}
Err(_) => {
let daemon_info = DaemonCheckInfo {
running: false,
pid: None,
uptime_secs: None,
};
let workers_info = WorkersCheckInfo {
total: 0,
healthy: 0,
unhealthy: vec![],
};
(
"not_ready".to_string(),
2,
daemon_info,
workers_info,
vec!["Daemon not running".to_string()],
)
}
};
let hook_info = HookCheckInfo {
installed: hook_installed,
};
let response = CheckResponse {
status: status.clone(),
exit_code,
daemon: Some(daemon_info),
workers: workers_info.clone(),
hook: hook_info.clone(),
issues: issues.clone(),
};
if ctx.is_json() {
ctx.json(&ApiResponse::ok("check", &response))?;
if exit_code != 0 {
std::process::exit(exit_code);
}
return Ok(());
}
let style = ctx.style();
match status.as_str() {
"ready" => {
println!(
"{} RCH is ready ({}/{} workers healthy)",
style.success("\u{2713}"),
workers_info.healthy,
workers_info.total
);
}
"degraded" => {
let issue = issues
.first()
.map(String::as_str)
.unwrap_or("some workers are not fully healthy");
println!("{} RCH degraded: {}", style.warning("\u{26A0}"), issue);
}
"not_ready" => {
let issue = issues
.first()
.map(|s| s.as_str())
.unwrap_or("unknown error");
println!("{} RCH not ready: {}", style.error("\u{2717}"), issue);
}
_ => {}
}
if ctx.is_verbose() {
println!();
println!("{}", style.format_header("Health Check Details"));
println!();
if let Some(ref d) = response.daemon {
let daemon_status = if d.running {
format!(
"{} running (pid {}, uptime {})",
style.success("\u{2713}"),
d.pid.unwrap_or(0),
humanize_duration(d.uptime_secs.unwrap_or(0))
)
} else {
format!("{} not running", style.error("\u{2717}"))
};
println!(
" {} {} {}",
style.key("Daemon"),
style.muted(":"),
daemon_status
);
}
let workers_status = if workers_info.total == 0 {
format!("{} no workers configured", style.warning("\u{26A0}"))
} else if workers_info.healthy == workers_info.total {
format!(
"{} all healthy ({}/{})",
style.success("\u{2713}"),
workers_info.healthy,
workers_info.total
)
} else {
format!(
"{} {}/{} healthy",
style.warning("\u{26A0}"),
workers_info.healthy,
workers_info.total
)
};
println!(
" {} {} {}",
style.key("Workers"),
style.muted(":"),
workers_status
);
let hook_status = if hook_info.installed {
format!("{} installed", style.success("\u{2713}"))
} else {
format!("{} not installed", style.warning("\u{26A0}"))
};
println!(
" {} {} {}",
style.key("Hook"),
style.muted(":"),
hook_status
);
if !issues.is_empty() {
println!();
println!(" {}", style.key("Issues:"));
for issue in &issues {
println!(" {} {}", style.warning("\u{26A0}"), issue);
}
}
println!();
println!(
" {} {} {}",
style.key("Overall"),
style.muted(":"),
match status.as_str() {
"ready" => style.success("RCH is ready").to_string(),
"degraded" => style.warning("RCH is degraded").to_string(),
_ => style.error("RCH is not ready").to_string(),
}
);
}
if exit_code != 0 {
std::process::exit(exit_code);
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ui::context::OutputConfig;
use crate::ui::writer::SharedOutputBuffer;
#[cfg(unix)]
#[tokio::test]
async fn admit_delivers_formats_and_propagates_real_output_failures() {
use crate::ui::context::{ColorChoice, OutputError, OutputFormat, OutputMode};
use crate::ui::writer::OutputWriter;
use std::os::unix::net::UnixStream;
let retained = tempfile::tempdir().unwrap().keep();
for format in [OutputFormat::Json, OutputFormat::Toon] {
for closed in [false, true] {
let path = retained.join(format!("{format:?}-{closed}.output"));
let stdout = if closed {
let (socket, peer) = UnixStream::pair().unwrap();
socket
.set_write_timeout(Some(std::time::Duration::from_secs(2)))
.unwrap();
socket.shutdown(std::net::Shutdown::Write).unwrap();
drop(peer);
OutputWriter::new(socket, false)
} else {
OutputWriter::new(std::fs::File::create_new(&path).unwrap(), false)
};
let ctx = OutputContext::with_writers(
OutputConfig {
force_mode: Some(OutputMode::Json),
color: ColorChoice::Never,
format,
..Default::default()
},
stdout,
OutputWriter::stderr(),
);
let result = admit("cargo build --release", false, Vec::new(), &ctx).await;
if closed {
let error = result.unwrap_err();
assert!(matches!(
error.downcast_ref::<OutputError>(),
Some(OutputError::Delivery(cause))
if cause.kind() == std::io::ErrorKind::BrokenPipe
));
assert!(!path.exists());
} else {
result.unwrap();
let bytes = std::fs::read(path).unwrap();
let report = if format == OutputFormat::Toon {
serde_json::from_str::<serde_json::Value>(
&toon_rust::toon_to_json(std::str::from_utf8(&bytes).unwrap()).unwrap(),
)
.unwrap()
} else {
serde_json::from_slice::<serde_json::Value>(&bytes).unwrap()
};
assert_eq!(report["api_version"], "1.0");
assert_eq!(report["command"], "admit");
assert_eq!(report["success"], true);
assert!(report["error"].is_null());
assert_eq!(report["data"]["command"], "cargo build --release");
assert_eq!(report["data"]["is_compilation"], true);
assert_eq!(report["data"]["family"], "cargo_build");
assert_eq!(report["data"]["base_recommendation"], "offload");
assert_eq!(report["data"]["required"]["needs_cargo"], true);
}
}
}
assert_eq!(std::fs::read_dir(retained).unwrap().count(), 2);
}
#[test]
fn local_fallback_issue_counts_recent_reasons_only() {
use rch_common::{
IncidentEvent, IncidentEventType, IncidentReasonCode, IncidentSource, SelectedMode,
};
let now = 1_800_000_000_000_u64;
let event = |code: IncidentReasonCode, reason: &str, at: u64| {
IncidentEvent::new(
IncidentEventType::Fallback,
code,
IncidentSource::Hook,
"proj",
"cargo build",
SelectedMode::Local,
true,
at,
)
.with_detail("reason", reason)
};
let hour = 60 * 60 * 1000;
let events = vec![
event(
IncidentReasonCode::LocalFallback,
"no admissible workers",
now - hour,
),
event(
IncidentReasonCode::LocalFallback,
"no admissible workers",
now - 2 * hour,
),
event(
IncidentReasonCode::LocalFallback,
"rch disabled (general.enabled=false)",
now - 3 * hour,
),
event(
IncidentReasonCode::LocalFallback,
"daemon unavailable",
now - 30 * hour,
),
event(
IncidentReasonCode::ProofRefusal,
"remote required",
now - hour,
),
];
let issue = local_fallback_issue_from(&events, now).expect("recent fallbacks => issue");
assert_eq!(issue.severity, "warning");
assert!(
issue.summary.starts_with("3 local fallback build(s)"),
"{}",
issue.summary
);
assert!(
issue
.summary
.contains("2× no admissible workers; 1× rch disabled"),
"{}",
issue.summary
);
assert!(!issue.summary.contains("daemon unavailable"));
assert!(local_fallback_issue_from(&events[3..], now).is_none());
}
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)
}
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": [],
"active_builds": [],
"recent_builds": [],
"issues": [{
"severity": "warning",
"summary": "worker pressure",
"remediation": "run rch workers probe --all"
}],
"alerts": [{
"id": "alert-1",
"kind": "worker",
"severity": "warning",
"message": "worker degraded",
"worker_id": "builder-1",
"created_at": "2026-01-01T00:00:30Z"
}],
"stats": {
"total_builds": 0,
"success_count": 0,
"failure_count": 0,
"remote_count": 0,
"local_count": 0,
"avg_duration_ms": 0
}
}))
.unwrap()
}
#[test]
fn local_build_status_hints_preserve_pid_and_daemon_failure_context() {
let observation = crate::local_builds::LocalBuildObservation {
builds: vec![crate::local_builds::LocalBuild {
pid: 4242,
comm: "cargo-rch-real".to_string(),
exe: None,
}],
scan_error: None,
state_error: Some("cache denied".to_string()),
transition: None,
};
let hints = local_build_status_hints(Some(&observation));
assert_eq!(hints.len(), 1);
assert!(hints[0].message.contains("pid=4242"));
assert!(hints[0].message.contains("cache denied"));
let error = status_error_with_local_builds(anyhow::anyhow!("socket absent"), &hints);
let api = error.downcast_ref::<rch_common::ApiError>().unwrap();
let json = serde_json::to_value(api).unwrap();
assert!(
json["context"]["local_build_warning"]
.as_str()
.unwrap()
.contains("pid=4242")
);
assert_eq!(json["details"], "socket absent");
let report =
status_view_with_local_builds(&serde_json::json!({"fleet": "down"}), &hints).unwrap();
assert_eq!(report["fleet"], "down");
assert!(
report["local_build_hints"][0]["message"]
.as_str()
.unwrap()
.contains("pid=4242")
);
assert!(local_build_status_hints(None).is_empty());
}
fn check_issue(severity: &str, summary: &str) -> IssueFromApi {
IssueFromApi {
severity: severity.to_string(),
summary: summary.to_string(),
remediation: None,
}
}
#[test]
fn diagnose_multi_worker_allow_set_honors_selected_member() {
let plan =
resolve_placement(|key| (key == "RCH_WORKER").then(|| "busy,available".to_string()));
let diagnostics = rch_common::SelectionDiagnostics {
required_runtime: RequiredRuntime::None,
estimated_cores: 2,
min_success_rate: 0.8,
fallback_min_success_rate: 0.5,
active_project_exclusion_count: 0,
workers: vec![rch_common::WorkerSelectionDiagnostic {
worker_id: rch_common::WorkerId::new("busy"),
status: "healthy".to_string(),
circuit_state: "closed".to_string(),
pressure_state: "healthy".to_string(),
pressure_reason_code: "pressure.healthy".to_string(),
success_rate: Some(1.0),
min_success_rate: 0.8,
fallback_min_success_rate: 0.5,
required_runtime: RequiredRuntime::None,
runtime_available: true,
available_slots: 0,
total_slots: 8,
estimated_cores: 2,
active_project_excluded: false,
final_decision: rch_common::WorkerSelectionDiagnosticDecision::Deny,
final_reason: "available slots 0 < 2".to_string(),
reason_codes: vec!["slots.insufficient".to_string()],
}],
};
let selection = Some(DiagnoseWorkerSelection {
estimated_cores: 2,
cargo_jobs: None,
worker: Some(rch_common::SelectedWorker {
id: rch_common::WorkerId::new("available"),
host: "worker.example".to_string(),
user: "builder".to_string(),
identity_file: "~/.ssh/id_rsa".to_string(),
slots_available: 6,
speed_score: 90.0,
declared_os: None,
}),
reason: rch_common::SelectionReason::Success,
diagnostics: Some(diagnostics),
});
let refined = refine_diagnose_placement(plan, &selection);
assert_eq!(
refined.requested_worker_outcome.status,
rch_common::RequestedWorkerStatus::Honored
);
assert_eq!(refined.effective_worker.as_deref(), Some("available"));
}
#[test]
fn test_status_verbose_includes_worker_details() {
let normal_ctx = make_context(OutputConfig::default());
assert_eq!(
status_overview_section_flags(false, false, &normal_ctx),
(false, false)
);
let verbose_ctx = make_context(OutputConfig {
verbose: true,
..Default::default()
});
assert_eq!(
status_overview_section_flags(false, false, &verbose_ctx),
(true, true)
);
let style = crate::ui::theme::Style::new(false, true, false);
let verbose_lines = render_status_verbose_detail_lines(&make_daemon_status(), &style);
assert_ne!(verbose_lines.first().map(String::as_str), Some(""));
let output = verbose_lines.join("\n");
assert!(output.contains("Verbose Details"));
assert!(output.contains("Socket"));
assert!(output.contains("/tmp/rch.sock"));
assert!(output.contains("Started"));
assert!(output.contains("Active Alerts"));
assert!(output.contains("Known Issues"));
}
#[test]
fn test_check_outcome_ready_requires_hook() {
let unhealthy = Vec::new();
let daemon_issues = Vec::new();
let (status, exit_code, issues) =
derive_check_outcome(2, 2, &unhealthy, &daemon_issues, false);
assert_eq!(status, "not_ready");
assert_eq!(exit_code, 2);
assert_eq!(
issues.first().map(String::as_str),
Some(CHECK_HOOK_NOT_INSTALLED_ISSUE)
);
}
#[test]
fn test_check_outcome_ready_when_hook_installed() {
let unhealthy = Vec::new();
let daemon_issues = Vec::new();
let (status, exit_code, issues) =
derive_check_outcome(2, 2, &unhealthy, &daemon_issues, true);
assert_eq!(status, "ready");
assert_eq!(exit_code, 0);
assert!(issues.is_empty());
}
#[test]
fn test_check_outcome_degraded_missing_hook_promotes_not_ready() {
let unhealthy = vec![mk_worker_status(
"builder-2",
"unreachable",
"closed",
4,
1.0,
)];
let daemon_issues = vec![check_issue("warning", "worker pressure")];
let (status, exit_code, issues) =
derive_check_outcome(2, 1, &unhealthy, &daemon_issues, false);
assert_eq!(status, "not_ready");
assert_eq!(exit_code, 2);
assert_eq!(
issues.first().map(String::as_str),
Some(CHECK_HOOK_NOT_INSTALLED_ISSUE)
);
assert!(issues.iter().any(|issue| issue == "worker pressure"));
}
#[test]
fn test_check_outcome_ready_with_daemon_warning_is_degraded() {
let unhealthy = Vec::new();
let daemon_issues = vec![check_issue("warning", "worker pressure")];
let (status, exit_code, issues) =
derive_check_outcome(2, 2, &unhealthy, &daemon_issues, true);
assert_eq!(status, "degraded");
assert_eq!(exit_code, 1);
assert_eq!(issues, vec!["worker pressure"]);
}
#[test]
fn test_check_outcome_daemon_error_is_not_ready() {
let unhealthy = Vec::new();
let daemon_issues = vec![check_issue("error", "daemon failed to clean up a build")];
let (status, exit_code, issues) =
derive_check_outcome(2, 2, &unhealthy, &daemon_issues, true);
assert_eq!(status, "not_ready");
assert_eq!(exit_code, 2);
assert_eq!(issues, vec!["daemon failed to clean up a build"]);
}
#[test]
fn check_critical_cause_precedes_disabled_worker_warning() {
let daemon_issues = vec![
check_issue("warning", "builder-2 administratively disabled"),
check_issue("critical", "builder-1 disk pressure"),
];
let (status, exit_code, issues) = derive_check_outcome(
2,
1,
&[mk_worker_status("builder-2", "disabled", "closed", 0, 1.0)],
&daemon_issues,
true,
);
assert_eq!(status, "not_ready");
assert_eq!(exit_code, 2);
assert_eq!(issues.first(), Some(&daemon_issues[1].summary));
assert!(!issues.iter().any(|issue| issue.contains("unreachable")));
}
#[test]
fn check_disabled_fleet_does_not_claim_connectivity_failure() {
let daemon_issues = vec![check_issue(
"warning",
"builder-1 administratively disabled",
)];
let (status, exit_code, issues) = derive_check_outcome(
1,
0,
&[mk_worker_status("builder-1", "disabled", "closed", 0, 1.0)],
&daemon_issues,
true,
);
assert_eq!(status, "not_ready");
assert_eq!(exit_code, 2);
assert!(issues.contains(&daemon_issues[0].summary));
assert!(!issues.iter().any(|issue| issue.contains("unreachable")));
}
#[test]
fn test_check_outcome_zero_workers_preserves_daemon_issues() {
let unhealthy = Vec::new();
let daemon_issues = vec![check_issue("warning", "stale pressure telemetry")];
let (status, exit_code, issues) =
derive_check_outcome(0, 0, &unhealthy, &daemon_issues, true);
assert_eq!(status, "not_ready");
assert_eq!(exit_code, 2);
assert_eq!(
issues.first().map(String::as_str),
Some("No workers configured")
);
assert!(
issues
.iter()
.any(|issue| issue == "stale pressure telemetry")
);
}
#[test]
fn test_check_outcome_partial_health_without_named_unhealthy_worker_explains_degraded() {
let unhealthy = Vec::new();
let daemon_issues = Vec::new();
let (status, exit_code, issues) =
derive_check_outcome(3, 2, &unhealthy, &daemon_issues, true);
assert_eq!(status, "degraded");
assert_eq!(exit_code, 1);
assert_eq!(
issues,
vec!["1 configured worker is not healthy (2/3 healthy)".to_string()]
);
}
#[test]
fn test_check_disabled_worker_preserves_state_without_claiming_unreachable() {
let workers = vec![
mk_worker_status("builder-1", "healthy", "closed", 4, 1.0),
mk_worker_status("builder-2", "disabled", "closed", 4, 1.0),
];
let (status, exit_code, issues) = derive_check_outcome(2, 1, &workers, &[], true);
assert_eq!((status.as_str(), exit_code), ("degraded", 1));
assert!(
issues
.iter()
.any(|issue| issue.contains("builder-2") && issue.contains("disabled"))
);
assert!(!issues.iter().any(|issue| issue.contains("unreachable")));
let (status, exit_code, issues) = derive_check_outcome(1, 0, &workers[1..], &[], true);
assert_eq!((status.as_str(), exit_code), ("not_ready", 2));
assert!(
issues
.iter()
.any(|issue| issue.contains("builder-2") && issue.contains("disabled"))
);
assert!(!issues.iter().any(|issue| issue.contains("unreachable")));
}
#[test]
fn test_check_highest_severity_cause_precedes_worker_and_daemon_warnings() {
let workers = vec![mk_worker_status("builder-2", "disabled", "closed", 4, 1.0)];
let daemon_issues = vec![
check_issue("info", "maintenance scheduled"),
check_issue("warning", "stale telemetry"),
check_issue("CRITICAL", "disk capacity exhausted"),
];
let (status, exit_code, issues) =
derive_check_outcome(2, 1, &workers, &daemon_issues, true);
assert_eq!((status.as_str(), exit_code), ("not_ready", 2));
assert_eq!(
issues.first().map(String::as_str),
Some("disk capacity exhausted")
);
assert!(issues.iter().any(|issue| issue.contains("disabled")));
assert!(issues.iter().any(|issue| issue == "stale telemetry"));
assert!(issues.iter().any(|issue| issue == "maintenance scheduled"));
}
#[test]
fn test_check_warning_precedes_info_and_missing_hook_precedes_warnings() {
let daemon_issues = vec![
check_issue("info", "maintenance scheduled"),
check_issue("warning", "worker pressure"),
];
let (status, exit_code, issues) = derive_check_outcome(1, 1, &[], &daemon_issues, true);
assert_eq!((status.as_str(), exit_code), ("degraded", 1));
assert_eq!(issues.first().map(String::as_str), Some("worker pressure"));
let (status, exit_code, issues) = derive_check_outcome(1, 1, &[], &daemon_issues, false);
assert_eq!((status.as_str(), exit_code), ("not_ready", 2));
assert_eq!(
issues.first().map(String::as_str),
Some(CHECK_HOOK_NOT_INSTALLED_ISSUE)
);
assert!(issues.iter().any(|issue| issue == "worker pressure"));
}
#[test]
fn test_check_unreachable_summary_requires_all_worker_states_to_be_unreachable() {
let workers = vec![mk_worker_status("builder-1", "unreachable", "open", 4, 1.0)];
let (status, exit_code, issues) = derive_check_outcome(1, 0, &workers, &[], true);
assert_eq!((status.as_str(), exit_code), ("not_ready", 2));
assert_eq!(
issues.first().map(String::as_str),
Some("All workers are unreachable")
);
let (_, _, issues) = derive_check_outcome(2, 0, &workers, &[], true);
assert!(
!issues
.iter()
.any(|issue| issue == "All workers are unreachable")
);
}
#[test]
fn bare_self_test_skips_admin_disabled_workers() {
use rch_common::bypass_record::{AdminDisableRecord, AdminDisableStore};
let worker = |id: &str| WorkerConfig {
id: rch_common::WorkerId::new(id),
host: format!("{id}.invalid"),
user: "u".to_string(),
identity_file: "/dev/null".to_string(),
total_slots: 2,
priority: 100,
tags: vec![],
tools: Vec::new(),
};
let workers = vec![worker("ovh-a"), worker("hz3"), worker("hz4")];
let dir = tempfile::tempdir().unwrap();
let mut store = AdminDisableStore::load(dir.path().join("admin_disables.json"));
assert_eq!(
default_self_test_worker(&workers, &store)
.unwrap()
.id
.as_str(),
"ovh-a"
);
store
.upsert(AdminDisableRecord {
worker_id: "ovh-a".into(),
reason: Some("maintenance".into()),
disabled_unix_ms: 1,
})
.unwrap();
assert_eq!(
default_self_test_worker(&workers, &store)
.unwrap()
.id
.as_str(),
"hz3"
);
for id in ["hz3", "hz4"] {
store
.upsert(AdminDisableRecord {
worker_id: id.into(),
reason: None,
disabled_unix_ms: 1,
})
.unwrap();
}
assert_eq!(
default_self_test_worker(&workers, &store)
.unwrap()
.id
.as_str(),
"ovh-a"
);
assert!(default_self_test_worker(&[], &store).is_none());
}
#[test]
fn test_self_test_verbose_shows_each_step() {
let style = crate::ui::theme::Style::new(false, true, false);
let passed: SelfTestResultRecordFromApi = serde_json::from_value(serde_json::json!({
"run_id": 1,
"worker_id": "builder-1",
"passed": true,
"local_hash": "abc",
"remote_hash": "abc",
"local_time_ms": 4000,
"remote_time_ms": 1000,
"error": null
}))
.unwrap();
let failed: SelfTestResultRecordFromApi = serde_json::from_value(serde_json::json!({
"run_id": 1,
"worker_id": "builder-2",
"passed": false,
"local_hash": null,
"remote_hash": null,
"local_time_ms": null,
"remote_time_ms": null,
"error": "this is a deliberately long worker self-test error message for verbose context"
}))
.unwrap();
let passed_output = render_self_test_result_verbose_lines(&passed, &style).join("\n");
assert!(passed_output.contains("4.0x speedup"));
assert!(passed_output.contains("hash match"));
let failed_output = render_self_test_result_verbose_lines(&failed, &style).join("\n");
assert!(failed_output.contains("error:"));
assert!(failed_output.contains("deliberately long"));
}
#[test]
fn build_smoke_inputs_maps_environment_and_flags() {
let i = build_smoke_inputs(true, true, false, false, None);
assert!(i.workers_configured);
assert!(i.remote_execution_available);
assert_eq!(i.mode, ProfileMode::Smoke);
assert!(!i.dry_run);
assert!(!build_smoke_inputs(false, true, false, false, None).remote_execution_available);
assert!(!build_smoke_inputs(true, false, false, false, None).remote_execution_available);
let i = build_smoke_inputs(true, true, true, true, Some("css".to_string()));
assert_eq!(i.mode, ProfileMode::Soak);
assert!(i.dry_run);
assert_eq!(i.selected_worker.as_deref(), Some("css"));
}
#[test]
fn smoke_planned_events_cover_every_scenario_with_reasons() {
let inputs = build_smoke_inputs(false, false, false, false, None);
let plan = plan_smoke_profile(&inputs);
let events = smoke_planned_events(&plan, "run-1", Some("css"));
assert_eq!(events.len(), SmokeScenario::ALL.len());
assert!(
events
.iter()
.all(|e| e.run_id == "run-1" && e.event == "planned" && e.bead_id == SMOKE_BEAD_ID)
);
let daemon = events
.iter()
.find(|e| e.scenario == "daemon_reachable")
.unwrap();
assert_eq!(daemon.worker_id, None);
assert_eq!(daemon.status, "run");
let canary = events
.iter()
.find(|e| e.scenario == "cargo_canary")
.unwrap();
assert_eq!(canary.worker_id.as_deref(), Some("css"));
assert_eq!(canary.status, "skip");
assert_eq!(canary.reason_code.as_deref(), Some("smoke_no_real_workers"));
let proof = events
.iter()
.find(|e| e.scenario == "proof_mode_refusal")
.unwrap();
assert_eq!(proof.status, "expect_refusal");
}
#[test]
fn smoke_planned_events_full_fleet_runs_per_worker_scenarios() {
let inputs = build_smoke_inputs(true, true, false, false, None);
let plan = plan_smoke_profile(&inputs);
let events = smoke_planned_events(&plan, "run-2", Some("hz1"));
let canary = events
.iter()
.find(|e| e.scenario == "cargo_canary")
.unwrap();
assert_eq!(canary.status, "run");
assert_eq!(canary.reason_code, None);
let proof = events
.iter()
.find(|e| e.scenario == "proof_mode_refusal")
.unwrap();
assert_eq!(proof.status, "skip");
assert_eq!(proof.reason_code.as_deref(), Some("smoke_remote_available"));
}
#[test]
fn smoke_fleet_consistency_pass_and_fail_paths() {
use rch_common::fleet_diff::WorkerObservation;
use rch_common::fleet_status::{
DEFAULT_ABSENCE_THRESHOLD_SECS, FleetWorkerSignal, compute_fleet_status,
};
let ready = |id: &str| FleetWorkerSignal {
observation: WorkerObservation {
worker_id: id.to_string(),
configured: true,
in_daemon_pool: true,
reachable: true,
admin_disabled: false,
temporarily_bypassed: false,
facts_known: true,
command_admissible: true,
},
disk_pressure: false,
slots_saturated: false,
absent_secs: None,
};
let healthy =
compute_fleet_status(&[ready("a"), ready("b")], DEFAULT_ABSENCE_THRESHOLD_SECS);
assert_eq!(smoke_fleet_consistency(&healthy), (true, None));
let mut down1 = ready("a");
down1.observation.reachable = false;
let mut down2 = ready("b");
down2.observation.reachable = false;
let collapsed = compute_fleet_status(&[down1, down2], DEFAULT_ABSENCE_THRESHOLD_SECS);
let (ok, reason) = smoke_fleet_consistency(&collapsed);
assert!(!ok);
assert_eq!(reason.as_deref(), Some("fleet_capacity_collapsed"));
let mut absent = ready("c");
absent.observation.reachable = false;
absent.observation.in_daemon_pool = false;
absent.absent_secs = Some(DEFAULT_ABSENCE_THRESHOLD_SECS + 10);
let drifted = compute_fleet_status(&[ready("a"), absent], DEFAULT_ABSENCE_THRESHOLD_SECS);
let (ok2, reason2) = smoke_fleet_consistency(&drifted);
assert!(!ok2);
assert_eq!(reason2.as_deref(), Some("fleet_workers_absent"));
}
#[test]
fn fleet_report_treats_zero_disk_capacity_as_saturated_until_recovery() {
use rch_common::fleet_status::FleetProblemClass;
let desired = vec![rch_common::WorkerConfig {
id: rch_common::WorkerId::new("disk-worker"),
..Default::default()
}];
let mut status = make_daemon_status();
let mut worker = mk_worker_status("disk-worker", "healthy", "closed", 0, 90.0);
worker.pressure_state = Some("healthy".to_string());
status.workers.push(worker);
for used in [0, 4] {
status.workers[0].used_slots = used;
let report = build_fleet_status_report_with_workers(&status, &desired);
assert_eq!(report.problem_class, FleetProblemClass::LocalOverload);
}
status.workers[0].total_slots = 5;
let report = build_fleet_status_report_with_workers(&status, &desired);
assert_eq!(report.problem_class, FleetProblemClass::Healthy);
}
fn self_test_result(
passed: bool,
local_hash: Option<&str>,
remote_hash: Option<&str>,
remote_time_ms: Option<u64>,
error: Option<&str>,
) -> SelfTestResultRecordFromApi {
SelfTestResultRecordFromApi {
run_id: 1,
worker_id: "hz1".to_string(),
passed,
local_hash: local_hash.map(str::to_string),
remote_hash: remote_hash.map(str::to_string),
local_time_ms: Some(10),
remote_time_ms,
error: error.map(str::to_string),
}
}
#[test]
fn smoke_canary_verdict_requires_successful_verification() {
let failed = self_test_result(
false,
Some("aaa"),
Some("bbb"),
Some(1500),
Some("unreadable marker"),
);
assert_eq!(
smoke_canary_verdict(&failed),
(false, Some("unreadable marker".to_string()))
);
let full_pass = self_test_result(true, Some("aaa"), Some("aaa"), Some(900), None);
assert_eq!(smoke_canary_verdict(&full_pass), (true, None));
}
#[test]
fn smoke_canary_verdict_fails_when_remote_build_never_ran() {
let connect_fail = self_test_result(false, None, None, None, Some("ssh connect refused"));
let (passed, reason) = smoke_canary_verdict(&connect_fail);
assert!(!passed);
assert_eq!(reason.as_deref(), Some("ssh connect refused"));
let no_error = self_test_result(false, None, None, None, None);
let (passed2, reason2) = smoke_canary_verdict(&no_error);
assert!(!passed2);
assert_eq!(reason2.as_deref(), Some("remote_build_failed"));
}
#[test]
fn smoke_artifact_verdict_passes_on_retrieval_regardless_of_byte_identity() {
let pass = self_test_result(true, Some("aaa"), Some("aaa"), Some(900), None);
assert_eq!(smoke_artifact_verdict(&pass), (true, None));
assert!(smoke_artifact_byte_identical(&pass));
let not_retrieved = self_test_result(false, Some("aaa"), None, Some(900), None);
let (p1, r1) = smoke_artifact_verdict(¬_retrieved);
assert!(!p1);
assert_eq!(r1.as_deref(), Some("artifact_not_retrieved"));
let drift = self_test_result(true, Some("aaa"), Some("bbb"), Some(900), None);
assert_eq!(smoke_artifact_verdict(&drift), (true, None));
assert!(!smoke_artifact_byte_identical(&drift));
let missing = self_test_result(true, None, None, Some(900), None);
assert!(!smoke_artifact_byte_identical(&missing));
}
#[test]
fn smoke_cancel_verdict_accepts_only_deterministic_not_found() {
assert_eq!(smoke_cancel_verdict(Some("error")), (true, None));
let (p, r) = smoke_cancel_verdict(Some("cancelled"));
assert!(!p);
assert_eq!(r.as_deref(), Some("unexpected_cancel_status:cancelled"));
let (p2, r2) = smoke_cancel_verdict(None);
assert!(!p2);
assert_eq!(r2.as_deref(), Some("cancel_response_malformed"));
}
fn mk_worker_status(
id: &str,
status: &str,
circuit: &str,
slots: u32,
speed: f64,
) -> WorkerStatusFromApi {
serde_json::from_value(serde_json::json!({
"id": id,
"host": "h",
"user": "u",
"status": status,
"circuit_state": circuit,
"used_slots": 0,
"total_slots": slots,
"speed_score": speed,
"last_error": null,
}))
.expect("worker status")
}
#[test]
fn admit_job_required_tools_merges_config_only_in_job_mode() {
let config_tools = vec![
"clang".to_string(),
" llvm-cov ".to_string(),
"".to_string(),
];
assert_eq!(
merge_job_required_tools(vec!["clang".to_string()], &config_tools, true),
vec!["clang".to_string(), "llvm-cov".to_string()]
);
assert_eq!(
merge_job_required_tools(vec!["clang".to_string()], &config_tools, false),
vec!["clang".to_string()]
);
}
#[test]
fn admit_human_capabilities_include_nix_go_node_and_os() {
use rch_common::admit_preflight::RequiredCapabilities;
let req = RequiredCapabilities {
needs_cargo: true,
needs_zig: true,
needs_bun: true,
needs_nix: true,
needs_go: true,
needs_node: true,
needs_os: Some("darwin".to_string()),
needs_targets: vec!["aarch64-apple-darwin".to_string()],
needs_toolchains: vec!["nightly".to_string()],
needs_tools: vec!["clang".to_string()],
};
assert_eq!(
format_admit_required_capabilities(&req),
vec![
"cargo",
"zig",
"bun",
"nix",
"go",
"node",
"os:darwin",
"target:aarch64-apple-darwin",
"toolchain:nightly",
"tool:clang",
]
);
}
}