//! PreToolUse hook implementation.
//!
//! Handles incoming hook requests from Claude Code, classifies commands,
//! and routes compilation commands to remote workers.
use crate::config::load_config;
use crate::error::{ArtifactRetrievalWarning, DaemonError, TransferError};
use crate::state::primitives::atomic_write;
use crate::status_types::format_bytes;
use crate::toolchain::{detect_toolchain, parse_channel_string};
use crate::transfer::{
SyncResult, TransferPipeline, WorkerPlatform, clean_overlay_include_patterns,
compute_project_hash_with_dependency_roots_and_policy, configure_clean_git_command,
default_bun_artifact_patterns, default_c_cpp_artifact_patterns, default_rust_artifact_patterns,
default_rust_test_artifact_patterns, default_zigbuild_artifact_patterns, project_id_from_path,
};
use crate::ui::console::RchConsole;
use anyhow::Context;
use rch_common::errors::catalog::ErrorCode;
use rch_common::job_identity::{DurableJobLease, JobIdentity, default_job_lease_directory};
use rch_common::repo_updater_contract::{
REPO_UPDATER_ALLOW_OVERRIDE_ENV, REPO_UPDATER_ALLOWED_HOSTS_ENV, REPO_UPDATER_ALLOWLIST_ENV,
REPO_UPDATER_AUTH_CREDENTIAL_ID_ENV, REPO_UPDATER_AUTH_EXPIRES_AT_MS_ENV,
REPO_UPDATER_AUTH_ISSUED_AT_MS_ENV, REPO_UPDATER_AUTH_MODE_ENV, REPO_UPDATER_AUTH_REVOKED_ENV,
REPO_UPDATER_AUTH_SCOPES_ENV, REPO_UPDATER_AUTH_SOURCE_ENV,
REPO_UPDATER_AUTH_VERIFIED_HOSTS_ENV, REPO_UPDATER_OVERRIDE_APPROVED_AT_MS_ENV,
REPO_UPDATER_OVERRIDE_AUDIT_EVENT_ID_ENV, REPO_UPDATER_OVERRIDE_JUSTIFICATION_ENV,
REPO_UPDATER_OVERRIDE_OPERATOR_ID_ENV, REPO_UPDATER_OVERRIDE_TICKET_REF_ENV,
REPO_UPDATER_REQUIRE_HOST_IDENTITY_ENV, REPO_UPDATER_REQUIRED_SCOPES_ENV,
REPO_UPDATER_ROTATION_MAX_AGE_SECS_ENV, REPO_UPDATER_TRUSTED_HOST_IDENTITIES_ENV,
RepoUpdaterAuthContext, RepoUpdaterAuthMode, RepoUpdaterCredentialSource,
RepoUpdaterOperatorOverride, RepoUpdaterTrustedHostIdentity, RepoUpdaterVerifiedHostIdentity,
};
use rch_common::{
BuildHeartbeatPhase, BuildHeartbeatRequest, Classification, ColorMode, CommandPriority,
CommandTimingBreakdown, CompilationKind, ControlState, DependencyClosurePlan, HookInput,
HookOutput, IncidentEvent, IncidentEventType, IncidentLedger, IncidentLedgerConfig,
IncidentReasonCode, IncidentSource, OutputVisibility, REPO_UPDATER_CANONICAL_PROJECTS_ROOT,
RepoUpdaterAdapterCommand, RepoUpdaterAdapterContract, RepoUpdaterAdapterRequest,
RepoUpdaterOutputFormat, RequestedWorkerFacts, RequestedWorkerOutcome, RequestedWorkerStatus,
RequiredRuntime, SelectedMode, SelectedWorker, SelectionDiagnostics, SelectionReason,
SelectionResponse, SelfHealingConfig, ToolchainInfo, TransferConfig, WorkerConfig, WorkerId,
build_dependency_closure_plan_with_policy, build_invocation, classify_command,
declined_compilation_due_to_structure, default_socket_path, evaluate_requested_worker, mock,
normalize_project_path_with_policy,
path_topology::PathTopologyPolicy,
redaction::{redact_path, redact_secrets},
ui::{
ArtifactSummary, CelebrationSummary, CompilationProgress, CompletionCelebration, Icons,
OutputContext, RchTheme, TransferProgress,
},
};
use rch_telemetry::protocol::{
PIGGYBACK_MARKER, TelemetrySource, TestRunRecord, WorkerTelemetry,
extract_piggybacked_telemetry,
};
use serde::{Deserialize, Serialize};
use std::collections::BTreeSet;
use std::fs::OpenOptions;
use std::io::{self, Read, Write};
use std::os::unix::fs::PermissionsExt;
use std::path::Component;
use std::path::{Path, PathBuf};
use std::process::{Output, Stdio};
use std::sync::{
Arc, Mutex,
atomic::{AtomicBool, AtomicU64, Ordering},
};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::net::UnixStream;
use tokio::process::Command;
use tokio::sync::oneshot;
use tokio::time::{sleep, timeout};
use tracing::{debug, info, warn};
use which::which;
#[cfg(all(feature = "rich-ui", unix))]
use rich_rust::renderables::Panel;
// ============================================================================
// Exit Code Constants
// ============================================================================
//
// Cargo test (and cargo build/check/clippy) use specific exit codes:
//
// - 0: Success (all tests passed, or build succeeded)
// - 1: Build/compilation error (couldn't compile tests or crate)
// - 101: Test failures (tests compiled and ran, but some failed)
// - 128+N: Process killed by signal N (e.g., 137 = SIGKILL, 143 = SIGTERM)
//
// For RCH, ALL non-zero exits should deny local re-execution because:
// 1. Exit 101: Tests failed remotely, re-running locally won't help
// 2. Exit 1: Build error would occur locally too
// 3. Exit 128+N: The termination cause is unknown; local reruns may also fail
//
// The only exception is toolchain failures (missing rust version), which
// should fall back to local in case the local machine has the toolchain.
/// Exit code for successful cargo command (tests passed, build succeeded).
#[allow(dead_code)]
const EXIT_SUCCESS: i32 = 0;
/// Exit code for build/compilation error.
const EXIT_BUILD_ERROR: i32 = 1;
/// Exit code for cargo test when tests ran but some failed.
#[allow(dead_code)] // Used in run_exec
const EXIT_TEST_FAILURES: i32 = 101;
/// Minimum exit code indicating the process was killed by a signal.
/// Exit code = 128 + signal number (e.g., 137 = 128 + 9 = SIGKILL).
#[allow(dead_code)] // Used in run_exec
const EXIT_SIGNAL_BASE: i32 = 128;
/// Process exit code returned when the remote compile SUCCEEDED but the build
/// artifacts could NOT be transferred back, leaving the local build incomplete
/// (no binary/lib where the agent expects one). From the caller's perspective the
/// build did not actually complete, so this must be a NON-zero, build-failure-class
/// code rather than the remote command's exit 0 — re-running locally is the right
/// recovery, exactly like the AGENTS.md "Build failed (remote compilation)" case.
/// Pairs with the `RCH-E309 BuildArtifactMissing` diagnostic on stderr.
const EXIT_ARTIFACT_TRANSFER_FAILED: i32 = 102;
/// Exit code for a fail-closed refusal that is *retryable* — the build did not run
/// because no worker could be assigned right now (all slots busy) or the daemon was
/// momentarily unavailable, NOT because the toolchain crashed or the build failed.
/// A distinct code lets wrapper scripts and agents back off and retry instead of
/// treating a transient refusal as a real failure (rch#31). Refusals that are
/// *permanent* for the current invocation (a non-compilation command, unreadable
/// config) keep [`EXIT_BUILD_ERROR`].
const EXIT_REMOTE_REQUIRED_REFUSED: i32 = 103;
const RCH_CARGO_WRAPPER_BYPASS_ENV: &str = "RCH_CARGO_WRAPPER_BYPASS";
const RCH_REQUIRE_REMOTE_ENV: &str = "RCH_REQUIRE_REMOTE";
/// Opt-out knob for remote target-dir REUSE. When set to a truthy value the hook
/// falls back to the legacy unique-per-job remote target dir name
/// (`remote_cargo_target_dir_name`) instead of the stable pooled name, for users
/// who hit problems with the shared pool.
const RCH_DISABLE_TARGET_REUSE_ENV: &str = "RCH_DISABLE_TARGET_REUSE";
/// Validated source-tree policy for an explicit clean-overlay `rch exec` run.
///
/// The commit is resolved to an object ID before worker selection, and every
/// overlay path is normalized relative to the Git toplevel. The transfer layer
/// can therefore materialize an immutable base and upload only these paths.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct CleanOverlaySpec {
base_commit: String,
tree_object: String,
overlay_paths: Vec<PathBuf>,
overlay_fingerprint: String,
dependencies: Vec<(PathBuf, CleanOverlaySpec)>,
primary_directory: Option<PathBuf>,
}
impl CleanOverlaySpec {
pub(super) fn base_commit(&self) -> &str {
&self.base_commit
}
pub(super) fn overlay_paths(&self) -> &[PathBuf] {
&self.overlay_paths
}
pub(super) fn is_base_only(&self) -> bool {
self.overlay_paths.is_empty()
}
pub(super) fn overlay_fingerprint(&self) -> &str {
&self.overlay_fingerprint
}
/// Receipt emitted only after the worker has successfully executed this
/// immutable base plus verified overlay.
pub(super) fn execution_receipt(&self) -> String {
let mut receipt = format!(
"[RCH] clean-overlay receipt: base={} overlay-fingerprint={} tree={}",
self.base_commit, self.overlay_fingerprint, self.tree_object
);
for (root, spec) in &self.dependencies {
receipt.push_str(&format!(
" dependency-root={} commit={} tree={} overlay-fingerprint={}",
root.display(),
spec.base_commit,
spec.tree_object,
spec.overlay_fingerprint
));
}
receipt
}
pub(super) async fn verify_archive_attributes(
&self,
project_root: &Path,
) -> anyhow::Result<()> {
validate_clean_overlay_archive_attributes(project_root, &self.base_commit).await
}
pub(super) fn verify_overlay_unchanged(&self, project_root: &Path) -> anyhow::Result<()> {
let current = clean_overlay_fingerprint(project_root, &self.overlay_paths)?;
if current != self.overlay_fingerprint {
anyhow::bail!(
"clean-overlay input changed after admission (expected {}, found {}); refusing remote execution",
self.overlay_fingerprint,
current
);
}
Ok(())
}
}
/// Return the daemon-history identity for one execution.
///
/// Ordinary syncs retain their project-wide exclusion because they share a
/// mutable remote root. A clean-overlay job instead owns a fresh remote root,
/// so its daemon identity includes the verified source inputs and a job nonce.
/// This permits two isolated snapshots of the same project to consume separate
/// worker slots without making default-mode syncs concurrent.
fn selection_project_for_execution(
project: &str,
clean_overlay: Option<&CleanOverlaySpec>,
job_nonce: uuid::Uuid,
) -> String {
let Some(spec) = clean_overlay else {
return project.to_string();
};
let mut hasher = blake3::Hasher::new();
hasher.update(b"rch-clean-overlay-selection-project-v1\0");
hasher.update(project.as_bytes());
hasher.update(b"\0");
hasher.update(spec.base_commit().as_bytes());
hasher.update(b"\0");
hasher.update(spec.overlay_fingerprint().as_bytes());
for (root, dependency) in &spec.dependencies {
hasher.update(b"\0dependency\0");
hasher.update(root.as_os_str().as_encoded_bytes());
hasher.update(b"\0");
hasher.update(dependency.base_commit.as_bytes());
}
hasher.update(b"\0");
hasher.update(job_nonce.as_bytes());
let suffix = hasher.finalize().to_hex();
format!("{project}::clean-overlay::{}", &suffix[..16])
}
static HOOK_MODE_PANIC_FAIL_OPEN: AtomicBool = AtomicBool::new(false);
static AUTOSTART_LOCK_SEQUENCE: AtomicU64 = AtomicU64::new(0);
use rch_common::util::mask_sensitive_command;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RemotePipelineFailurePolicy {
AllowLocalFallback,
FailClosedNoLocalFallback,
}
fn classify_remote_pipeline_failure(error: &anyhow::Error) -> RemotePipelineFailurePolicy {
if is_remote_execution_unconfirmed(error) || is_ssh_command_timeout_error(error) {
RemotePipelineFailurePolicy::FailClosedNoLocalFallback
} else {
RemotePipelineFailurePolicy::AllowLocalFallback
}
}
fn release_unconfirmed_error(worker_id: &WorkerId, build_id: u64) -> anyhow::Error {
anyhow::anyhow!(
"daemon release for build {build_id} on {worker_id} was not acknowledged; ownership remains uncertain and the command must not be replayed"
)
.context(crate::transfer::RemoteExecutionUnconfirmed)
}
fn is_remote_execution_unconfirmed(error: &anyhow::Error) -> bool {
// Anyhow downcasting also finds a typed context, preserving the original
// ownership/transport error beneath the no-replay classification.
error
.downcast_ref::<crate::transfer::RemoteExecutionUnconfirmed>()
.is_some()
}
fn is_ssh_command_timeout_error(error: &anyhow::Error) -> bool {
error.chain().any(|cause| {
let message = cause.to_string();
message.contains("SSH command timed out after")
|| message.contains("Command timed out after")
})
}
fn remote_pipeline_failure_summary(worker_id: &WorkerId, error: &anyhow::Error) -> String {
if is_remote_execution_unconfirmed(error) {
return format!(
"[RCH] remote {} completion unconfirmed; ownership retained for recovery (no local fallback)",
worker_id
);
}
format!(
"[RCH] remote {} failed [{}] SSH command timed out (no local fallback)",
worker_id,
ErrorCode::SshTimeout.code_string()
)
}
/// Issue #62: extract the typed SSH-timeout error when its post-timeout remote
/// process-group kill could NOT be verified. `None` for every other failure —
/// including timeouts whose cleanup was verified or had nothing to kill.
fn ssh_timeout_with_unverified_cleanup(
error: &anyhow::Error,
) -> Option<&crate::transfer::SshCommandTimedOut> {
error
.chain()
.find_map(|cause| cause.downcast_ref::<crate::transfer::SshCommandTimedOut>())
.filter(|timeout| timeout.cleanup == crate::transfer::RemoteTimeoutCleanup::Unverified)
}
/// Issue #62: after an E104 SSH-timeout failure whose remote cleanup could not
/// be verified, quarantine the worker at the daemon (the same
/// `POST /workers/{id}/disable` mechanism as the SIGILL CPU-fault path). The
/// orphaned remote cargo may still hold the project's Cargo build-directory
/// lock; without this flag the next job is immediately re-admitted onto the
/// held lock and blocks on `Blocking waiting for file lock on build directory`.
async fn quarantine_worker_on_unverified_timeout_cleanup(
error: &anyhow::Error,
socket_path: &str,
worker_id: &WorkerId,
reporter: &HookReporter,
) {
let Some(timeout_err) = ssh_timeout_with_unverified_cleanup(error) else {
return;
};
warn!(
"E104 timeout on {} left an unverified remote process group ({}); quarantining worker",
worker_id, timeout_err.detail
);
reporter.summary_critical(&format!(
"[RCH] worker {} flagged for quarantine [{}]: E104 cleanup unverified ({}) — possible orphan holding the project's Cargo target lock",
worker_id,
ErrorCode::SshTimeout.code_string(),
timeout_err.detail
));
// With evidence, rchd re-runs this build's kill probe and clears the
// quarantine once the group is verified dead (bd-g8m4g).
let reason = rch_common::orphan_quarantine::quarantine_reason(timeout_err.evidence.as_ref());
if let Err(error) = disable_worker_for_fault(socket_path, worker_id, &reason).await {
warn!(
"Failed to quarantine worker {} after unverified E104 cleanup: {}",
worker_id, error
);
}
}
/// Hook stdin cap, shared by the fast path and the full path.
const HOOK_INPUT_LIMIT: u64 = 10 * 1024 * 1024;
/// Input the fast path read before handing a request to the full path.
static PRE_READ_HOOK_INPUT: std::sync::OnceLock<String> = std::sync::OnceLock::new();
/// Every agent Bash command runs `rch` as a PreToolUse hook, and almost all
/// of them are not compilations. Answer that pass-through before the CLI,
/// logging, the update check and the async runtime are built: they cost
/// ~2ms on a quiet Linux dispatcher and ~8ms of CPU on macOS (framework
/// loading), against a 1ms budget (bd-1nhd).
///
/// Returns true when the request is fully answered (allow: empty stdout,
/// exit 0), exactly as [`process_hook`] would answer it. Anything else,
/// including a compilation hidden behind shell structure (which reports a
/// summary), falls back to the full path with the input already read.
pub fn try_fast_passthrough() -> bool {
use std::io::{IsTerminal, Read};
// Only a bare `rch` fed by a pipe is a hook request; `COMPLETE` selects
// shell-completion mode.
if std::env::args_os().len() != 1
|| std::env::var_os("COMPLETE").is_some()
|| std::io::stdin().is_terminal()
{
return false;
}
install_hook_mode_panic_handler();
let mut input = String::new();
if std::io::stdin()
.take(HOOK_INPUT_LIMIT)
.read_to_string(&mut input)
.is_err()
{
// The full path also allows on an unreadable stdin.
return true;
}
if fast_passthrough_allows(input.trim()) {
return true;
}
let _ = PRE_READ_HOOK_INPUT.set(input);
false
}
/// Whether [`process_hook`] would answer this input with a plain allow and
/// no other effect. Unparseable input goes to the full path, which logs it.
fn fast_passthrough_allows(input: &str) -> bool {
if input.is_empty() {
return true;
}
let Ok(hook_input) = serde_json::from_str::<HookInput>(input) else {
return false;
};
if hook_input.tool_name != "Bash" {
return true;
}
let command = &hook_input.tool_input.command;
!crate::cache::classify_hook_command(command, classify_command).is_compilation
&& declined_compilation_due_to_structure(command).is_none()
}
/// Run the hook, reading from stdin and writing to stdout.
///
/// **Fail-open contract**: this function MUST return `Ok(())` for every
/// non-fatal failure mode. The hook runs synchronously in the agent's
/// Bash invocation path; any error we propagate becomes a non-zero exit
/// from `rch`, which Claude Code interprets as "hook said deny". We
/// would rather silently allow the command than block on stdin EOF, a
/// flushing hiccup, or a serialization edge case.
///
/// The single legitimate Err return is one we cannot fix locally
/// (e.g., `init_logging` error before this is called). All I/O within
/// run_hook is degraded to a silent allow.
pub async fn run_hook() -> anyhow::Result<()> {
let mut stdout = io::stdout();
// Read input from stdin with a 10MB limit to prevent OOM.
// A truncated/closed pipe is treated as "no input" (fail-open).
// The fast path may already have consumed it.
let mut input = PRE_READ_HOOK_INPUT.get().cloned().unwrap_or_default();
if PRE_READ_HOOK_INPUT.get().is_none() {
use tokio::io::{AsyncReadExt, stdin};
if let Err(e) = stdin()
.take(HOOK_INPUT_LIMIT)
.read_to_string(&mut input)
.await
{
warn!(target: "rch::hook", error = %e, "stdin read failed; allowing command (fail-open)");
return Ok(());
}
}
let input = input.trim();
if input.is_empty() {
// No input - just allow
return Ok(());
}
// Parse the hook input
let hook_input: HookInput = match serde_json::from_str(input) {
Ok(hi) => hi,
Err(e) => {
warn!("Failed to parse hook input: {}", e);
// On parse error, allow the command (fail-open)
return Ok(());
}
};
// Process the hook request
let output = process_hook(hook_input).await;
// Write output:
// - Deny: write JSON to block the command
// - AllowWithModifiedCommand: write JSON to replace the command (transparent interception)
// - Allow: output nothing (empty stdout = allow unchanged)
//
// serde / writeln errors here would be near-impossible (we just
// built the value from typed Rust), but if they occur we log and
// fall open rather than non-zero-exit and block the agent's Bash.
match &output {
HookOutput::Deny(_) | HookOutput::AllowWithModifiedCommand(_) => {
// Only a rewrite needs the original tool input (to carry `timeout`,
// `run_in_background`, ... through `updatedInput`), so the second
// parse stays off the non-compilation hot path.
let original_tool_input = matches!(output, HookOutput::AllowWithModifiedCommand(_))
.then(|| serde_json::from_str::<serde_json::Value>(input).ok())
.flatten()
.and_then(|mut raw| raw.get_mut("tool_input").map(serde_json::Value::take));
match output.to_hook_json(original_tool_input.as_ref()) {
Ok(json) => {
if let Err(e) = writeln!(stdout, "{}", json) {
warn!(target: "rch::hook", error = %e, "stdout write failed; falling open");
return Ok(());
}
if let Err(e) = stdout.flush() {
// Explicit flush: io::stdout() is fully buffered when
// attached to a pipe (Claude Code reads via pipe).
// Without this flush, abnormal exit could lose the JSON.
warn!(target: "rch::hook", error = %e, "stdout flush failed; falling open");
return Ok(());
}
}
Err(e) => {
warn!(target: "rch::hook", error = %e, "JSON serialization failed; falling open");
return Ok(());
}
}
}
HookOutput::Allow(_) => {
// Empty stdout = allow command unchanged
}
}
Ok(())
}
/// Install a panic hook that suppresses panic output and exits 0 when
/// the process is invoked as a Claude Code hook. Without this, any
/// panic in classify / serde / cache propagates as a non-zero exit,
/// which Claude Code interprets as "deny" and BLOCKS the agent's Bash.
///
/// The trade-off: a real bug in the hook becomes silent. That's the
/// correct call here — a hook that crashes silently and lets the
/// command run is strictly better for the agent than a hook that
/// blocks every Bash command on a regression. Real-world bug reports
/// surface via the daemon-side error logs, not the hook stderr.
///
/// Call this BEFORE any code that could panic.
pub fn install_hook_mode_panic_handler() {
enable_hook_mode_panic_fail_open();
// Idempotent guard: only install once.
use std::sync::Once;
static ONCE: Once = Once::new();
ONCE.call_once(|| {
// Capture the original hook so non-hook-mode invocations
// (e.g. `rch exec`) keep their normal panic output.
let original = std::panic::take_hook();
std::panic::set_hook(Box::new(move |info| {
if hook_mode_panic_fail_open_enabled() {
// Hook mode: log to stderr quietly, then exit 0 so the
// agent's Bash command runs locally. Don't print the
// backtrace to stderr (Claude Code may surface it).
eprintln!("[rch] hook panicked; falling open. (set RUST_BACKTRACE=1 to see)");
if std::env::var("RUST_BACKTRACE")
.ok()
.as_deref()
.is_some_and(|v| v == "1" || v.eq_ignore_ascii_case("full"))
{
original(info);
}
std::process::exit(0);
} else {
// Non-hook invocation: original panic behavior.
original(info);
}
}));
});
}
fn enable_hook_mode_panic_fail_open() {
HOOK_MODE_PANIC_FAIL_OPEN.store(true, Ordering::Release);
}
fn hook_mode_panic_fail_open_enabled() -> bool {
HOOK_MODE_PANIC_FAIL_OPEN.load(Ordering::Acquire)
|| std::env::var("RCH_HOOK_MODE")
.ok()
.as_deref()
.is_some_and(|v| v == "1" || v.eq_ignore_ascii_case("true"))
}
/// Execute a compilation command on a remote worker.
///
/// This is called by `rch exec -- <command>` which is invoked after the hook
/// rewrites the original compilation command. This separation allows the hook
/// to return immediately (<50ms) while the actual compilation runs as a
/// normal command invocation.
/// Re-assemble an argv vector into a shell command string that preserves
/// word boundaries under `sh -c` re-parsing.
///
/// Using a plain `parts.join(" ")` is wrong whenever an argv entry contains
/// shell-meaningful characters (spaces, quotes, `$`, etc.): the outer shell
/// that dispatched us already stripped the original quoting, leaving such
/// bytes as literal content in a single argv entry. `sh -c` would then
/// re-split on those literals and silently corrupt the command.
///
/// `shell_words::join` re-quotes each entry so round-tripping through
/// `sh -c` is a no-op. Some callers pass `rch exec -- "<whole shell command>"`
/// as a single argv entry; split that shell command once before re-quoting so
/// `sh -c` sees `env VAR=... cargo ...` instead of one quoted command name.
fn join_exec_command(command_parts: &[String]) -> String {
let mut normalized_parts = normalize_exec_command_parts(command_parts);
// `shell_words::join` quotes `FOO=1` as a whole word, and a quoted
// assignment is no longer an assignment: `'FOO=1' cargo test` runs a
// program named `FOO=1` (exit 127) on every local fallback and every
// non-cargo remote run. An explicit `env` keeps the assignment bytes as
// real argv, the same rule the Cargo token parser applies.
if normalized_parts
.first()
.is_some_and(|part| is_shell_assignment(part))
{
normalized_parts.insert(0, "env".to_string());
}
shell_words::join(normalized_parts)
}
/// `NAME=value` where NAME is a valid POSIX shell variable name.
pub(crate) fn is_shell_assignment(token: &str) -> bool {
token.split_once('=').is_some_and(|(key, _)| {
!key.is_empty()
&& key.chars().enumerate().all(|(index, ch)| {
ch == '_' || ch.is_ascii_alphabetic() || index > 0 && ch.is_ascii_digit()
})
})
}
fn normalize_exec_command_parts(command_parts: &[String]) -> Vec<String> {
if command_parts.len() == 1 {
match shell_words::split(&command_parts[0]) {
Ok(parts) if parts.len() > 1 => return parts,
_ => {}
}
}
command_parts.to_vec()
}
/// `rch exec` guarantees artifact synchronization only for a direct Cargo
/// invocation. A shell wrapper hides the artifact destination: running `sh -c
/// 'cargo build'` remotely may succeed while leaving the caller's local binary
/// stale.
///
/// Reject that specific shape rather than falling back to the local shell. The
/// caller must invoke cargo directly, where the target-dir and artifact-sync
/// pipeline can prove which bytes came home.
fn shell_wrapped_cargo_command(command_parts: &[String]) -> bool {
let parts = normalize_exec_command_parts(command_parts);
let Some(shell) = parts.first() else {
return false;
};
let shell_name = Path::new(shell).file_name().and_then(|name| name.to_str());
if !matches!(shell_name, Some("sh" | "bash" | "dash" | "zsh")) {
return false;
}
let script = parts.iter().skip(1).enumerate().find_map(|(index, arg)| {
let short_option_with_command = arg.strip_prefix('-').is_some_and(|options| {
!options.starts_with('-')
&& options.chars().all(char::is_alphabetic)
&& options.contains('c')
});
if arg == "-c" || short_option_with_command {
parts.get(index + 2).map(String::as_str)
} else {
None
}
});
script.is_some_and(|script| classify_command(script).is_compilation)
}
fn env_flag_enabled(value: &str) -> bool {
matches!(
value.to_ascii_lowercase().as_str(),
"1" | "true" | "yes" | "on"
)
}
/// The rustup toolchain explicitly requested inside a delegated command: an
/// env-style `RUSTUP_TOOLCHAIN=<name>` assignment token (bare or behind
/// `env`) or a `cargo +<name>` selector (issue #63). Returns the first match.
fn rustup_toolchain_from_command_tokens(tokens: &[String]) -> Option<String> {
let mut prev_is_cargo = false;
for token in tokens {
let token = token.trim_matches(['\'', '"']);
if let Some(value) = token.strip_prefix("RUSTUP_TOOLCHAIN=") {
let value = value.trim_matches(['\'', '"']);
if !value.is_empty() {
return Some(value.to_string());
}
}
if prev_is_cargo
&& let Some(value) = token.strip_prefix('+')
&& !value.is_empty()
{
return Some(value.to_string());
}
prev_is_cargo = token == "cargo"
|| token
.rsplit_once('/')
.is_some_and(|(_, name)| name == "cargo");
}
None
}
fn exec_requires_remote() -> bool {
std::env::var(RCH_REQUIRE_REMOTE_ENV).is_ok_and(|value| env_flag_enabled(&value))
}
/// Box-role default for strict-remote (bd-wywsj): on a DISPATCHER box,
/// offloadable builds are fail-closed by default — env stays the
/// per-call override (presence of RCH_REQUIRE_REMOTE or
/// RCH_FORCE_REMOTE, even falsy, is an explicit decision that wins).
fn role_requires_remote(role: rch_common::BoxRole) -> bool {
role == rch_common::BoxRole::Dispatcher
&& std::env::var(RCH_REQUIRE_REMOTE_ENV).is_err()
&& std::env::var("RCH_FORCE_REMOTE").is_err()
}
fn requested_worker_outcome(
requested: &WorkerId,
diagnostics: Option<&SelectionDiagnostics>,
) -> RequestedWorkerOutcome {
let Some(diagnostics) = diagnostics else {
return RequestedWorkerOutcome::requested();
};
let Some(worker) = diagnostics
.workers
.iter()
.find(|worker| worker.worker_id == *requested)
else {
return evaluate_requested_worker(&RequestedWorkerFacts {
requested: Some(requested.to_string()),
exists: false,
..RequestedWorkerFacts::none()
});
};
evaluate_requested_worker(&RequestedWorkerFacts::from_diagnostic(
requested.to_string(),
worker,
))
}
/// A structured requested-worker refusal: the human summary line plus the stable
/// incident reason code, so both the stream summary (#31) and the durable
/// incident ledger (#35) draw from one selection decision.
struct RequestedWorkerRefusal {
summary: String,
reason_code: IncidentReasonCode,
}
fn requested_worker_refusal(
requested_workers: &[WorkerId],
response: &SelectionResponse,
) -> Option<RequestedWorkerRefusal> {
if requested_workers.is_empty() || response.worker.is_some() {
return None;
}
let outcomes = requested_workers
.iter()
.map(|worker| {
(
worker,
requested_worker_outcome(worker, response.diagnostics.as_ref()),
)
})
.collect::<Vec<_>>();
let preferred_status = match response.reason {
SelectionReason::AllWorkersBusy => Some(RequestedWorkerStatus::NoFreeSlots),
SelectionReason::NoWorkersWithRuntime(_) => Some(RequestedWorkerStatus::MissingRuntime),
SelectionReason::AllCircuitsOpen => Some(RequestedWorkerStatus::TemporarilyBypassed),
_ => None,
};
let selected = preferred_status
.and_then(|status| {
outcomes
.iter()
.find(|(_, outcome)| outcome.status == status)
})
.or_else(|| {
outcomes
.iter()
.find(|(_, outcome)| outcome.status.is_refusal())
});
let requested = requested_workers
.iter()
.map(WorkerId::as_str)
.collect::<Vec<_>>()
.join(",");
let (code, summary) = if let Some((_, outcome)) = selected {
let code = outcome.reason_code.as_deref().unwrap_or("RCH-I001");
let next_action = outcome
.next_action
.as_deref()
.unwrap_or("run `rch diagnose -- <command>` and request an admissible worker");
(
code.to_string(),
format!(
"[{code}] requested worker set [{requested}] refused ({}); {next_action}",
outcome.status.as_str()
),
)
} else {
(
"RCH-I001".to_string(),
format!(
"[RCH-I001] requested worker set [{requested}] refused ({}); run `rch diagnose -- <command>` and request an admissible worker",
response.reason
),
)
};
let reason_code =
IncidentReasonCode::from_code_str(&code).unwrap_or(IncidentReasonCode::NoAdmissibleWorkers);
Some(RequestedWorkerRefusal {
summary,
reason_code,
})
}
/// Test-only projection of [`requested_worker_refusal`] onto just its summary
/// line, keeping the stable-and-actionable summary contract under test while
/// production code consumes the full structured refusal (summary + reason code).
#[cfg(test)]
fn requested_worker_refusal_summary(
requested_workers: &[WorkerId],
response: &SelectionResponse,
) -> Option<String> {
requested_worker_refusal(requested_workers, response).map(|refusal| refusal.summary)
}
/// Build a durable incident row for a requested-worker refusal (RCH-I0nn), so the
/// "your pin could not be honored" failure the allow-set fix exists for leaves a
/// postmortem trace even when the summary line is suppressed at stock visibility
/// (rch#35). Mirrors the shape of the socket-failure incidents already recorded.
fn build_requested_worker_refusal_incident(
reason_code: IncidentReasonCode,
requested_workers: &[WorkerId],
project: &str,
command_fingerprint: &str,
strict_remote: bool,
now_ms: u64,
) -> IncidentEvent {
let requested = requested_workers
.iter()
.map(WorkerId::as_str)
.collect::<Vec<_>>()
.join(",");
IncidentEvent::new(
IncidentEventType::Selection,
reason_code,
IncidentSource::Hook,
project,
command_fingerprint,
SelectedMode::Local,
!strict_remote,
now_ms,
)
.with_detail("requested_worker_set", requested)
.with_detail("refusal", "requested_worker")
.with_control(ControlState {
strict_remote_policy: strict_remote,
..ControlState::default()
})
}
fn selected_worker_is_requested(worker: &WorkerId, requested_workers: &[WorkerId]) -> bool {
requested_workers.is_empty() || requested_workers.contains(worker)
}
fn local_fallback_command(command: &str) -> std::process::Command {
let mut child = std::process::Command::new("sh");
child
.env(RCH_CARGO_WRAPPER_BYPASS_ENV, "1")
.arg("-c")
.arg(command);
child
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum LocalFallbackRefusal {
RemoteRequired,
}
fn local_fallback_command_for_policy(
command: &str,
require_remote: bool,
) -> Result<std::process::Command, LocalFallbackRefusal> {
if require_remote {
Err(LocalFallbackRefusal::RemoteRequired)
} else {
Ok(local_fallback_command(command))
}
}
/// Whether a fail-closed refusal for `reason` is retryable (transient capacity /
/// daemon unavailability) versus permanent for this invocation (a non-compilation
/// command, unreadable config). Retryable refusals get [`EXIT_REMOTE_REQUIRED_REFUSED`]
/// so wrappers can back off; permanent ones keep [`EXIT_BUILD_ERROR`] (rch#31).
fn remote_required_refusal_is_retryable(reason: &str) -> bool {
!matches!(reason, "non-compilation command" | "config unavailable")
&& !ConfigLocalPolicy::is_policy_reason(reason)
}
fn remote_required_refusal_summary(reason: &str) -> String {
if reason == "non-compilation command" {
format!(
"[RCH] remote required; refusing local fallback [{}] ({reason})",
ErrorCode::BuildUnknownCommand.code_string()
)
} else if remote_required_refusal_is_retryable(reason) {
// Typed, retryable marker so agents react to a signal instead of encoding
// folklore ("empty rc=1 means slot busy") — the message is now always
// emitted (summary_critical) and the exit code is distinct.
format!("[RCH] remote required; refusing local fallback ({reason}) — retryable")
} else {
format!("[RCH] remote required; refusing local fallback ({reason})")
}
}
/// A config-driven reason to run locally instead of offloading.
///
/// Issue #55: `general.enabled`, `general.force_local` and
/// `execution.allowlist` used to be consulted only by the Claude Code hook.
/// `rch exec` ignored them, and the cargo shim always execs `rch exec`, so on
/// a shim-only box (Codex, scripts, CI) `force_local = true` did nothing and on
/// a hook box the shim offloaded a command the hook had just allowed locally.
/// One policy, evaluated identically by every interceptor.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ConfigLocalPolicy {
/// `general.enabled = false`.
Disabled,
/// `general.force_local = true`.
ForceLocal,
/// Both force flags set — validation rejects this; fail safe (local).
ConflictingForceFlags,
/// The classified command's base is not in `execution.allowlist`.
NotAllowlisted(&'static str),
}
impl ConfigLocalPolicy {
const DISABLED_REASON: &'static str = "rch disabled (general.enabled=false)";
const FORCE_LOCAL_REASON: &'static str = "force_local";
const CONFLICT_REASON: &'static str = "invalid config: force_local+force_remote";
const NOT_ALLOWLISTED_SUFFIX: &'static str = "not in execution.allowlist";
/// Human-readable reason, in the `[RCH] local (<reason>)` vocabulary.
pub(crate) fn reason(self) -> String {
match self {
Self::Disabled => Self::DISABLED_REASON.to_string(),
Self::ForceLocal => Self::FORCE_LOCAL_REASON.to_string(),
Self::ConflictingForceFlags => Self::CONFLICT_REASON.to_string(),
Self::NotAllowlisted(base) => {
format!("command '{base}' {}", Self::NOT_ALLOWLISTED_SUFFIX)
}
}
}
/// Whether this policy is an explicit operator instruction that outranks a
/// `RCH_REQUIRE_REMOTE=1` baked into a generic interceptor (the shim).
///
/// `RCH_REQUIRE_REMOTE` exists to stop *silent* local fallback. A
/// configured, announced `force_local` is neither silent nor a fallback.
/// The allowlist is different: it is a capability gate, and under a strict
/// remote policy a non-allowlisted command is refused rather than run.
pub(crate) fn overrides_require_remote(self) -> bool {
!matches!(self, Self::NotAllowlisted(_))
}
/// Whether `reason` was produced by [`Self::reason`]. Policy refusals are
/// permanent for the invocation, never retryable.
fn is_policy_reason(reason: &str) -> bool {
matches!(
reason,
Self::DISABLED_REASON | Self::FORCE_LOCAL_REASON | Self::CONFLICT_REASON
) || (reason.starts_with("command '") && reason.ends_with(Self::NOT_ALLOWLISTED_SUFFIX))
}
}
/// Evaluate the config-driven local policy for a command of `kind`.
///
/// Returns `None` when config permits offload. Explicit job admission
/// (`rch exec --job`) bypasses the allowlist — there is no allowlist entry for
/// an arbitrary job — but still honors `enabled` / `force_local`.
pub(crate) fn config_local_policy(
config: &rch_common::RchConfig,
kind: Option<CompilationKind>,
) -> Option<ConfigLocalPolicy> {
if !config.general.enabled {
return Some(ConfigLocalPolicy::Disabled);
}
if config.general.force_local && config.general.force_remote {
return Some(ConfigLocalPolicy::ConflictingForceFlags);
}
if config.general.force_local {
return Some(ConfigLocalPolicy::ForceLocal);
}
match kind {
Some(CompilationKind::Job) | None => None,
Some(kind) => {
let base = kind.command_base();
(!config.execution.is_allowed(base)).then_some(ConfigLocalPolicy::NotAllowlisted(base))
}
}
}
fn exit_with_local_fallback(
command: &str,
reporter: &HookReporter,
reason: &str,
require_remote: bool,
) -> ! {
exit_with_local_fallback_recording(command, reporter, reason, require_remote, None)
}
/// [`exit_with_local_fallback`] that also records the underlying error chain
/// in the incident, so a refusal on a dispatcher says WHY remote execution
/// failed instead of only "remote execution failed".
fn exit_with_local_fallback_recording(
command: &str,
reporter: &HookReporter,
reason: &str,
require_remote: bool,
error: Option<&str>,
) -> ! {
// bd-uoh4x: exactly ONE envelope per invocation, emitted at the
// TERMINAL path only. A pre-run emit here would double-emit on the
// completed path (once before the child runs, again below with the
// real exit code) and lie on the refusal path (a `completed`
// envelope immediately followed by `refused`). Observed live via
// job-mode local fallback during bd-uoh4x verification.
//
// bd-qawj7: every terminal local run or refusal leaves a durable record
// of WHY. Before this only the daemon-unavailable lane was recorded, so a
// dispatcher melting under local builds (trj, 2026-09-25) left nothing to
// say whether rch chose local or an agent bypassed it.
// A non-compilation command running locally is not a build falling back,
// so it must not inflate `rch status`'s local-fallback-build count. Under
// require-remote it is refused instead, and that refusal is recorded.
if reason != "non-compilation command" || require_remote {
record_hook_incident(&build_local_fallback_incident(
command,
reason,
require_remote,
now_unix_ms(),
error,
));
}
let mut child = match local_fallback_command_for_policy(command, require_remote) {
Ok(child) => child,
Err(LocalFallbackRefusal::RemoteRequired) => {
// The one line explaining a non-zero exit MUST reach the agent even at
// stock `output.visibility = "none"` — otherwise a fail-closed refusal
// is an indistinguishable empty rc=1 (rch#31). Route it through the
// always-on stderr channel, and give retryable refusals a distinct code.
reporter.summary_critical(&remote_required_refusal_summary(reason));
emit_exec_envelope(&ExecResultEnvelope {
api_version: "1.0",
command,
outcome: "refused",
location: "local",
fallback_reason: Some(reason),
worker_id: None,
remote_exit_code: None,
duration_ms: None,
timing: None,
result_dirs: None,
error_code: None,
});
if remote_required_refusal_is_retryable(reason) {
std::process::exit(EXIT_REMOTE_REQUIRED_REFUSED);
}
std::process::exit(EXIT_BUILD_ERROR);
}
};
match child.status() {
Ok(status) => {
// A signal death reports 128+N like a shell, not a generic 1 that
// would hide an OOM kill or interrupt from the caller.
let code = {
use std::os::unix::process::ExitStatusExt as _;
status
.code()
.or_else(|| status.signal().map(|signal| 128 + signal))
.unwrap_or(1)
};
emit_exec_envelope(&ExecResultEnvelope {
api_version: "1.0",
command,
outcome: "completed",
location: "local",
fallback_reason: Some(reason),
worker_id: None,
remote_exit_code: Some(code),
duration_ms: None,
timing: None,
result_dirs: None,
error_code: None,
});
std::process::exit(code)
}
Err(error) => {
reporter.summary(&format!("[RCH] local fallback failed: {error}"));
emit_exec_envelope(&ExecResultEnvelope {
api_version: "1.0",
command,
outcome: "transport_error",
location: "local",
fallback_reason: Some(reason),
worker_id: None,
remote_exit_code: None,
duration_ms: None,
timing: None,
result_dirs: None,
error_code: None,
});
std::process::exit(EXIT_BUILD_ERROR);
}
}
}
// ---------------------------------------------------------------------------
// FIX 1: remote-failure fallback must prefer a bigger worker, not the local
// orchestrator. On a worker-fault build failure `run_exec` retries on a
// higher-capacity worker (ranked from daemon telemetry) before ever touching
// local execution, and any terminal local fallback is gated by
// `compilation.allow_local_fallback` so the orchestrator is not flooded.
// ---------------------------------------------------------------------------
/// Env override for the maximum number of remote workers to try for one build
/// (the first attempt plus capacity-ranked retries). Clamped to at least 1.
const RCH_MAX_REMOTE_ATTEMPTS_ENV: &str = "RCH_MAX_REMOTE_ATTEMPTS";
/// Default cap on remote attempts per build (first worker + up to two retries).
const DEFAULT_MAX_REMOTE_ATTEMPTS: u32 = 3;
fn max_remote_attempts() -> u32 {
std::env::var(RCH_MAX_REMOTE_ATTEMPTS_ENV)
.ok()
.and_then(|raw| raw.trim().parse::<u32>().ok())
.map(|value| value.max(1))
.unwrap_or(DEFAULT_MAX_REMOTE_ATTEMPTS)
}
/// Issue #63(c): whether an exhausted remote admission must surface a typed
/// remote-required refusal instead of silently starting a local build.
///
/// An explicit `rch exec --job` invocation requested the remote job rail; a
/// silent local compile violates that boundary exactly like a strict-remote
/// violation (the same class the #61 fix closed), so job admission counts as
/// strict-remote once every remote option is exhausted. The refusal exits
/// with the retryable [`EXIT_REMOTE_REQUIRED_REFUSED`] code and a `refused`
/// envelope, so wrappers can back off and retry rather than misreading a
/// local build's exit status as the remote result.
fn strict_remote_for_exhausted_admission(require_remote: bool, job_admission: bool) -> bool {
require_remote || job_admission
}
/// Terminal local-fallback for a remote FAILURE, gated so a failed remote build
/// only runs on the local orchestrator when the operator permits it.
///
/// Fails closed (exit `EXIT_BUILD_ERROR`) when remote execution is required
/// (`require_remote` / proof mode / `RCH_REQUIRE_REMOTE`) OR when
/// `compilation.allow_local_fallback = false`. Otherwise runs the command
/// locally exactly like [`exit_with_local_fallback`]. Emits the user-facing
/// summary itself in every branch. Never returns.
fn exit_with_gated_local_fallback(
command: &str,
reporter: &HookReporter,
reason: &str,
require_remote: bool,
allow_local_fallback: bool,
error: Option<&str>,
) -> ! {
if require_remote {
// Canonical remote-required refusal (proof mode / RCH_REQUIRE_REMOTE).
exit_with_local_fallback_recording(command, reporter, reason, true, error);
}
if !allow_local_fallback {
warn!(
"Local fallback disabled (compilation.allow_local_fallback=false); refusing to run '{}' on the orchestrator after remote failure: {reason}",
mask_sensitive_command(command)
);
// Fail-closed refusal: the line explaining the non-zero exit must reach the
// agent even at stock visibility (rch#31).
reporter.summary_critical(&format!(
"[RCH] refusing local fallback (compilation.allow_local_fallback=false; {reason})"
));
std::process::exit(EXIT_BUILD_ERROR);
}
reporter.summary(&format!("[RCH] local ({reason})"));
exit_with_local_fallback_recording(command, reporter, reason, false, error)
}
/// A remote build failed for a *worker-fault* reason (OOM/signal-kill, missing
/// worker system dependency, missing toolchain, or a generic pipeline error):
/// retry it on a different, higher-capacity worker before the terminal action.
struct RetryableRemoteFault {
/// Human-readable reason for logs (why this worker failed).
log_reason: String,
/// The worker that just failed (already released).
failed_worker: WorkerId,
/// What to do once every eligible worker is exhausted.
on_exhaust: RemoteFaultExhaustAction,
}
/// Terminal action taken once capacity-aware retries are exhausted.
enum RemoteFaultExhaustAction {
/// Surface the remote failure by exiting with this code. Used for OOM /
/// signal kills: a crate that OOMs every worker must NOT then be run on (and
/// likely OOM) the orchestrator, so this never falls back to local.
ExitWithCode { code: i32, summary: String },
/// Exit with a code after emitting worker-system-dependency remediation.
ExitWithEnvRemediation {
code: i32,
summary: String,
remediation: String,
},
/// Fall back to local, gated by `allow_local_fallback` / `require_remote`.
/// `reason` stays stable (it is grouped in `rch status`); `error` carries
/// the underlying error chain into the incident ledger.
GatedLocalFallback {
reason: String,
error: Option<String>,
},
}
/// Fetch daemon status, rank the remaining workers by capacity, and re-query the
/// daemon pinned to each untried worker, biggest first, until one is admitted.
/// Returns the fresh selection response plus the chosen worker id, or `None`
/// when no untried worker can serve.
#[allow(clippy::too_many_arguments)]
async fn try_retry_on_bigger_worker(
socket_path: &str,
project: &str,
estimated_cores: u32,
remote_command: &str,
toolchain: Option<&ToolchainInfo>,
required_runtime: RequiredRuntime,
command_priority: CommandPriority,
tried_workers: &[WorkerId],
worker_pin: &[WorkerId],
local_wrapper_id: Option<&str>,
reporter: &HookReporter,
) -> Option<(SelectionResponse, WorkerId)> {
let status = match crate::status_display::query_daemon_full_status().await {
Ok(status) => status,
Err(e) => {
reporter.verbose(&format!(
"[RCH] retry: could not fetch worker status ({e}); no bigger worker to try"
));
return None;
}
};
let snapshots = build_capacity_snapshots(&status);
// The biggest candidate may be full right now. Walk down the capacity
// order rather than ending retries while smaller workers sit idle.
let mut passed_over = tried_workers.to_vec();
while let Some(chosen) = pick_bigger_worker(snapshots.as_slice(), &passed_over, worker_pin) {
let preferred = vec![chosen.clone()];
match query_daemon(
socket_path,
project,
estimated_cores,
remote_command,
toolchain,
required_runtime,
command_priority,
0,
Some(std::process::id()),
local_wrapper_id,
false, // do not block waiting on one specific worker during a retry
&preferred,
false, // retry upsizing is compilation-scoped; never job mode
&[], // ...and therefore carries no named-tool requirements
)
.await
{
Ok(response) if response.worker.is_some() => return Some((response, chosen)),
Ok(_) => {
reporter.verbose(&format!(
"[RCH] retry: worker {chosen} is not currently admissible; trying the next"
));
passed_over.push(chosen);
}
Err(e) => {
reporter.verbose(&format!(
"[RCH] retry: re-query for {chosen} failed ({e}); ending retries"
));
return None;
}
}
}
None
}
// ---------------------------------------------------------------------------
// Hook daemon-recovery: socket-failure classification, configured-vs-canonical
// socket mismatch detection, and durable structured-incident emission
// (bd-session-history-remediation-ocv9i.3.1).
//
// When the hook cannot reach the daemon it must record *why* — a missing,
// refused, or stale socket, or a configured-vs-canonical socket-path mismatch —
// as a durable structured incident, attempt a bounded daemon autostart and one
// selection retry, then either proceed remotely or fall back / refuse (proof
// mode) loudly. All of this lives on the slow recovery path; the fast
// non-compilation classification budget is never touched.
//
// The decision cores are pure so the six bead scenarios (refused / stale /
// wrong-configured socket, daemon start success / failure, proof-mode refusal)
// are unit-testable without spawning a daemon; the side effects (autostart,
// ledger append) are thin wrappers around them.
// ---------------------------------------------------------------------------
/// Why the hook could not reach the daemon over its Unix socket. Reported in
/// the `socket_failure` incident detail so postmortems can tell a never-started
/// daemon (`missing`) from a crashed one (`refused`/`stale`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SocketFailureKind {
/// The socket file does not exist (daemon never created it, or wrong path).
Missing,
/// The socket exists but refused the connection (no live listener).
Refused,
/// The socket exists and connected but the daemon did not respond in time.
Stale,
/// Any other daemon-query failure (protocol/read error, malformed response).
Other,
}
impl SocketFailureKind {
fn as_str(self) -> &'static str {
match self {
SocketFailureKind::Missing => "missing",
SocketFailureKind::Refused => "refused",
SocketFailureKind::Stale => "stale",
SocketFailureKind::Other => "other",
}
}
}
/// Classify a [`query_daemon`] failure for incident reporting. Pure: inspects
/// the error chain plus whether the socket file is present on disk.
fn classify_socket_failure(err: &anyhow::Error, socket_exists: bool) -> SocketFailureKind {
// Explicit daemon-side signals from query_daemon.
if let Some(daemon_err) = err.downcast_ref::<DaemonError>() {
match daemon_err {
DaemonError::SocketNotFound { .. } | DaemonError::NotRunning => {
return SocketFailureKind::Missing;
}
DaemonError::ConnectionFailed { .. } | DaemonError::SocketPermissionDenied { .. } => {
return SocketFailureKind::Refused;
}
_ => {}
}
}
// Raw std::io errors surfaced from UnixStream::connect (wrapped by `?`).
for cause in err.chain() {
if let Some(io_err) = cause.downcast_ref::<std::io::Error>() {
return match io_err.kind() {
std::io::ErrorKind::NotFound => SocketFailureKind::Missing,
std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::PermissionDenied => {
SocketFailureKind::Refused
}
std::io::ErrorKind::TimedOut => SocketFailureKind::Stale,
_ if socket_exists => SocketFailureKind::Stale,
_ => SocketFailureKind::Other,
};
}
}
// The 5s connect timeout is an anyhow string error with no io::Error source.
if err.to_string().contains("timed out") {
return SocketFailureKind::Stale;
}
if socket_exists {
SocketFailureKind::Stale
} else {
SocketFailureKind::Missing
}
}
/// A configured-vs-canonical socket-path disagreement. Like the daemon's
/// startup-consistency probe, the hook reports this drift but **never** rewrites
/// operator-owned config — detection and loud reporting only.
#[derive(Debug, Clone, PartialEq, Eq)]
struct SocketPathMismatch {
configured: String,
canonical: String,
/// Whether the canonical default socket exists (a live daemon there is the
/// likely reason the hook missed it on the configured path).
canonical_exists: bool,
}
/// Lexical socket-path equivalence (trims surrounding whitespace; exact match
/// otherwise — mirrors the daemon startup-consistency `PathBuf` comparison).
fn socket_paths_equivalent(a: &str, b: &str) -> bool {
a.trim() == b.trim()
}
/// Detect a "wrong configured socket" condition: the configured socket path
/// differs from the canonical default. Returns `None` when they agree. Pure —
/// the caller supplies `canonical` and `canonical_exists` so this is testable
/// without filesystem or environment access.
fn detect_socket_path_mismatch(
configured: &str,
canonical: &str,
canonical_exists: bool,
) -> Option<SocketPathMismatch> {
if socket_paths_equivalent(configured, canonical) {
return None;
}
Some(SocketPathMismatch {
configured: configured.to_string(),
canonical: canonical.to_string(),
canonical_exists,
})
}
/// Terminal action after a daemon-socket failure plus a bounded autostart and
/// one selection retry. Pure so the bead's daemon-start-success / failure /
/// proof-mode-refusal scenarios are unit-testable without spawning a daemon.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DaemonRecoveryAction {
/// The daemon answered after autostart + retry — proceed remotely.
ProceedRemote,
/// Fail-open: run the command locally (records `LocalFallback`, RCH-I011).
LocalFallback,
/// Proof mode: refuse local fallback (records `ProofRefusal`, RCH-I012) and
/// exit fail-closed.
Refuse,
}
/// Decide the terminal action. `retry_succeeded` is whether the post-autostart
/// selection retry produced a usable daemon response. A successful retry always
/// wins; otherwise proof mode refuses and convenience mode falls back.
fn decide_recovery_action(retry_succeeded: bool, strict_remote: bool) -> DaemonRecoveryAction {
if retry_succeeded {
DaemonRecoveryAction::ProceedRemote
} else if strict_remote {
DaemonRecoveryAction::Refuse
} else {
DaemonRecoveryAction::LocalFallback
}
}
/// Current wall-clock time as Unix epoch milliseconds. The hook is a real
/// process, so wall-clock is appropriate here (unlike the clock-free pure
/// rch-common modules); a pre-epoch clock — impossible in practice — yields 0.
fn now_unix_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX))
.unwrap_or(0)
}
/// Write one crash-safe lease file for the lifetime of an `rch exec` wrapper.
/// The file is intentionally retained after completion so a later recovery
/// scan can distinguish an acknowledged terminal command from an uncertain
/// dead wrapper.
#[derive(Clone)]
pub(crate) struct DurableLeaseWriter {
path: PathBuf,
lease: Arc<Mutex<DurableJobLease>>,
}
impl DurableLeaseWriter {
pub(crate) fn load(wrapper_id: &str) -> anyhow::Result<Self> {
anyhow::ensure!(
wrapper_id.starts_with("rchw-") && uuid::Uuid::parse_str(&wrapper_id[5..]).is_ok(),
"invalid wrapper identity"
);
let path = durable_lease_path(wrapper_id);
let lease: DurableJobLease = serde_json::from_slice(&std::fs::read(&path)?)?;
anyhow::ensure!(
lease.schema_version == 1 && lease.identity.local_wrapper_id == wrapper_id,
"durable lease identity or schema mismatch"
);
Ok(Self {
path,
lease: Arc::new(Mutex::new(lease)),
})
}
pub(crate) fn snapshot(&self) -> DurableJobLease {
self.lease.lock().unwrap_or_else(|p| p.into_inner()).clone()
}
pub(crate) fn set_recovery(&self, recovery: serde_json::Value) -> anyhow::Result<()> {
self.lease
.lock()
.unwrap_or_else(|p| p.into_inner())
.recovery = Some(recovery);
self.persist()
}
pub(crate) fn record_exit(&self, exit_code: i32) -> anyhow::Result<()> {
self.lease
.lock()
.unwrap_or_else(|p| p.into_inner())
.exit_code = Some(exit_code);
self.persist()
}
fn create(
command: &str,
strict_remote: bool,
self_healing_enabled: bool,
) -> anyhow::Result<Self> {
let identity = JobIdentity::new_local();
let path = durable_lease_path(&identity.local_wrapper_id);
let command_fingerprint = format!("blake3:{}", blake3::hash(command.as_bytes()).to_hex());
let lease = DurableJobLease::new(
identity,
std::process::id(),
current_process_start_ticks(),
current_boot_id(),
now_unix_ms(),
strict_remote,
self_healing_enabled,
command_fingerprint,
);
let writer = Self {
path,
lease: Arc::new(Mutex::new(lease)),
};
writer.persist()?;
Ok(writer)
}
fn wrapper_id(&self) -> String {
self.lease
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.identity
.local_wrapper_id
.clone()
}
fn admit(&self, remote_build_id: u64, worker_id: &WorkerId) -> anyhow::Result<()> {
{
let mut lease = self
.lease
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if let Some(recovery) = lease.recovery.as_ref() {
anyhow::ensure!(
recovery.get("retired").and_then(serde_json::Value::as_bool) == Some(true),
"previous remote source ownership is unresolved; recover this wrapper before another admission"
);
}
lease.recovery = None;
lease.exit_code = None;
lease.terminal_acknowledged = false;
lease.admit(
remote_build_id,
worker_id.as_str().to_string(),
now_unix_ms(),
);
}
self.persist()
}
fn ensure_released_for_retry(&self) -> anyhow::Result<()> {
let lease = self.snapshot();
if let Some(recovery) = lease.recovery.as_ref() {
anyhow::ensure!(
recovery.get("retired").and_then(serde_json::Value::as_bool) == Some(true),
"remote source ownership is unresolved; run rch jobs recover {} before retrying",
lease.identity.local_wrapper_id
);
}
Ok(())
}
pub(crate) fn heartbeat(&self, phase: &str) -> anyhow::Result<()> {
{
let mut lease = self
.lease
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
lease.heartbeat(phase, now_unix_ms());
}
self.persist()
}
/// A terminal acknowledgement requires observed delivery completion: the
/// exact command result is recorded, no validated retrieval recipe is
/// outstanding (or it reports fully-returned outputs), and no later
/// heartbeats are expected. Refuses to fake completion otherwise.
pub(crate) fn acknowledge_terminal(&self) -> anyhow::Result<()> {
{
let mut lease = self
.lease
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if let Some(recovery) = lease.recovery.as_ref()
&& (recovery
.get("returned")
.and_then(serde_json::Value::as_i64)
.is_none()
|| recovery.get("retired").and_then(serde_json::Value::as_bool) != Some(true))
{
anyhow::bail!(
"durable retrieval or source release evidence is incomplete; terminal acknowledgement refused (state stays recoverable)"
);
}
lease.acknowledge_terminal(now_unix_ms());
}
self.persist()
}
fn persist(&self) -> anyhow::Result<()> {
self.persist_snapshot(atomic_write)
}
fn persist_snapshot(
&self,
publish: impl FnOnce(&Path, &[u8]) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
// Keep publication inside the same mutex as snapshot serialization.
// Otherwise a heartbeat can write its older Preparing snapshot after
// execution admission was durably recorded, enabling unsafe recovery.
let lease = self
.lease
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let bytes = serde_json::to_vec_pretty(&*lease)?;
publish(&self.path, &bytes)
}
}
/// Only failures known to precede queued ownership may reach autorecovery or
/// local fallback. The typed marker survives anyhow context layers.
fn selection_error_for_recovery(
error: anyhow::Error,
lease: &DurableLeaseWriter,
) -> anyhow::Result<anyhow::Error> {
if error
.downcast_ref::<daemon_ipc::SelectionOutcomeUnconfirmed>()
.is_none()
{
return Ok(error);
}
if let Err(persist_error) = lease.heartbeat("selection_unconfirmed") {
return Err(error.context(format!(
"selection remains unconfirmed; could not persist lease phase: {persist_error}"
)));
}
Err(error.context(format!(
"selection remains unconfirmed for {}; lease retained for inspection or same-identity cancellation; command will not be replayed",
lease.wrapper_id()
)))
}
fn durable_lease_path(local_wrapper_id: &str) -> PathBuf {
default_job_lease_directory().join(format!("{local_wrapper_id}.json"))
}
fn current_process_start_ticks() -> Option<u64> {
let stat = std::fs::read_to_string("/proc/self/stat").ok()?;
let after_comm = stat.rsplit_once(") ")?.1;
after_comm.split_whitespace().nth(19)?.parse().ok()
}
fn current_boot_id() -> Option<String> {
std::fs::read_to_string("/proc/sys/kernel/random/boot_id")
.ok()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
}
/// Build the structured incident for a daemon-socket failure (RCH-I010). The
/// `selected_mode` is recorded as `Local` because at detection time the build
/// has not yet been steered; the terminal local-fallback / proof-refusal
/// incident records the final disposition.
fn build_socket_failure_incident(
kind: SocketFailureKind,
mismatch: Option<&SocketPathMismatch>,
project: &str,
command_fingerprint: &str,
strict_remote: bool,
now_ms: u64,
) -> IncidentEvent {
let mut event = IncidentEvent::new(
IncidentEventType::Selection,
IncidentReasonCode::DaemonSocketRefused,
IncidentSource::Hook,
project,
command_fingerprint,
SelectedMode::Local,
!strict_remote,
now_ms,
)
.with_detail("socket_failure", kind.as_str())
.with_control(ControlState {
strict_remote_policy: strict_remote,
..ControlState::default()
});
if let Some(m) = mismatch {
event = event
.with_detail("socket_path_mismatch", "true")
.with_detail("configured_socket", redact_path(&m.configured))
.with_detail("canonical_socket", redact_path(&m.canonical))
.with_detail("canonical_socket_exists", m.canonical_exists.to_string());
}
event
}
/// Build the terminal incident after autostart + retry could not restore the
/// daemon: `ProofRefusal` (RCH-I012) when proof mode forbids local fallback,
/// else `LocalFallback` (RCH-I011).
fn build_recovery_terminal_incident(
strict_remote: bool,
project: &str,
command_fingerprint: &str,
detail_reason: &str,
now_ms: u64,
) -> IncidentEvent {
let (reason_code, event_type) = if strict_remote {
(IncidentReasonCode::ProofRefusal, IncidentEventType::Proof)
} else {
(
IncidentReasonCode::LocalFallback,
IncidentEventType::Fallback,
)
};
IncidentEvent::new(
event_type,
reason_code,
IncidentSource::Hook,
project,
command_fingerprint,
SelectedMode::Local,
!strict_remote,
now_ms,
)
.with_detail("reason", detail_reason.to_string())
.with_control(ControlState {
strict_remote_policy: strict_remote,
..ControlState::default()
})
}
/// The incident recorded by every terminal local fallback: `LocalFallback`
/// (RCH-I011) when the command runs locally, `ProofRefusal` (RCH-I012) when a
/// remote-required policy refuses it. The fallback reason goes in `details`;
/// the command is recorded only as a secret-redacted fingerprint.
fn build_local_fallback_incident(
command: &str,
reason: &str,
refused: bool,
now_ms: u64,
error: Option<&str>,
) -> IncidentEvent {
let event = build_recovery_terminal_incident(
refused,
&extract_project_name(),
&redact_secrets(command),
reason,
now_ms,
);
match error {
Some(error) => event.with_detail("error", bounded_incident_error(error)),
None => event,
}
}
/// Secret-redacted, length-bounded error chain for an incident record.
fn bounded_incident_error(error: &str) -> String {
const MAX_CHARS: usize = 600;
let redacted = redact_secrets(error);
match redacted.char_indices().nth(MAX_CHARS) {
Some((cut, _)) => format!("{}…", &redacted[..cut]),
None => redacted,
}
}
/// The incident ledger as `[remediation.incident_ledger]` configures it
/// (path and retention), or the defaults when config cannot load.
pub(crate) fn configured_incident_ledger() -> IncidentLedger {
let config = crate::config::load_config()
.map(|config| IncidentLedgerConfig::from(&config.remediation.incident_ledger))
.unwrap_or_default();
IncidentLedger::new(config)
}
/// Append `event` to the durable incident ledger, best-effort. Incident logging
/// must never break a build, so a write failure is logged and swallowed. A
/// tracing breadcrumb is always emitted so the incident is visible even when the
/// ledger write fails. The ledger lives off the hot path, so the append cost
/// (one buffered line) does not affect the classification budgets.
fn record_hook_incident(event: &IncidentEvent) {
// A local fallback (RCH-I011) is already announced by the
// `[RCH] local (<reason>)` summary and a refusal (RCH-I012) by its critical
// `remote required; refusing local fallback` line. Every one of them is now
// recorded, so a WARN here would add multi-line stderr noise to each.
if matches!(
event.reason_code,
IncidentReasonCode::LocalFallback | IncidentReasonCode::ProofRefusal
) {
debug!(
target: "rch::hook::incident",
reason_code = %event.reason_code,
selected_mode = ?event.selected_mode,
"hook incident recorded",
);
} else {
warn!(
target: "rch::hook::incident",
reason_code = %event.reason_code,
failure_class = event.reason_code.failure_class(),
selected_mode = ?event.selected_mode,
local_fallback_allowed = event.local_fallback_allowed,
"hook incident recorded",
);
}
if let Err(e) = configured_incident_ledger().append(event) {
// Best-effort by contract. An unwritable state dir (sandbox, unset
// HOME) would otherwise print this on every fallback and refusal.
debug!(
target: "rch::hook::incident",
error = %e,
"failed to append incident to ledger (continuing)",
);
}
}
pub(crate) fn normalize_repository_relative_path(
label: &str,
path: &Path,
) -> anyhow::Result<PathBuf> {
let mut normalized = PathBuf::new();
for component in path.components() {
match component {
Component::CurDir => {}
Component::Normal(part) => normalized.push(part),
Component::ParentDir | Component::RootDir | Component::Prefix(_) => {
anyhow::bail!(
"{label} must be repository-relative without '..': {}",
path.display()
);
}
}
}
if normalized.as_os_str().is_empty() {
anyhow::bail!("{label} must not be empty or the repository root");
}
let normalized_label = normalized
.to_str()
.ok_or_else(|| anyhow::anyhow!("{label} must be valid UTF-8: {}", path.display()))?;
if !normalized_label.is_ascii() {
anyhow::bail!(
"{label} must contain only ASCII characters to avoid cross-platform filesystem aliases: {}",
path.display()
);
}
if normalized_label.contains('\\') || normalized_label.chars().any(char::is_control) {
anyhow::bail!(
"{label} may not contain backslashes or control characters: {}",
path.display()
);
}
Ok(normalized)
}
/// Normalize a clean-overlay overlay path (shared rules with result dirs).
fn normalize_clean_overlay_path(path: &Path) -> anyhow::Result<PathBuf> {
normalize_repository_relative_path("clean-overlay path", path)
}
/// Normalize and dedupe declared job result directories (`rch exec --job
/// --result-dir`, bd-p0yoo). Same repository-relative safety rules as
/// clean-overlay paths: no traversal, no absolute paths, ASCII-only, no
/// backslashes or control characters — these strings are interpolated into
/// remote rsync source specs, so an adversarial value must not be able to
/// escape the synced project tree.
fn validate_job_result_dirs(dirs: Vec<PathBuf>) -> anyhow::Result<Vec<PathBuf>> {
let mut seen = std::collections::BTreeSet::new();
let mut validated = Vec::with_capacity(dirs.len());
for dir in dirs {
let normalized = normalize_repository_relative_path("--result-dir", &dir)?;
if seen.insert(normalized.clone()) {
validated.push(normalized);
}
}
Ok(validated)
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct CleanBaseTreeEntry {
mode: u32,
object_id: String,
path: PathBuf,
}
fn parse_clean_base_tree(output: &[u8]) -> anyhow::Result<Vec<CleanBaseTreeEntry>> {
let mut entries = Vec::new();
for record in output
.split(|byte| *byte == 0)
.filter(|record| !record.is_empty())
{
let Some(tab) = record.iter().position(|byte| *byte == b'\t') else {
anyhow::bail!("clean-overlay base tree contains a malformed ls-tree record");
};
let metadata = std::str::from_utf8(&record[..tab])
.context("clean-overlay base tree metadata is not UTF-8")?;
let mut fields = metadata.split_ascii_whitespace();
let mode = fields
.next()
.ok_or_else(|| anyhow::anyhow!("clean-overlay base tree record has no mode"))?;
let object_type = fields
.next()
.ok_or_else(|| anyhow::anyhow!("clean-overlay base tree record has no object type"))?;
let object_id = fields
.next()
.ok_or_else(|| anyhow::anyhow!("clean-overlay base tree record has no object ID"))?;
if fields.next().is_some() {
anyhow::bail!("clean-overlay base tree contains unexpected metadata fields");
}
if object_id.is_empty() || !object_id.bytes().all(|byte| byte.is_ascii_hexdigit()) {
anyhow::bail!("clean-overlay base tree contains an invalid object ID");
}
let mode = u32::from_str_radix(mode, 8).context("parse clean-overlay Git tree mode")?;
if object_type != "blob" && !(mode == 0o160000 && object_type == "commit") {
anyhow::bail!("clean-overlay base tree contains unsupported object type {object_type}");
}
let path = std::str::from_utf8(&record[tab + 1..])
.context("clean-overlay base contains a non-UTF-8 Git path")?;
entries.push(CleanBaseTreeEntry {
mode,
object_id: object_id.to_owned(),
path: PathBuf::from(path),
});
}
Ok(entries)
}
fn clean_overlay_path_has_exact_spelling(
project_root: &Path,
relative: &Path,
) -> anyhow::Result<bool> {
let mut parent = project_root.to_path_buf();
for component in relative.components() {
let requested = component.as_os_str();
let mut found = false;
for entry in std::fs::read_dir(&parent)
.with_context(|| format!("read clean-overlay path parent {}", parent.display()))?
{
if entry?.file_name() == requested {
found = true;
break;
}
}
if !found {
return Ok(false);
}
parent.push(requested);
}
Ok(true)
}
fn validate_clean_overlay_live_path(project_root: &Path, relative: &Path) -> anyhow::Result<()> {
if !clean_overlay_path_has_exact_spelling(project_root, relative)? {
anyhow::bail!(
"clean-overlay path spelling does not exactly match the filesystem entry: {}",
relative.display()
);
}
let mut absolute = project_root.to_path_buf();
for component in relative.components() {
absolute.push(component.as_os_str());
let metadata = std::fs::symlink_metadata(&absolute).with_context(|| {
format!(
"clean-overlay path does not exist and deletions are unsupported: {}",
relative.display()
)
})?;
if metadata.file_type().is_symlink() {
anyhow::bail!(
"clean-overlay overlays do not support symlinks: {}",
relative.display()
);
}
}
Ok(())
}
fn validate_clean_overlay_against_base(
project_root: &Path,
overlay_paths: &[PathBuf],
base_entries: &[CleanBaseTreeEntry],
) -> anyhow::Result<()> {
for overlay in overlay_paths {
let overlay_label = overlay
.to_str()
.ok_or_else(|| anyhow::anyhow!("clean-overlay path must be valid UTF-8"))?;
let overlay_components = overlay_label.split('/').collect::<Vec<_>>();
for entry in base_entries {
let base_label = entry
.path
.to_str()
.ok_or_else(|| anyhow::anyhow!("clean-overlay base path must be valid UTF-8"))?;
let base_components = base_label.split('/').collect::<Vec<_>>();
for (base, selected) in base_components.iter().zip(&overlay_components) {
if base == selected {
continue;
}
if base.eq_ignore_ascii_case(selected) {
anyhow::bail!(
"clean-overlay path {} aliases differently spelled base path {}; case-only overlays are unsupported",
overlay.display(),
entry.path.display()
);
}
break;
}
}
let overlay_prefix = format!("{overlay_label}/");
let metadata = std::fs::symlink_metadata(project_root.join(overlay))
.with_context(|| format!("inspect clean-overlay path {}", overlay.display()))?;
let exact = base_entries
.iter()
.find(|entry| entry.path == overlay.as_path());
let descendants = base_entries
.iter()
.filter(|entry| {
entry
.path
.to_str()
.is_some_and(|path| path.starts_with(&overlay_prefix))
})
.collect::<Vec<_>>();
if metadata.is_dir() {
if exact.is_some() {
anyhow::bail!(
"clean-overlay path changes a base file into a directory, which is unsupported: {}",
overlay.display()
);
}
for entry in descendants {
if !clean_overlay_path_has_exact_spelling(project_root, &entry.path)? {
anyhow::bail!(
"clean-overlay directory omits or renames base path {}; selected deletions are unsupported",
entry.path.display()
);
}
let local = std::fs::symlink_metadata(project_root.join(&entry.path))
.with_context(|| {
format!("inspect base path under overlay {}", entry.path.display())
})?;
let base_is_symlink = entry.mode == 0o120000;
if base_is_symlink != local.file_type().is_symlink()
|| (!base_is_symlink && !local.is_file())
{
anyhow::bail!(
"clean-overlay directory changes the type of base path {}, which is unsupported",
entry.path.display()
);
}
}
} else {
if !descendants.is_empty() {
anyhow::bail!(
"clean-overlay path changes a base directory into a file, which is unsupported: {}",
overlay.display()
);
}
if let Some(entry) = exact {
let base_is_symlink = entry.mode == 0o120000;
if base_is_symlink != metadata.file_type().is_symlink()
|| (!base_is_symlink && !metadata.is_file())
{
anyhow::bail!(
"clean-overlay path changes the type of a base path, which is unsupported: {}",
overlay.display()
);
}
}
}
}
Ok(())
}
fn hash_clean_overlay_path(
project_root: &Path,
relative: &Path,
hasher: &mut blake3::Hasher,
) -> anyhow::Result<()> {
let relative_label = relative.to_str().ok_or_else(|| {
anyhow::anyhow!(
"clean-overlay directory contains a non-UTF-8 path: {}",
relative.display()
)
})?;
if !relative_label.is_ascii() {
anyhow::bail!(
"clean-overlay paths must contain only ASCII characters to avoid cross-platform filesystem aliases: {}",
relative.display()
);
}
if relative_label.contains('\\') || relative_label.chars().any(char::is_control) {
anyhow::bail!(
"clean-overlay paths may not contain backslashes or control characters: {}",
relative.display()
);
}
if relative.components().any(|component| {
component
.as_os_str()
.to_str()
.is_some_and(|name| name.eq_ignore_ascii_case(".git"))
}) {
anyhow::bail!(
"clean-overlay paths may not contain Git metadata: {}",
relative.display()
);
}
let absolute = project_root.join(relative);
let metadata = std::fs::symlink_metadata(&absolute)
.with_context(|| format!("inspect clean-overlay input {}", relative.display()))?;
hasher.update(&(relative_label.len() as u64).to_le_bytes());
hasher.update(relative_label.as_bytes());
hasher.update(&(metadata.permissions().mode() & 0o7777).to_le_bytes());
if metadata.file_type().is_symlink() {
anyhow::bail!(
"clean-overlay overlays do not support symlinks: {}",
relative.display()
);
} else if metadata.is_dir() {
hasher.update(b"directory\0");
let mut children = std::fs::read_dir(&absolute)
.with_context(|| format!("read clean-overlay directory {}", relative.display()))?
.map(|entry| entry.map(|entry| relative.join(entry.file_name())))
.collect::<Result<Vec<_>, _>>()?;
children.sort();
if children.is_empty() {
anyhow::bail!(
"clean-overlay overlays do not support empty directories: {}",
relative.display()
);
}
for child in children {
hash_clean_overlay_path(project_root, &child, hasher)?;
}
} else if metadata.is_file() {
hasher.update(b"file\0");
hasher.update(&metadata.len().to_le_bytes());
let mut file = std::fs::File::open(&absolute)
.with_context(|| format!("open clean-overlay input {}", relative.display()))?;
let mut buffer = [0_u8; 64 * 1024];
loop {
let read = file
.read(&mut buffer)
.with_context(|| format!("read clean-overlay input {}", relative.display()))?;
if read == 0 {
break;
}
hasher.update(&buffer[..read]);
}
} else {
anyhow::bail!(
"clean-overlay path has unsupported file type: {}",
relative.display()
);
}
Ok(())
}
fn clean_overlay_fingerprint(
project_root: &Path,
overlay_paths: &[PathBuf],
) -> anyhow::Result<String> {
let mut hasher = blake3::Hasher::new();
hasher.update(b"rch-clean-overlay-input-v1\0");
for path in overlay_paths {
validate_clean_overlay_live_path(project_root, path)?;
hash_clean_overlay_path(project_root, path, &mut hasher)?;
validate_clean_overlay_live_path(project_root, path)?;
}
Ok(hasher.finalize().to_hex().to_string())
}
async fn git_output_bytes(project_root: &Path, args: &[&str]) -> anyhow::Result<Vec<u8>> {
let mut command = Command::new("git");
configure_clean_git_command(&mut command);
let output = command
.current_dir(project_root)
.args(args)
.output()
.await
.with_context(|| format!("failed to run git in {}", project_root.display()))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
anyhow::bail!("git {} failed: {}", args.join(" "), stderr.trim());
}
Ok(output.stdout)
}
async fn git_output(project_root: &Path, args: &[&str]) -> anyhow::Result<String> {
Ok(
String::from_utf8(git_output_bytes(project_root, args).await?)
.context("git returned non-UTF-8 output")?
.trim()
.to_string(),
)
}
fn contains_git_archive_transform_attribute(contents: &str) -> bool {
contents.lines().any(|line| {
let line = line.trim();
if line.is_empty() || line.starts_with('#') {
return false;
}
line.split_ascii_whitespace().skip(1).any(|token| {
let name = token
.trim_start_matches(['-', '!'])
.split_once('=')
.map_or(token.trim_start_matches(['-', '!']), |(name, _)| name);
matches!(name, "export-ignore" | "export-subst")
})
})
}
async fn validate_clean_overlay_archive_attributes(
project_root: &Path,
commit: &str,
) -> anyhow::Result<()> {
let paths = git_output_bytes(
project_root,
&["ls-tree", "-rz", "--full-tree", "--name-only", commit],
)
.await?;
for path in paths
.split(|byte| *byte == 0)
.filter(|path| !path.is_empty())
{
let path = std::str::from_utf8(path)
.context("clean-overlay base contains a non-UTF-8 Git path")?;
if path != ".gitattributes" && !path.ends_with("/.gitattributes") {
continue;
}
let contents = git_show_optional(project_root, commit, path)
.await?
.ok_or_else(|| {
anyhow::anyhow!(
"failed to read known .gitattributes path {path} from clean-overlay base {commit}"
)
})?;
if contains_git_archive_transform_attribute(&contents) {
anyhow::bail!(
"clean-overlay base uses export-ignore/export-subst in {path}; refusing a non-checkout-equivalent Git archive"
);
}
}
let info_attributes = git_output(
project_root,
&["rev-parse", "--git-path", "info/attributes"],
)
.await?;
let info_attributes = if Path::new(&info_attributes).is_absolute() {
PathBuf::from(info_attributes)
} else {
project_root.join(info_attributes)
};
match std::fs::read_to_string(&info_attributes) {
Ok(contents) if contains_git_archive_transform_attribute(&contents) => {
anyhow::bail!(
"clean-overlay repository info/attributes uses export-ignore/export-subst; refusing a non-checkout-equivalent Git archive"
);
}
Ok(_) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => {
return Err(error).with_context(|| {
format!(
"read clean-overlay repository attributes {}",
info_attributes.display()
)
});
}
}
Ok(())
}
async fn git_show_optional(
project_root: &Path,
commit: &str,
relative: &str,
) -> anyhow::Result<Option<String>> {
let object = format!("{commit}:{relative}");
let mut command = Command::new("git");
configure_clean_git_command(&mut command);
let output = command
.current_dir(project_root)
.args(["show", "--no-ext-diff", &object])
.output()
.await
.with_context(|| format!("read {relative} from clean-overlay base {commit}"))?;
if !output.status.success() {
return Ok(None);
}
Ok(Some(String::from_utf8(output.stdout).with_context(
|| format!("{relative} in clean-overlay base is not UTF-8"),
)?))
}
fn is_selected_cargo_metadata(path: &Path) -> bool {
path.file_name().is_some_and(|name| name == "Cargo.toml")
|| (path
.parent()
.and_then(Path::file_name)
.is_some_and(|name| name == ".cargo")
&& path
.file_name()
.is_some_and(|name| name == "config" || name == "config.toml"))
}
/// Inventory only the immutable base and explicitly selected overlays. Never
/// discover dependency manifests by walking the ambient working tree.
async fn selected_clean_overlay_cargo_tree(
project_root: &Path,
spec: &CleanOverlaySpec,
) -> anyhow::Result<
std::collections::BTreeMap<PathBuf, rch_common::cargo_path_deps::SelectedCargoEntry>,
> {
use rch_common::cargo_path_deps::{SelectedCargoEntry, validate_selected_cargo_path};
use std::collections::BTreeMap;
fn add_overlay(
root: &Path,
relative: &Path,
entries: &mut BTreeMap<PathBuf, SelectedCargoEntry>,
) -> anyhow::Result<()> {
let metadata = std::fs::symlink_metadata(root.join(relative))?;
if metadata.file_type().is_symlink() {
anyhow::bail!(
"clean-overlay selected path became a symlink: {}",
relative.display()
);
}
if metadata.is_dir() {
for child in std::fs::read_dir(root.join(relative))? {
add_overlay(root, &relative.join(child?.file_name()), entries)?;
}
} else if metadata.is_file() {
entries.insert(relative.to_path_buf(), SelectedCargoEntry::File(None));
} else {
anyhow::bail!(
"unsupported clean-overlay source entry: {}",
relative.display()
);
}
Ok(())
}
spec.verify_overlay_unchanged(project_root)?;
let tree = git_output_bytes(
project_root,
&["ls-tree", "-rz", "--full-tree", spec.base_commit()],
)
.await?;
let base = parse_clean_base_tree(&tree)?;
let mut entries = BTreeMap::new();
for entry in &base {
let selected = match entry.mode {
0o100644 | 0o100755 => SelectedCargoEntry::File(None),
0o120000 => SelectedCargoEntry::Symlink(
String::from_utf8(
git_output_bytes(project_root, &["cat-file", "blob", &entry.object_id]).await?,
)
.with_context(|| format!("non-UTF-8 selected symlink {}", entry.path.display()))?,
),
_ => anyhow::bail!("unsupported selected Git mode for {}", entry.path.display()),
};
entries.insert(entry.path.clone(), selected);
}
for overlay in spec.overlay_paths() {
add_overlay(project_root, overlay, &mut entries)?;
}
let mut metadata_paths: Vec<_> = entries
.keys()
.filter(|path| is_selected_cargo_metadata(path))
.cloned()
.collect();
for (path, entry) in &entries {
if path.file_name().is_some_and(|name| name == ".cargo")
&& matches!(entry, SelectedCargoEntry::Symlink(_))
{
for name in ["config", "config.toml"] {
let candidate = path.join(name);
if validate_selected_cargo_path(&entries, &candidate).is_ok() {
metadata_paths.push(candidate);
}
}
}
}
for path in metadata_paths {
let target = validate_selected_cargo_path(&entries, &path)?;
let bytes = if spec
.overlay_paths()
.iter()
.any(|overlay| target.starts_with(overlay))
{
std::fs::read(project_root.join(&target))
.with_context(|| format!("read selected Cargo metadata {}", target.display()))?
} else {
let entry = base
.iter()
.find(|entry| entry.path == target)
.ok_or_else(|| {
anyhow::anyhow!(
"selected Cargo metadata has no base object: {}",
target.display()
)
})?;
git_output_bytes(project_root, &["cat-file", "blob", &entry.object_id]).await?
};
entries.insert(
target.clone(),
SelectedCargoEntry::File(Some(String::from_utf8(bytes).with_context(|| {
format!("selected Cargo metadata is not UTF-8: {}", target.display())
})?)),
);
}
spec.verify_overlay_unchanged(project_root)?;
Ok(entries)
}
async fn validate_clean_overlay_cargo_sources(
project_root: &Path,
spec: &CleanOverlaySpec,
command: &str,
) -> anyhow::Result<()> {
use rch_common::cargo_path_deps::{
SelectedCargoEntry, validate_selected_cargo_config_at, validate_selected_cargo_path,
validate_selected_cargo_tree_at,
};
let primary_directory = spec.primary_directory.as_deref().unwrap_or(Path::new(""));
let mut entries = selected_clean_overlay_cargo_tree(project_root, spec)
.await?
.into_iter()
.map(|(path, entry)| (primary_directory.join(path), entry))
.collect::<std::collections::BTreeMap<_, _>>();
for (root, selected) in &spec.dependencies {
let directory = root
.file_name()
.ok_or_else(|| anyhow::anyhow!("missing dependency directory"))?;
for (path, entry) in selected_clean_overlay_cargo_tree(root, selected).await? {
anyhow::ensure!(
entries
.insert(Path::new(directory).join(path), entry)
.is_none(),
"overlapping selected dependency inventory"
);
}
}
let primary_manifest = primary_directory.join("Cargo.toml");
validate_selected_cargo_tree_at(&entries, &primary_manifest)?;
let (tokens, cargo_index) = cargo_target_dir::managed_clean_overlay_cargo_tokens(command)?;
// Changing cwd also changes Cargo's configuration search. Until that
// search is represented in the selected-source plan, do not guess.
for token in &tokens[..cargo_index] {
if token == "-C" || token == "--chdir" || token.starts_with("--chdir=") {
anyhow::bail!("clean-overlay Cargo source validation does not support wrapper chdir");
}
}
let mut index = cargo_index + 1;
while let Some(token) = tokens.get(index) {
if token == "--" {
break;
}
if token.starts_with("-C") {
anyhow::bail!("clean-overlay Cargo source validation does not support Cargo -C");
}
let value = if token == "--manifest-path" || token == "--config" {
index += 1;
Some(
tokens
.get(index)
.ok_or_else(|| anyhow::anyhow!("missing {token} value"))?
.as_str(),
)
} else {
token
.strip_prefix("--manifest-path=")
.or_else(|| token.strip_prefix("--config="))
};
if let Some(value) = value {
if token.starts_with("--manifest-path") {
anyhow::ensure!(
Path::new(value)
.file_name()
.is_some_and(|name| name == "Cargo.toml"),
"clean-overlay --manifest-path must select a Cargo.toml manifest"
);
anyhow::ensure!(
!Path::new(value).is_absolute(),
"absolute manifest path is not selected"
);
let path = validate_selected_cargo_path(&entries, &primary_directory.join(value))?;
anyhow::ensure!(
matches!(entries.get(&path), Some(SelectedCargoEntry::File(Some(_)))),
"Cargo manifest argument is not selected Cargo metadata: {}",
path.display()
);
} else {
// Inline TOML is parsed by the same dependency-path validator.
// File-based --config needs an explicit selected-file mapping;
// never read it from the ambient controller or worker tree.
let _: toml::Value = toml::from_str(value).context(
"clean-overlay supports inline TOML --config only; config files are not admitted",
)?;
validate_selected_cargo_config_at(
&entries,
&primary_manifest,
primary_directory,
value,
)?;
}
}
index += 1;
}
spec.verify_overlay_unchanged(project_root)?;
Ok(())
}
async fn detect_clean_overlay_toolchain(
project_root: &Path,
spec: &CleanOverlaySpec,
) -> anyhow::Result<Option<ToolchainInfo>> {
let toml_path = Path::new("rust-toolchain.toml");
let legacy_path = Path::new("rust-toolchain");
let toml_contents = if spec.overlay_paths().iter().any(|path| path == toml_path) {
Some(
std::fs::read_to_string(project_root.join(toml_path))
.context("read overlaid rust-toolchain.toml")?,
)
} else {
git_show_optional(project_root, spec.base_commit(), "rust-toolchain.toml").await?
};
if let Some(contents) = toml_contents {
let document = toml::from_str::<toml::Value>(&contents)
.context("parse rust-toolchain.toml from clean-overlay base")?;
let channel = document
.get("toolchain")
.and_then(|toolchain| toolchain.get("channel"))
.and_then(toml::Value::as_str)
.ok_or_else(|| {
anyhow::anyhow!(
"rust-toolchain.toml in clean-overlay base has no toolchain.channel"
)
})?;
return Ok(Some(parse_channel_string(channel)?));
}
let legacy_contents = if spec.overlay_paths().iter().any(|path| path == legacy_path) {
Some(
std::fs::read_to_string(project_root.join(legacy_path))
.context("read overlaid rust-toolchain")?,
)
} else {
git_show_optional(project_root, spec.base_commit(), "rust-toolchain").await?
};
if let Some(contents) = legacy_contents {
if let Ok(document) = toml::from_str::<toml::Value>(&contents)
&& let Some(channel) = document
.get("toolchain")
.and_then(|toolchain| toolchain.get("channel"))
.and_then(toml::Value::as_str)
{
return Ok(Some(parse_channel_string(channel)?));
}
let channel = contents.trim();
if channel.is_empty() {
anyhow::bail!("rust-toolchain in clean-overlay base is empty");
}
return Ok(Some(parse_channel_string(channel)?));
}
// No committed override means the remote worker's default toolchain is the
// honest analogue. Consulting the ambient worktree here could let an
// unselected dirty rust-toolchain file influence the supposedly clean run.
Ok(None)
}
async fn prepare_clean_overlay_spec(
project_root: &Path,
base: Option<String>,
clean_overlay: bool,
overlay_paths: Vec<PathBuf>,
no_overlay: bool,
) -> anyhow::Result<Option<CleanOverlaySpec>> {
if !clean_overlay {
if base.is_some() || !overlay_paths.is_empty() || no_overlay {
anyhow::bail!("--base, --overlay-path, and --no-overlay require --clean-overlay");
}
return Ok(None);
}
let base = base.ok_or_else(|| anyhow::anyhow!("--clean-overlay requires --base <COMMIT>"))?;
if base.trim().is_empty() || base.chars().any(char::is_control) {
anyhow::bail!("--base must be a non-empty Git revision without control characters");
}
if no_overlay == !overlay_paths.is_empty() {
anyhow::bail!(
"--clean-overlay requires exactly one of --no-overlay or at least one --overlay-path"
);
}
let git_toplevel = git_output(project_root, &["rev-parse", "--show-toplevel"]).await?;
let git_toplevel = std::fs::canonicalize(&git_toplevel)
.with_context(|| format!("canonicalize Git toplevel {git_toplevel}"))?;
let canonical_project_root = std::fs::canonicalize(project_root)
.with_context(|| format!("canonicalize project root {}", project_root.display()))?;
if git_toplevel != canonical_project_root {
anyhow::bail!(
"clean-overlay exec must run from the Git toplevel (current: {}, toplevel: {})",
canonical_project_root.display(),
git_toplevel.display()
);
}
let commit_expression = format!("{base}^{{commit}}");
let base_commit = git_output(
&canonical_project_root,
&[
"rev-parse",
"--verify",
"--end-of-options",
&commit_expression,
],
)
.await?;
if base_commit.is_empty() || !base_commit.chars().all(|ch| ch.is_ascii_hexdigit()) {
anyhow::bail!("--base did not resolve to a hexadecimal commit object ID");
}
validate_clean_overlay_archive_attributes(&canonical_project_root, &base_commit).await?;
let tree_object = git_output(
&canonical_project_root,
&["rev-parse", "--verify", &format!("{base_commit}^{{tree}}")],
)
.await?;
let base_tree = git_output_bytes(
&canonical_project_root,
&["ls-tree", "-rz", "--full-tree", &base_commit],
)
.await?;
let base_entries = parse_clean_base_tree(&base_tree)?;
if let Some(gitlink) = base_entries.iter().find(|entry| entry.mode == 0o160000) {
anyhow::bail!(
"clean-overlay does not support Git submodules yet (found {})",
gitlink.path.display()
);
}
let mut normalized_paths = BTreeSet::new();
for path in overlay_paths {
let normalized = normalize_clean_overlay_path(&path)?;
if normalized.components().any(|component| {
component
.as_os_str()
.to_str()
.is_some_and(|name| name.eq_ignore_ascii_case(".git"))
}) {
anyhow::bail!(
"clean-overlay paths may not contain Git metadata: {}",
normalized.display()
);
}
validate_clean_overlay_live_path(&canonical_project_root, &normalized)?;
normalized_paths.insert(normalized);
}
let overlay_paths = normalized_paths.into_iter().collect::<Vec<_>>();
validate_clean_overlay_against_base(&canonical_project_root, &overlay_paths, &base_entries)?;
let overlay_fingerprint = clean_overlay_fingerprint(&canonical_project_root, &overlay_paths)?;
Ok(Some(CleanOverlaySpec {
base_commit,
tree_object,
overlay_paths,
overlay_fingerprint,
dependencies: Vec::new(),
primary_directory: None,
}))
}
/// Bind every external repository explicitly; never infer a sibling revision
/// from ambient working-tree bytes. The first supported layout is sibling Git
/// roots, materialized together under one leased container.
async fn bind_clean_overlay_dependencies(
primary: &Path,
spec: &mut CleanOverlaySpec,
bindings: &[String],
) -> anyhow::Result<()> {
if bindings.is_empty() {
return Ok(());
}
let primary = std::fs::canonicalize(primary)?;
let parent = primary
.parent()
.ok_or_else(|| anyhow::anyhow!("missing primary parent"))?;
let primary_name = normalize_clean_overlay_path(Path::new(
primary
.file_name()
.ok_or_else(|| anyhow::anyhow!("missing primary directory name"))?,
))?;
let mut roots = BTreeSet::new();
let mut names = BTreeSet::from([primary_name.to_string_lossy().to_ascii_lowercase()]);
for binding in bindings {
let (path, revision) = binding
.split_once('=')
.ok_or_else(|| anyhow::anyhow!("--dependency-base requires PATH=REV"))?;
anyhow::ensure!(
!path.is_empty() && !revision.is_empty(),
"empty dependency path or revision"
);
let root = std::fs::canonicalize(primary.join(path))
.with_context(|| format!("resolve selected dependency root {path}"))?;
anyhow::ensure!(
root != primary && root.parent() == Some(parent),
"--dependency-base currently requires sibling Git roots: {}",
root.display()
);
let name = normalize_clean_overlay_path(Path::new(
root.file_name()
.ok_or_else(|| anyhow::anyhow!("missing dependency directory name"))?,
))?;
anyhow::ensure!(
names.insert(name.to_string_lossy().to_ascii_lowercase()),
"ambiguous dependency directory"
);
anyhow::ensure!(
roots.insert(root.clone()),
"duplicate dependency binding: {}",
root.display()
);
let selected = prepare_clean_overlay_spec(&root, Some(revision.into()), true, vec![], true)
.await?
.ok_or_else(|| anyhow::anyhow!("missing dependency selection"))?;
spec.dependencies.push((root, selected));
}
spec.dependencies.sort_by(|a, b| a.0.cmp(&b.0));
spec.primary_directory = Some(primary_name);
Ok(())
}
fn is_clean_overlay_cargo_fmt_check(command_parts: &[String]) -> bool {
let mut parts = command_parts.iter();
let Some(program) = parts.next() else {
return false;
};
if Path::new(program)
.file_name()
.and_then(|name| name.to_str())
!= Some("cargo")
{
return false;
}
let Some(mut subcommand) = parts.next() else {
return false;
};
if subcommand.starts_with('+') {
let Some(next) = parts.next() else {
return false;
};
subcommand = next;
}
subcommand == "fmt" && parts.any(|part| part == "--check")
}
#[allow(clippy::too_many_arguments)] // Pipeline wiring favors explicit params
pub async fn run_exec(
base: Option<String>,
dependency_bases: Vec<String>,
clean_overlay: bool,
overlay_paths: Vec<PathBuf>,
no_overlay: bool,
source_content_receipt: bool,
job: bool,
result_dirs: Vec<PathBuf>,
required_tools: Vec<String>,
command_parts: Vec<String>,
out_ctx: &crate::ui::context::OutputContext,
) -> anyhow::Result<()> {
let exec_start = std::time::Instant::now();
// bd-uoh4x: machine consumers (`--json` / `--format`) get a result
// envelope on stdout at every terminal path. Remote command output keeps
// streaming to stderr, so stdout stays envelope-only.
set_machine_output(out_ctx.is_json());
let command = join_exec_command(&command_parts);
if command.is_empty() {
anyhow::bail!("No command provided to exec");
}
if shell_wrapped_cargo_command(&command_parts) {
let reporter = HookReporter::new(OutputVisibility::Summary);
reporter.summary_critical(
"[RCH-E301] refusing shell-wrapped cargo command: RCH cannot safely synchronize its artifacts; invoke `rch exec -- cargo ...` directly",
);
emit_exec_envelope(&ExecResultEnvelope {
api_version: "1.0",
command: &command,
outcome: "refused",
location: "local",
fallback_reason: Some("shell-wrapped cargo command"),
worker_id: None,
remote_exit_code: None,
duration_ms: Some(exec_start.elapsed().as_millis() as u64),
timing: None,
result_dirs: None,
error_code: Some("RCH-E301"),
});
std::process::exit(EXIT_BUILD_ERROR);
}
// A clean-overlay request can never fall back to the ambient local tree,
// even if the caller omitted RCH_REQUIRE_REMOTE. Doing so would silently
// defeat the entire peer-dirt exclusion guarantee.
let require_remote = exec_requires_remote() || clean_overlay || source_content_receipt;
// Classify the command. In explicit job-admission mode (`rch exec --job`,
// bd-bu3fb) the classifier is bypassed entirely and the command is admitted
// as a `CompilationKind::Job`: the hook can never produce this kind, so
// auto-delegation of non-compilation workloads remains impossible.
// Declared job result directories (bd-p0yoo) are validated up front so an
// unsafe path fails before any worker selection or transfer happens.
let result_dirs = match validate_job_result_dirs(result_dirs) {
Ok(validated) => validated,
Err(e) => {
let reporter = HookReporter::new(OutputVisibility::Summary);
reporter.summary_critical(&format!("[RCH-E001] {e}"));
emit_exec_envelope(&ExecResultEnvelope {
api_version: "1.0",
command: &command,
outcome: "collection_error",
location: "local",
fallback_reason: None,
worker_id: None,
remote_exit_code: None,
duration_ms: Some(exec_start.elapsed().as_millis() as u64),
timing: None,
result_dirs: None,
error_code: Some("RCH-E001"),
});
std::process::exit(2);
}
};
let classification = if job {
Classification::compilation(
CompilationKind::Job,
1.0,
"explicit `rch exec --job` admission",
)
} else {
classify_command(&command)
};
let clean_overlay_fmt_check = clean_overlay && is_clean_overlay_cargo_fmt_check(&command_parts);
if !classification.is_compilation && !clean_overlay_fmt_check {
// This should not normally happen because the hook only rewrites
// compilations. Preserve the ordinary local behavior, but honor
// RCH_REQUIRE_REMOTE for explicit `rch exec` invocations.
warn!("exec called with non-compilation command: {}", command);
let reporter = HookReporter::new(OutputVisibility::Summary);
exit_with_local_fallback(
&command,
&reporter,
"non-compilation command",
require_remote,
);
}
let config = match load_config() {
Ok(cfg) => cfg,
Err(e) => {
warn!("Failed to load config: {}, running locally", e);
let reporter = HookReporter::new(OutputVisibility::Summary);
exit_with_local_fallback(&command, &reporter, "config unavailable", require_remote);
}
};
// A project's `[jobs] required_tools` is a statement that its jobs cannot
// run without those tools, so it ADDS to `--require-tool` rather than being
// replaced by it. Job mode only: a project-wide requirement that silently
// narrowed every ordinary build's worker pool would be a surprising way to
// lose the fleet. Duplicates collapse so the daemon sees each name once.
let required_tools = {
let mut tools = required_tools;
if job {
for tool in &config.jobs.required_tools {
let tool = tool.trim();
if !tool.is_empty() && !tools.iter().any(|existing| existing == tool) {
tools.push(tool.to_string());
}
}
}
tools
};
// Issue #55: honor the same config knobs the hook honors, so
// `general.force_local` / `general.enabled` / `execution.allowlist` mean the
// same thing whether a build arrives via the hook, the cargo shim, or an
// explicit `rch exec`. Explicit config outranks the shim's baked
// RCH_REQUIRE_REMOTE=1, but never a mode whose guarantee cannot be met by
// a local run (clean overlay, source-content receipt).
if let Some(policy) = config_local_policy(&config, classification.kind) {
let reporter = HookReporter::new(config.output.visibility);
let reason = policy.reason();
let refuse_local = if policy.overrides_require_remote() {
clean_overlay || source_content_receipt
} else {
require_remote || role_requires_remote(config.general.role)
};
if policy.overrides_require_remote() && require_remote && !refuse_local {
warn!("RCH_REQUIRE_REMOTE is set but config says {reason}; running locally per config");
}
if !refuse_local {
reporter.summary(&format!("[RCH] local ({reason})"));
}
exit_with_local_fallback(&command, &reporter, &reason, refuse_local);
}
// bd-wywsj: a dispatcher box defaults offloadable builds to
// fail-closed + queue — box policy, not a per-call env. Explicit
// env (even RCH_REQUIRE_REMOTE=0) remains the per-call override.
let require_remote = require_remote || role_requires_remote(config.general.role);
if role_requires_remote(config.general.role) {
info!("role=dispatcher: offloadable build defaults to fail-closed + queue");
}
let reporter = HookReporter::new(config.output.visibility);
// Build path topology policy from loaded config so that any normalization
// warnings reference the configured roots rather than compiled-in defaults.
let topology_policy = config.path_topology.to_policy();
// bd-raobv: transfer cannot mirror a project outside the canonical root.
// Decide that before a worker slot is reserved, with a reason an operator
// can act on, and record it like every other fallback. A mode that forbids
// local execution refuses instead.
if let Ok(cwd) = std::env::current_dir()
&& let Some(reason) = project_topology_local_reason(&topology_policy, &cwd)
{
let refuse_local = require_remote || clean_overlay || source_content_receipt;
if !refuse_local {
reporter.summary(&format!("[RCH] local ({reason})"));
}
exit_with_local_fallback(&command, &reporter, &reason, refuse_local);
}
// Extract project name honoring configured path topology.
let project = extract_project_name_with_policy(&topology_policy);
// Layer 0 pack (bd-bqu38): resolve knob env pairs once; empty unless the
// `[layer0]` section enables the pack.
let layer0_env: Vec<(String, String)> = config
.layer0
.active_env()
.into_iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect();
// Estimate cores needed
let estimated_cores =
estimate_cores_for_command(classification.kind, &command, &config.compilation);
// Detect toolchain — ONLY for Rust kinds.
//
// `detect_toolchain` returns the ambient rustup toolchain (e.g. from
// rust-toolchain.toml, or the active default). Attaching it to a NON-Rust
// request is actively harmful: rchd runs a per-worker toolchain preflight, and
// a worker that does not happen to have that exact nightly installed fails it
// and is excluded as a HARD preflight failure. With every worker excluded the
// build falls back to LOCAL — so a `go build` on a box whose rustup default is
// `nightly-2026-07-11` would silently keep running on the orchestrator, which
// is precisely the bug this feature exists to fix. Go/TS/Bun/Nix builds need no
// rustup toolchain, so send none.
let project_root = std::env::current_dir().ok();
let mut clean_overlay_spec = prepare_clean_overlay_spec(
project_root.as_deref().unwrap_or_else(|| Path::new(".")),
base,
clean_overlay,
overlay_paths,
no_overlay,
)
.await?;
if !dependency_bases.is_empty() {
anyhow::ensure!(
clean_overlay_fmt_check
|| classify_command(&command)
.kind
.is_some_and(|kind| kind.command_base() == "cargo"),
"--dependency-base requires a directly classified Cargo command"
);
let spec = clean_overlay_spec
.as_mut()
.ok_or_else(|| anyhow::anyhow!("--dependency-base requires --clean-overlay"))?;
bind_clean_overlay_dependencies(
project_root.as_deref().unwrap_or_else(|| Path::new(".")),
spec,
&dependency_bases,
)
.await?;
}
let selection_project = selection_project_for_execution(
&project,
clean_overlay_spec.as_ref(),
uuid::Uuid::new_v4(),
);
// Issue #63: `--job` bypasses classification (kind = Job), so the request
// used to carry NO toolchain even for a delegated `cargo clippy` — and the
// daemon's clippy component gate then fail-closed EVERY worker as
// `capability_missing:rustup_component:<unknown>:clippy` and the build
// silently ran locally. Derive the toolchain for Rust-shaped job commands
// exactly like the classified rail does.
let job_needs_rust_toolchain = job
&& matches!(
required_runtime_for_kind(classify_command(&command).kind),
RequiredRuntime::Rust
);
let needs_rust_toolchain = clean_overlay_fmt_check
|| matches!(
required_runtime_for_kind(classification.kind),
RequiredRuntime::Rust
)
|| job_needs_rust_toolchain;
let toolchain = if needs_rust_toolchain {
if let (Some(root), Some(spec)) = (project_root.as_deref(), clean_overlay_spec.as_ref()) {
detect_clean_overlay_toolchain(root, spec).await?
} else if let Some(name) = rustup_toolchain_from_command_tokens(&command_parts) {
// An explicit `RUSTUP_TOOLCHAIN=…` / `cargo +…` in the delegated
// command outranks the ambient project toolchain (rustup's own
// precedence), so preflight and the component gate must check the
// toolchain the remote cargo will actually use (issue #63).
parse_channel_string(&name).ok()
} else {
project_root
.as_ref()
.and_then(|root| detect_toolchain(root).ok())
}
} else {
None
};
let forwarded_cargo_target_dir = resolve_forwarded_cargo_target_dir(
classification.kind,
project_root.as_deref().unwrap_or_else(|| Path::new(".")),
&reporter,
Some(&command_parts),
);
let remote_command = rewrite_cargo_target_dir_command_for_remote(
&command,
Some(&command_parts),
forwarded_cargo_target_dir.as_ref(),
&reporter,
);
// Determine required runtime
let required_runtime = if clean_overlay_fmt_check {
RequiredRuntime::Rust
} else {
required_runtime_for_kind(classification.kind)
};
let command_priority = command_priority_from_env(&reporter);
let wait_for_worker = queue_when_busy_enabled();
let preferred_workers = command_parsing::preferred_workers();
let durable_lease = DurableLeaseWriter::create(
&command,
require_remote,
config.self_healing.hook_starts_daemon,
)?;
let wrapper_id = durable_lease.wrapper_id();
if restart_admission_is_closed(&config.general.socket_path)
.await
.unwrap_or(false)
{
durable_lease.heartbeat("restart_admission_blocked")?;
return Err(anyhow::anyhow!(
"remote build admission is paused while daemon restart remediation is active"
));
}
// Query daemon for worker selection
let response = match query_daemon(
&config.general.socket_path,
&selection_project,
estimated_cores,
&remote_command,
toolchain.as_ref(),
required_runtime,
command_priority,
0, // classification duration not relevant here
Some(std::process::id()),
Some(&wrapper_id),
wait_for_worker,
&preferred_workers,
classification.kind == Some(CompilationKind::Job),
&required_tools,
)
.await
{
Ok(resp) => resp,
Err(e) => {
let e = selection_error_for_recovery(e, &durable_lease)?;
warn!("Failed to query daemon: {}, attempting recovery", e);
// Classify the failure and detect a configured-vs-canonical socket
// mismatch, then record a durable structured incident (RCH-I010)
// so postmortems can see *why* the hook could not reach the daemon.
let socket_path = config.general.socket_path.clone();
let socket_exists = Path::new(&socket_path).exists();
let failure_kind = classify_socket_failure(&e, socket_exists);
let strict_remote = require_remote;
let canonical_socket = default_socket_path();
let canonical_exists = Path::new(&canonical_socket).exists();
let mismatch =
detect_socket_path_mismatch(&socket_path, &canonical_socket, canonical_exists);
// Privacy-safe fingerprint (secrets and home paths masked).
let command_fingerprint = redact_secrets(&command);
record_hook_incident(&build_socket_failure_incident(
failure_kind,
mismatch.as_ref(),
&project,
&command_fingerprint,
strict_remote,
now_unix_ms(),
));
// Attempt a bounded daemon autostart, then retry selection ONCE.
let retry =
if auto_start::try_auto_start_daemon(&config.self_healing, Path::new(&socket_path))
.await
.is_ok()
{
match query_daemon(
&socket_path,
&selection_project,
estimated_cores,
&remote_command,
toolchain.as_ref(),
required_runtime,
command_priority,
0,
Some(std::process::id()),
Some(&wrapper_id),
wait_for_worker,
&preferred_workers,
classification.kind == Some(CompilationKind::Job),
&required_tools,
)
.await
{
Ok(response) => Some(response),
Err(error) => {
let _ = selection_error_for_recovery(error, &durable_lease)?;
None
}
}
} else {
None
};
match decide_recovery_action(retry.is_some(), strict_remote) {
// Daemon came back after autostart + retry — proceed remotely.
// ProceedRemote implies `retry` is Some; fail open defensively
// rather than panicking if that invariant is ever violated.
DaemonRecoveryAction::ProceedRemote => retry.unwrap_or_else(|| {
reporter.summary("[RCH] local (daemon unavailable)");
exit_with_local_fallback(
&command,
&reporter,
"daemon unavailable",
require_remote,
);
}),
// Fail-open convenience lane: run local. exit_with_local_fallback
// records the RCH-I011 incident (bd-qawj7).
DaemonRecoveryAction::LocalFallback => {
reporter.summary("[RCH] local (daemon unavailable)");
exit_with_local_fallback(
&command,
&reporter,
"daemon unavailable",
require_remote,
);
}
// Proof lane: fail closed. exit_with_local_fallback refuses under
// proof mode, prints the explicit "remote required" refusal
// summary, and records the RCH-I012 incident (bd-qawj7).
DaemonRecoveryAction::Refuse => {
exit_with_local_fallback(
&command,
&reporter,
"daemon unavailable",
require_remote,
);
}
}
}
};
// FIX 1: capacity-aware remote-failure fallback loop.
//
// A worker-fault failure (OOM/signal-kill, missing worker system dependency,
// missing toolchain, or a generic pipeline error) must NOT immediately revert
// the heavy compile to the local orchestrator (which floods the box that runs
// the whole fleet). Instead we retry on a different, higher-capacity worker
// ranked from daemon telemetry; local execution is a last resort gated by
// `compilation.allow_local_fallback`. Genuine build/test failures and
// SSH-timeout fail-closed cases stay terminal (retrying wastes fleet cycles).
let allow_local_fallback = config.compilation.allow_local_fallback;
// A source-content receipt is bound to one selected worker and one remote
// root set. Retrying on another worker would emit multiple competing proof
// envelopes for one invocation, so proof mode is deliberately single-shot.
let max_attempts = if source_content_receipt {
1
} else {
max_remote_attempts()
};
let mut tried_workers: Vec<WorkerId> = Vec::new();
// The worker set requested for the CURRENT `response`. Attempt 1 uses the
// operator's env pin (`RCH_WORKER(S)`); each retry pins the chosen bigger
// worker so the daemon reserves exactly it.
let mut current_query_preferred = preferred_workers.clone();
let mut attempt: u32 = 1;
let mut response = response;
loop {
// Only the first iteration can observe an unassigned worker: a retry
// re-query replaces `response` solely when it carries a worker.
if selection_cancelled_before_start(&response) {
durable_lease.record_exit(130)?;
durable_lease.acknowledge_terminal()?;
reporter.summary("[RCH] cancelled before remote admission");
std::process::exit(130);
}
let Some(worker) = response.worker.clone() else {
let requested_refusal = requested_worker_refusal(&preferred_workers, &response);
// A pin that could not be honored is the exact failure the allow-set fix
// exists for; record a durable incident row so postmortems have a trace
// even when the summary line is suppressed at stock visibility (rch#35).
if let Some(refusal) = requested_refusal.as_ref() {
record_hook_incident(&build_requested_worker_refusal_incident(
refusal.reason_code,
&preferred_workers,
&project,
&redact_secrets(&command),
require_remote,
now_unix_ms(),
));
}
let mut reason = requested_refusal
.map(|refusal| refusal.summary)
.unwrap_or_else(|| response.reason.to_string());
// Issue #63(c): an explicit --job admission with no admissible
// worker refuses (typed, retryable) instead of silently
// compiling locally on the orchestrator.
if job && !require_remote {
reason = format!("job admission exhausted: {reason}");
}
exit_with_gated_local_fallback(
&command,
&reporter,
&reason,
strict_remote_for_exhausted_admission(require_remote, job),
allow_local_fallback,
None,
);
};
let Some(remote_build_id) = response.build_id else {
// A worker reservation without a daemon build identity cannot be
// correlated to this durable wrapper lease. Release it rather
// than executing an untrackable build; the lease intentionally
// remains nonterminal because release evidence was not obtained.
let release = release_worker(
&config.general.socket_path,
&worker.id,
estimated_cores,
None,
Some(EXIT_BUILD_ERROR),
None,
None,
None,
Some(&wrapper_id),
)
.await;
if let Err(error) = release {
warn!(
"Worker {} selected without build id and release was not acknowledged: {}",
worker.id, error
);
}
anyhow::bail!(
"daemon selected worker {} without a durable build id; refusing untrackable execution",
worker.id
);
};
if let Err(error) = durable_lease.admit(remote_build_id, &worker.id) {
let warning = rch_common::job_recovery::diagnose_stuck_wrapper(
&durable_lease.snapshot().identity,
&rch_common::job_recovery::WrapperState::waiting(
rch_common::job_identity::JobLifecycleState::Queued,
)
.reservation_failed(),
);
eprintln!("{}", serde_json::to_string(&warning)?);
release_worker(&config.general.socket_path, &worker.id, estimated_cores,
Some(remote_build_id), Some(EXIT_BUILD_ERROR), None, None, None, Some(&wrapper_id)).await
.context("reservation release failed after durable admission failure; command was not executed")?;
exit_with_gated_local_fallback(
&command,
&reporter,
&format!("durable reservation persistence failed: {error}"),
require_remote || job,
allow_local_fallback,
None,
);
}
if !selected_worker_is_requested(&worker.id, ¤t_query_preferred) {
let requested = current_query_preferred
.iter()
.map(WorkerId::as_str)
.collect::<Vec<_>>()
.join(",");
let reason = format!(
"[RCH-I001] daemon selected unrequested worker '{}' for requested set [{}]; upgrade and restart rchd to the same version as rch, or request another worker",
worker.id, requested
);
// A pre-fix daemon may already have reserved slots and opened an
// active-build record for the out-of-set worker. Release both before
// refusing so the client-side mixed-version guard cannot leak capacity.
let release_error = release_worker(
&config.general.socket_path,
&worker.id,
estimated_cores,
response.build_id,
Some(EXIT_BUILD_ERROR),
None,
None,
None,
Some(&wrapper_id),
)
.await
.err();
if let Some(error) = release_error.as_ref() {
warn!(
"Failed to release unrequested worker {} after selection refusal: {}",
worker.id, error
);
} else if let Err(error) = durable_lease.acknowledge_terminal() {
warn!(
"Failed to persist acknowledged durable lease terminal state: {}",
error
);
}
let reason = if let Some(error) = release_error {
format!(
"{reason}; reservation release was not acknowledged ({error}); restart rchd and verify capacity before retrying"
)
} else {
reason
};
reporter.summary(&format!("[RCH] local ({reason})"));
exit_with_local_fallback(&command, &reporter, &reason, require_remote);
}
info!(
"Selected worker: {} at {}@{} ({} slots remaining after reservation, speed {:.1}){}",
worker.id,
worker.user,
worker.host,
worker.slots_available,
worker.speed_score,
if attempt > 1 {
format!(" [remote retry {attempt}/{max_attempts}]")
} else {
String::new()
}
);
tried_workers.push(worker.id.clone());
// Execute remote compilation pipeline (topology_policy was built earlier
// from the loaded config so diagnostics reference configured roots).
let remote_start = Instant::now();
let result = execute_remote_compilation(
&worker,
&remote_command,
config.transfer.clone(),
config.environment.allowlist.clone(),
forwarded_cargo_target_dir.clone(),
&config.compilation,
toolchain.as_ref(),
classification.kind,
&reporter,
&config.general.socket_path,
config.output.color_mode,
Some(remote_build_id),
Some(&wrapper_id),
Some(&durable_lease),
&topology_policy,
clean_overlay_spec.as_ref(),
source_content_receipt,
&result_dirs,
&layer0_env,
config.remediation.pooled_target.reaper_pooled_idle_hours,
config.remediation.pooled_target.store_base.as_deref(),
)
.await;
let remote_elapsed = remote_start.elapsed();
// bd-uoh4x: snapshot for envelope emission — the match below may move
// `result`, and transport-failure paths have no successful result at
// all (their envelopes carry None for both).
let exec_timing = result.as_ref().ok().map(|r| r.timing.clone());
let exec_dir_stats = result.as_ref().ok().map(|r| r.result_dirs.clone());
if let Err(error) = durable_lease.heartbeat("finalize") {
warn!(
"Failed to persist durable lease finalize heartbeat: {}",
error
);
}
// An unconfirmed execution still owns its reservation. A generic
// failure release would fabricate completion and permit another job.
let retain_unconfirmed_ownership = result
.as_ref()
.err()
.is_some_and(is_remote_execution_unconfirmed);
let release_exit_code = result
.as_ref()
.map(|ok| ok.exit_code)
.unwrap_or(EXIT_BUILD_ERROR);
let release_timing = result.as_ref().ok().map(|ok| {
let mut timing = ok.timing.clone();
timing.total = Some(remote_elapsed);
timing
});
// A worker-caused failure must not warm that worker's cache for the
// project (review of GH #81), or the next build is routed back to it.
let release_worker_fault = result
.as_ref()
.is_ok_and(|ok| remote_failure_is_worker_fault(&ok.stderr, ok.exit_code));
let release_acknowledged = if retain_unconfirmed_ownership {
warn!(
"Remote completion unconfirmed; retaining build {} ownership",
remote_build_id
);
false
} else {
match release_worker_with_fault(
&config.general.socket_path,
&worker.id,
estimated_cores,
Some(remote_build_id),
Some(release_exit_code),
None,
None,
release_timing.as_ref(),
Some(&wrapper_id),
release_worker_fault,
)
.await
{
Ok(()) => true,
Err(error) => {
warn!("Failed to release worker slots: {}", error);
false
}
}
};
if let Ok(result) = result.as_ref() {
// Persist the observed command result even when the daemon's release
// acknowledgement is lost. Recovery then knows the command outcome
// without pretending its reservation/ownership was retired.
durable_lease.record_exit(result.exit_code)?;
}
if !retain_unconfirmed_ownership && !release_acknowledged {
if let Err(persist_error) = durable_lease.heartbeat("release_unconfirmed") {
return Err(release_unconfirmed_error(&worker.id, remote_build_id).context(
format!("could not persist release uncertainty in the durable lease: {persist_error}"),
));
}
return Err(release_unconfirmed_error(&worker.id, remote_build_id));
}
// Classify the outcome. Terminal cases (success, real build/test failure,
// preflight/transfer-skip decisions, SSH-timeout fail-closed) exit here;
// worker-fault cases yield a `RetryableRemoteFault` handled below.
let fault: RetryableRemoteFault = match result {
Ok(result) => {
// RABS B002: completed exec-path invocations join the
// corpus too (off the latency path; fail-open inside).
{
let command_for_corpus = command.to_string();
let cwd_for_corpus =
std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
let kind_for_corpus = classification.kind;
let exit_code = result.exit_code;
let duration_ms = result.duration_ms;
tokio::task::spawn_blocking(move || {
record_invocation(
&command_for_corpus,
&cwd_for_corpus,
kind_for_corpus,
exit_code,
duration_ms,
);
});
}
if result.exit_code == 0 {
reporter.summary(&format!(
"[RCH] remote {} ({})",
worker.id,
format_duration_ms(remote_elapsed)
));
if let Some(spec) = clean_overlay_spec.as_ref() {
// A clean-overlay caller opted into a content-bound
// execution. Keep the receipt visible even when normal
// progress output is suppressed, so a successful remote
// run remains attributable to the exact bytes executed.
reporter.summary_critical(&spec.execution_receipt());
}
// Record successful build
let is_test = classification
.kind
.map(|kind| kind.is_test_command())
.unwrap_or(false);
if let Err(e) =
record_build(&config.general.socket_path, &worker.id, &project, is_test)
.await
{
warn!("Failed to record build: {}", e);
}
if release_acknowledged && let Err(error) = durable_lease.acknowledge_terminal()
{
warn!(
"Failed to persist acknowledged durable lease terminal state: {}",
error
);
}
emit_exec_envelope(&ExecResultEnvelope {
api_version: "1.0",
command: &remote_command,
outcome: "completed",
location: "remote",
fallback_reason: None,
worker_id: Some(worker.id.as_str()),
remote_exit_code: Some(result.exit_code),
duration_ms: Some(exec_start.elapsed().as_millis() as u64),
timing: Some(&result.timing),
result_dirs: Some(&result.result_dirs),
error_code: None,
});
std::process::exit(0);
// bd-bu3fb: a job's completed remote command is terminal truth.
// The toolchain and worker-system-dependency heuristics
// reinterpret the command's own stderr as a WORKER fault and
// re-execute the workload (remotely, then possibly locally via
// GatedLocalFallback). A non-compilation job must surface its
// remote exit status verbatim instead — so both heuristics are
// skipped for jobs and control falls through to the signal /
// genuine-failure arms below (which never rerun the command
// locally).
} else if result.deadline_triggered {
reporter.summary(&format!(
"[RCH] remote {} exceeded its configured deadline (exit {}); not retrying",
worker.id, result.exit_code
));
emit_exec_envelope(&ExecResultEnvelope {
api_version: "1.0",
command: &remote_command,
outcome: "deadline_exceeded",
location: "remote",
fallback_reason: None,
worker_id: Some(worker.id.as_str()),
remote_exit_code: Some(result.exit_code),
duration_ms: Some(exec_start.elapsed().as_millis() as u64),
timing: exec_timing.as_ref(),
result_dirs: exec_dir_stats.as_deref(),
error_code: None,
});
if release_acknowledged && let Err(error) = durable_lease.acknowledge_terminal()
{
warn!("Failed to persist acknowledged deadline terminal state: {error}");
}
std::process::exit(result.exit_code);
} else if !job && is_toolchain_failure(&result.stderr, result.exit_code) {
// Worker missing the toolchain — another worker may have it.
warn!(
"Remote toolchain failure on {}; will retry on another worker if available",
worker.id
);
RetryableRemoteFault {
log_reason: "remote toolchain missing".to_string(),
failed_worker: worker.id.clone(),
on_exhaust: RemoteFaultExhaustAction::GatedLocalFallback {
reason: format!("toolchain missing on {}", worker.id),
error: None,
},
}
} else if !job
&& let Some(env_failure) =
detect_worker_system_dependency_failure(&result.stderr, result.exit_code)
{
// Worker missing a system dependency — another worker may have it.
let error = ErrorCode::BuildEnvError;
warn!(
"Remote worker build-environment failure on {} [{}]: {}; will retry on another worker if available",
worker.id,
error.code_string(),
env_failure.log_detail()
);
RetryableRemoteFault {
log_reason: format!(
"worker build-environment failure ({})",
env_failure.summary()
),
failed_worker: worker.id.clone(),
on_exhaust: RemoteFaultExhaustAction::ExitWithEnvRemediation {
code: result.exit_code,
summary: format!(
"[RCH] remote {} failed [{}] {}",
worker.id,
error.code_string(),
env_failure.summary()
),
remediation: format!(
"[RCH] remediation [{}]: {}",
error.code_string(),
env_failure.remediation()
),
},
}
} else if let Some(signal) = is_signal_killed(result.exit_code)
&& is_cpu_capability_signal(signal)
{
// SIGILL = worker CPU-capability fault (bd-68hon), NOT
// resource exhaustion: the worker's CPU cannot execute
// the build's codegen (e.g. AVX2 on a pre-x86-64-v3
// box). The build is fine; the WORKER is incompatible.
// Quarantine it at the daemon so the first SIGILL
// auto-routes every later build around it, then retry
// this build on any other worker.
warn!(
"Remote build killed by SIGILL (exit {}) on {} — worker CPU cannot execute build (likely missing ISA, e.g. AVX2); quarantining worker and retrying elsewhere",
result.exit_code, worker.id
);
if let Err(e) = disable_worker_for_fault(
&config.general.socket_path,
&worker.id,
"cpu-capability-fault (SIGILL) — likely missing ISA (e.g. AVX2)",
)
.await
{
warn!("Failed to quarantine SIGILL worker {}: {}", worker.id, e);
}
RetryableRemoteFault {
log_reason: format!(
"worker CPU cannot execute build (SIGILL, exit {}) — quarantined",
result.exit_code
),
failed_worker: worker.id.clone(),
on_exhaust: RemoteFaultExhaustAction::ExitWithCode {
code: result.exit_code,
summary: format!(
"[RCH] remote {} killed (SIGILL — CPU capability)",
worker.id
),
},
}
} else if let Some(signal) =
wrapped_cpu_capability_signal(result.exit_code, &result.stderr)
{
// bd-68hon (wrapped shape): a build-script/proc-macro
// SIGILL is reported by cargo as exit 101 with the signal
// named only in its diagnostics — the exit-code arm above
// never sees it. Same fault, same handling: quarantine
// the worker, retry this build on any other worker.
warn!(
"Remote build's build-script/proc-macro killed by {} on {} (cargo exit {}) — worker CPU cannot execute build (likely missing ISA, e.g. AVX2); quarantining worker and retrying elsewhere",
signal_name(signal),
worker.id,
result.exit_code
);
if let Err(e) = disable_worker_for_fault(
&config.general.socket_path,
&worker.id,
"cpu-capability-fault (build-script SIGILL) — likely missing ISA (e.g. AVX2)",
)
.await
{
warn!("Failed to quarantine SIGILL worker {}: {}", worker.id, e);
}
RetryableRemoteFault {
log_reason: format!(
"worker CPU cannot execute build ({} in build script, cargo exit {}) — quarantined",
signal_name(signal),
result.exit_code
),
failed_worker: worker.id.clone(),
on_exhaust: RemoteFaultExhaustAction::ExitWithCode {
code: result.exit_code,
summary: format!(
"[RCH] remote {} build script killed ({} — CPU capability)",
worker.id,
signal_name(signal)
),
},
}
} else if let Some(signal) = is_signal_killed(result.exit_code) {
// The exit code cannot distinguish OOM from an external
// kill or SSH/session cleanup. Preserve the retry policy.
warn!(
"Remote build killed by {} (exit {}) on {} — cause unconfirmed; inspect worker OOM and SSH/session logs; will retry on another worker if available",
signal_name(signal),
result.exit_code,
worker.id
);
RetryableRemoteFault {
log_reason: format!(
"killed by {} (exit {})",
signal_name(signal),
result.exit_code
),
failed_worker: worker.id.clone(),
on_exhaust: RemoteFaultExhaustAction::ExitWithCode {
code: result.exit_code,
summary: format!(
"[RCH] remote {} killed ({}) — cause unconfirmed; inspect worker OOM and SSH/session logs",
worker.id,
signal_name(signal)
),
},
}
} else {
emit_exec_envelope(&ExecResultEnvelope {
api_version: "1.0",
command: &remote_command,
outcome: "completed",
location: "remote",
fallback_reason: None,
worker_id: Some(worker.id.as_str()),
remote_exit_code: Some(result.exit_code),
duration_ms: Some(exec_start.elapsed().as_millis() as u64),
timing: exec_timing.as_ref(),
result_dirs: exec_dir_stats.as_deref(),
error_code: None,
});
// Genuine build/test failure — the crate is broken and fails
// identically on every worker, so do NOT retry or flood local.
reporter.summary(&format!(
"[RCH] remote {} failed (exit {})",
worker.id, result.exit_code
));
if release_acknowledged && let Err(error) = durable_lease.acknowledge_terminal()
{
warn!(
"Failed to persist acknowledged durable lease terminal state: {}",
error
);
}
std::process::exit(result.exit_code);
}
}
Err(e) => {
if classify_remote_pipeline_failure(&e)
== RemotePipelineFailurePolicy::AllowLocalFallback
&& let Some(preflight_err) = e.downcast_ref::<DependencyPreflightFailure>()
{
// Project-specific: retrying a different worker cannot help.
let evidence_summary = preflight_err.evidence_summary();
warn!(
"Dependency preflight blocked remote execution [{}]: {}; evidence='{}'",
preflight_err.reason_code, preflight_err.remediation, evidence_summary
);
reporter.summary(&format!(
"[RCH] local (dependency preflight {}: {}; evidence: {})",
preflight_err.reason_code, preflight_err.remediation, evidence_summary
));
reporter.verbose(&format!(
"[RCH] dependency preflight report: {}",
preflight_err.report_json()
));
let fallback_reason =
format!("dependency preflight failed: {evidence_summary}");
if release_acknowledged && let Err(error) = durable_lease.acknowledge_terminal()
{
warn!(
"Failed to persist acknowledged durable lease terminal state: {}",
error
);
}
exit_with_local_fallback(&command, &reporter, &fallback_reason, require_remote);
}
// Transfer skip (a "run locally instead" decision, not a failure).
if classify_remote_pipeline_failure(&e)
== RemotePipelineFailurePolicy::AllowLocalFallback
&& let Some(skip_err) = e.downcast_ref::<TransferError>()
&& let TransferError::TransferSkipped { reason } = skip_err
{
reporter.summary(&format!("[RCH] local ({})", reason));
if release_acknowledged && let Err(error) = durable_lease.acknowledge_terminal()
{
warn!(
"Failed to persist acknowledged durable lease terminal state: {}",
error
);
}
exit_with_local_fallback(
&command,
&reporter,
"transfer skipped",
require_remote,
);
}
if classify_remote_pipeline_failure(&e)
== RemotePipelineFailurePolicy::FailClosedNoLocalFallback
{
warn!(
"Remote execution failed on {}; refusing local fallback: {:#}",
worker.id, e
);
// Issue #62: an unverified post-timeout cleanup means the
// orphaned remote group may still hold the project's Cargo
// target lock — quarantine the worker before exiting.
quarantine_worker_on_unverified_timeout_cleanup(
&e,
&config.general.socket_path,
&worker.id,
&reporter,
)
.await;
// Fail-closed refusal: the line explaining the non-zero exit must
// reach the agent even at stock visibility (rch#31).
reporter.summary_critical(&remote_pipeline_failure_summary(&worker.id, &e));
if release_acknowledged && let Err(error) = durable_lease.acknowledge_terminal()
{
warn!(
"Failed to persist acknowledged durable lease terminal state: {}",
error
);
}
emit_exec_envelope(&ExecResultEnvelope {
api_version: "1.0",
command: &remote_command,
outcome: "transport_error",
location: "remote",
fallback_reason: Some("fail_closed_no_local_fallback"),
worker_id: Some(worker.id.as_str()),
timing: exec_timing.as_ref(),
result_dirs: exec_dir_stats.as_deref(),
remote_exit_code: None,
duration_ms: Some(exec_start.elapsed().as_millis() as u64),
error_code: None,
});
std::process::exit(EXIT_BUILD_ERROR);
}
// Issue #59: a silence-detected source-sync stall is a typed
// worker-path fault naming the phase and worker. The
// reservation was already released above (release_worker runs
// unconditionally); falling through to the capacity-aware
// retry re-enters selection EXCLUDING this worker
// (tried_workers), mirroring the E104 failover semantics
// instead of dying inside the sync leg or falling straight to
// local.
if let Some(stall) = crate::transfer::find_source_sync_stall(&e) {
warn!(
"Source sync stalled on {} (phase {}, no output for {}s); failing over to another worker",
worker.id,
stall.phase,
stall.silence.as_secs()
);
reporter.summary(&format!(
"[RCH] source sync stalled on {} (phase {}, no output for {}s); retrying on another worker",
worker.id,
stall.phase,
stall.silence.as_secs()
));
RetryableRemoteFault {
log_reason: format!(
"source sync stalled (phase {}, {}s silence)",
stall.phase,
stall.silence.as_secs()
),
failed_worker: worker.id.clone(),
on_exhaust: RemoteFaultExhaustAction::GatedLocalFallback {
reason: format!(
"source sync stalled on {} (phase {})",
worker.id, stall.phase
),
error: None,
},
}
} else {
// Generic pipeline failure — retry on a different worker first.
warn!(
"Remote execution failed on {}: {:#}; will retry on another worker if available",
worker.id, e
);
RetryableRemoteFault {
log_reason: "remote execution failed".to_string(),
failed_worker: worker.id.clone(),
on_exhaust: RemoteFaultExhaustAction::GatedLocalFallback {
reason: "remote execution failed".to_string(),
error: Some(format!("{}: {e:#}", worker.id)),
},
}
}
}
};
// A retry must not overwrite the only recovery identity for an older
// source grant. This check precedes requesting another reservation.
durable_lease
.ensure_released_for_retry()
.context(crate::transfer::RemoteExecutionUnconfirmed)?;
// Worker-fault failure: try a bigger/different worker before going terminal.
if attempt < max_attempts
&& let Some((next_response, next_worker)) = try_retry_on_bigger_worker(
&config.general.socket_path,
&selection_project,
estimated_cores,
&remote_command,
toolchain.as_ref(),
required_runtime,
command_priority,
&tried_workers,
&preferred_workers,
Some(&wrapper_id),
&reporter,
)
.await
{
attempt += 1;
warn!(
"Remote build {} on {}; retrying on higher-capacity worker {} (attempt {}/{})",
fault.log_reason, fault.failed_worker, next_worker, attempt, max_attempts
);
reporter.summary(&format!(
"[RCH] retry on bigger worker {} (attempt {}/{}) after {} on {}",
next_worker, attempt, max_attempts, fault.log_reason, fault.failed_worker
));
current_query_preferred = vec![next_worker];
response = next_response;
continue;
}
// Retries exhausted (or no bigger worker available): terminal action.
match fault.on_exhaust {
RemoteFaultExhaustAction::ExitWithCode { code, summary } => {
warn!(
"Remote build failed and no higher-capacity worker was available after {} attempt(s); surfacing exit {}",
tried_workers.len(),
code
);
// Terminal non-zero exit: the line explaining it must reach the
// agent even at stock visibility (rch#31).
reporter.summary_critical(&summary);
emit_exec_envelope(&ExecResultEnvelope {
api_version: "1.0",
command: &remote_command,
outcome: "completed",
location: "remote",
fallback_reason: None,
worker_id: Some(fault.failed_worker.as_str()),
remote_exit_code: Some(code),
duration_ms: Some(exec_start.elapsed().as_millis() as u64),
timing: exec_timing.as_ref(),
result_dirs: exec_dir_stats.as_deref(),
error_code: None,
});
if release_acknowledged && let Err(error) = durable_lease.acknowledge_terminal() {
warn!(
"Failed to persist acknowledged durable lease terminal state: {}",
error
);
}
std::process::exit(code);
}
RemoteFaultExhaustAction::ExitWithEnvRemediation {
code,
summary,
remediation,
} => {
warn!(
"Remote build failed (worker environment) and no other worker was available after {} attempt(s); surfacing exit {}",
tried_workers.len(),
code
);
// Terminal non-zero exit: the line explaining it must reach the
// agent even at stock visibility (rch#31).
reporter.summary_critical(&summary);
reporter.verbose(&remediation);
if release_acknowledged && let Err(error) = durable_lease.acknowledge_terminal() {
warn!(
"Failed to persist acknowledged durable lease terminal state: {}",
error
);
}
emit_exec_envelope(&ExecResultEnvelope {
api_version: "1.0",
command: &remote_command,
outcome: "completed",
location: "remote",
fallback_reason: Some("worker_env_remediation"),
worker_id: Some(fault.failed_worker.as_str()),
remote_exit_code: Some(code),
duration_ms: Some(exec_start.elapsed().as_millis() as u64),
timing: exec_timing.as_ref(),
result_dirs: exec_dir_stats.as_deref(),
error_code: None,
});
std::process::exit(code);
}
RemoteFaultExhaustAction::GatedLocalFallback { reason, error } => {
// Issue #63(c): --job admission counts as strict-remote here.
let strict_remote = strict_remote_for_exhausted_admission(require_remote, job);
warn!(
"Remote build failed on all {} tried worker(s); {}",
tried_workers.len(),
if allow_local_fallback && !strict_remote {
"falling back to local"
} else {
"refusing local fallback"
}
);
let mut reason = format!("{reason}; remote retries exhausted");
if job && !require_remote {
reason = format!("job admission exhausted: {reason}");
}
if release_acknowledged && let Err(error) = durable_lease.acknowledge_terminal() {
warn!(
"Failed to persist acknowledged durable lease terminal state: {}",
error
);
}
exit_with_gated_local_fallback(
&command,
&reporter,
&reason,
strict_remote,
allow_local_fallback,
error.as_deref(),
);
}
}
}
}
#[derive(Clone, Copy)]
struct HookReporter {
visibility: OutputVisibility,
}
impl HookReporter {
fn new(visibility: OutputVisibility) -> Self {
Self { visibility }
}
fn summary(&self, message: &str) {
if self.visibility != OutputVisibility::None {
eprintln!("{}", message);
}
}
/// Emit a message that MUST reach the operator/agent regardless of the
/// configured output visibility. Reserved for the one line explaining a
/// non-zero exit — a fail-closed refusal or a hard error — which visibility
/// settings (progress/celebration chrome) must never suppress. Same
/// always-on-stderr contract as the RCH-E309 artifact-transfer diagnostic
/// (rch#31).
fn summary_critical(&self, message: &str) {
eprintln!("{}", message);
}
fn verbose(&self, message: &str) {
if self.visibility == OutputVisibility::Verbose {
eprintln!("{}", message);
}
}
}
// ============================================================================
// Daemon Auto-Start (Self-Healing)
// ============================================================================
//
// The bounded daemon-autostart cluster (lock/cooldown/spawn/health-probe/
// socket-wait) lives in the `auto_start` submodule. `try_auto_start_daemon`
// is its only cross-module entry point (called from `run_exec` below).
mod auto_start;
// The build-heartbeat / progress-reporting cluster (the periodic snapshot, the
// background loop, the progress-counter bump, and the socket send) lives in the
// `progress_reporting` submodule. Its `BuildHeartbeatLoop` /
// `mark_heartbeat_progress` are consumed by `execute_remote_compilation`, which
// now lives in the sibling `transfer_orchestration` submodule and imports them
// directly.
mod progress_reporting;
// The remote-build execution pipeline (`execute_remote_compilation` plus its leaf
// telemetry-forwarding helpers) lives in the `transfer_orchestration` submodule.
// `execute_remote_compilation` is imported so `run_hook` / `run_exec` call it
// unqualified.
mod transfer_orchestration;
use transfer_orchestration::execute_remote_compilation;
pub(crate) use transfer_orchestration::recovery::recover_job;
// Exact source-byte manifest construction and worker-side re-verification for
// `rch exec --source-content-receipt` proof runs.
mod repo_updater;
// The offload-pipeline SSH primitives (`run_offload_ssh_command`, the remote
// topology-enforcement preflight, and the mock-mode skip gate) live in the
// `ssh` submodule. They are consumed only by the sibling submodules
// (`dependency_closure`, `transfer_orchestration`, `repo_updater`) import what
// they need directly from `super::ssh`, and the doctor's mirror-ownership probe
// consumes the same items via `crate::hook::ssh`, so the module stays
// crate-internal (`pub(crate)`) rather than fully private.
pub(crate) mod ssh;
pub(crate) use ssh::{source_authority_activity_prefix, source_authority_cleanup_prefix};
/// Worker setup can change a shared ancestor of many source trees. Give each
/// explicit mutation the same durable exclusion and activity draining as a
/// build, including when SSH stops reporting before the command finishes.
pub(crate) async fn run_owned_worker_topology_command(
worker: &WorkerConfig,
canonical_root: &Path,
alias_root: &Path,
command: &str,
) -> anyhow::Result<Output> {
let mut roots = [canonical_root, alias_root]
.into_iter()
.map(|root| {
root.to_str()
.map(str::to_owned)
.context("worker topology root is not a UTF-8 path")
})
.collect::<anyhow::Result<Vec<_>>>()?;
roots.sort();
roots.dedup();
let identity = uuid::Uuid::new_v4().simple().to_string();
let activity_command = ssh::wrap_remote_source_activity(command, &identity)?;
let guard = match ssh::acquire_remote_source_authority_lock(
worker,
&roots,
None,
&identity,
Duration::from_secs(30),
)
.await
{
Ok(guard) => guard,
Err(acquisition_error) => {
// No command was dispatched, but a delayed holder may still gain
// its kernel locks. Fence that exact intent before reporting that
// setup failed; a missing acknowledgment is not absent ownership.
let cancellation = async {
if ssh::cancel_remote_source_authority_intent(worker, &roots, &identity).await? {
ssh::finish_cancel_remote_source_authority_intent(worker, &roots, &identity)
.await?;
}
Ok::<(), anyhow::Error>(())
}
.await;
return match cancellation {
Ok(()) => Err(acquisition_error.context("worker topology setup was not started")),
Err(cancellation_error) => Err(acquisition_error.context(format!(
"worker topology intent {identity} on {} remains unresolved; source ownership is retained: {cancellation_error:#}",
worker.id,
))),
};
}
};
let result =
ssh::run_offload_ssh_command(worker, &activity_command, Duration::from_secs(30)).await;
// Release waits for the activity supervisor even after a transport error.
// A command arriving later must recheck the now-released token and refuse.
// If draining cannot be proved, keep the durable claim and name it instead
// of allowing a later build to share a changing topology.
guard.release().await.with_context(|| {
format!(
"worker topology intent {identity} on {} could not acknowledge release; inspect the retained source claim before retrying setup",
worker.id,
)
})?;
result
}
// The dependency-closure sync planning + remote dependency-preflight cluster
// (sync-closure plan/manifest, sync-topology predicates, cargo manifest/workspace
// parsers, and the remote dependency-manifest verifier) lives in the
// `dependency_closure` submodule. The dependency-preflight types/consts below are
// consumed by `build_dependency_runtime_fail_open_report` and the `run_hook` /
// `run_exec` error downcasts; the sibling `transfer_orchestration` imports the
// sync-closure planners + verifier directly from `super::dependency_closure`.
mod dependency_closure;
use dependency_closure::{
DEPENDENCY_PREFLIGHT_CODE_MATERIALIZATION, DEPENDENCY_PREFLIGHT_CODE_POLICY,
DEPENDENCY_PREFLIGHT_CODE_TIMEOUT, DEPENDENCY_PREFLIGHT_CODE_UNKNOWN,
DEPENDENCY_PREFLIGHT_REMEDIATION_MATERIALIZATION, DEPENDENCY_PREFLIGHT_REMEDIATION_POLICY,
DEPENDENCY_PREFLIGHT_REMEDIATION_TIMEOUT, DEPENDENCY_PREFLIGHT_REMEDIATION_UNKNOWN,
DEPENDENCY_PREFLIGHT_SCHEMA_VERSION, DependencyPreflightEvidence, DependencyPreflightFailure,
DependencyPreflightReport, DependencyPreflightStatus,
};
// Exact source-byte manifest construction and worker-side re-verification for
// `rch exec --source-content-receipt` proof runs.
mod source_fidelity;
// The remote-execution result type (`RemoteExecutionResult`) and the outcome
// classifiers that interpret it live in the `remote_result` submodule. The four
// classifier fns below are consumed by `run_hook` / `run_exec`; the sibling
// `transfer_orchestration` constructs and returns `RemoteExecutionResult`
// directly from `super::remote_result`.
mod remote_result;
use remote_result::{
ExecResultDirStat, ExecResultEnvelope, detect_cargo_workspace_inheritance_failure,
detect_worker_system_dependency_failure, emit_exec_envelope, is_cpu_capability_signal,
is_signal_killed, is_toolchain_failure, remote_failure_is_worker_fault, set_machine_output,
signal_name, wrapped_cpu_capability_signal,
};
// The remote cargo target-dir resolution / naming / command-rewrite cluster
// (CARGO_TARGET_DIR forwarding, the unique-per-job + stable-pooled remote dir
// names, and the helpers that strip a local target-dir from a delegated command)
// lives in the `cargo_target_dir` submodule. `run_hook` / `run_exec` call
// `resolve_forwarded_cargo_target_dir` + `rewrite_cargo_target_dir_command_for_remote`,
// and `add_cargo_isolation` shares `sanitize_cargo_home_token`, so those three are
// imported here; the sibling `transfer_orchestration` imports the dir-naming / env
// overrides directly from `super::cargo_target_dir`.
mod cargo_target_dir;
use cargo_target_dir::{
resolve_forwarded_cargo_target_dir, rewrite_cargo_target_dir_command_for_remote,
sanitize_cargo_home_token,
};
// The remote artifact-pattern selection cluster (which files travel back from a
// worker, keyed on `CompilationKind` plus the command's cargo `--profile`
// selection, and the zero-build-output sync-back detector) lives in the
// `artifact_patterns` submodule. `get_artifact_patterns` /
// `get_custom_target_artifact_patterns` /
// `kind_produces_transferable_artifacts` have no non-test caller in `hook`
// itself — they are consumed by the sibling `transfer_orchestration`
// (`execute_remote_compilation`), which imports them directly, so nothing is
// re-exported into the non-test hook namespace here.
mod artifact_patterns;
// The retrieved-artifact executable-typing gate (GitHub #65) lives in the
// `artifact_triple` submodule: it types the files a successful sync-back placed
// in the local target tree against the triple the caller's build was for, so an
// unpinned offload to a foreign-platform worker can never leave an unrunnable
// binary behind a green existence check. Its entry points are consumed by the
// sibling `transfer_orchestration`, which imports them directly.
mod artifact_triple;
// The daemon selection-response wire-deserialization cluster (the `*Wire` DTOs,
// their `From` conversions into the `rch_common` domain types, and the
// protocol-version-checked parse entry point) lives in the `selection_response`
// submodule. `parse_selection_response` is the only cross-module item —
// `run_hook` / `run_exec` call it — so it is re-exported here; the wire types and
// validation helpers stay private to that submodule.
mod selection_response;
use selection_response::parse_selection_response;
// Build-timing history (persistence + offload-gating estimation) lives in the
// `timing_history` submodule. `record_build_timing` is the only item the hook
// hot path calls (two sites in the remote-classification path), so it is
// re-exported here; the on-disk model, the process-global cache, and the
// estimator surface stay `pub(super)` for the test suite and otherwise private.
mod timing_history;
use timing_history::record_build_timing;
// RABS B002: redacted invocation recorder — every completed offloaded
// build emits one B001 corpus record, off the SLO path, fail-open.
mod rabs_recorder;
use rabs_recorder::record_invocation;
// The daemon IPC client (worker-selection / release / build-record requests
// over the `rchd` Unix socket, plus request-timeout + queue-when-busy policy
// helpers) lives in the `daemon_ipc` submodule. `query_daemon` / `release_worker`
// are re-exported `pub(crate)` because `commands::status` also calls them;
// `record_build` / `queue_when_busy_enabled` are re-exported for the hook hot
// path. The timeout helpers and `urlencoding_encode` stay `pub(super)` for tests.
mod daemon_ipc;
use daemon_ipc::{
disable_worker_for_fault, queue_when_busy_enabled, record_build, release_worker_with_fault,
};
pub(crate) use daemon_ipc::{
query_daemon, query_daemon_dry_run, release_worker, restart_admission_is_closed,
};
// Command-string parsing utilities (tokenization + cargo flag/env analyzers +
// offload core estimation) live in the `command_parsing` submodule.
// `cargo_job_count_for_command` / `estimate_cores_for_command` are re-exported
// `pub(crate)` because `commands::status` also calls them; the `--test-threads` /
// `-j` / `--ignored` / `--exact` / filtered-test detectors stay `pub(super)` for
// the test suite, and the numeric `parse_*` helpers stay module-private.
mod command_parsing;
pub(crate) use command_parsing::{
cargo_job_count_for_command, estimate_cores_for_command, extract_project_name,
extract_project_name_with_policy, preferred_workers,
};
// Human-facing job-output rendering (compile-summary panel, job banner, and the
// duration/speed/profile/target formatting + detection helpers) lives in the
// `formatting` submodule. `format_duration_ms` / `estimate_local_time_ms` are
// re-exported for the hook hot path; `emit_job_banner` / `render_compile_summary`
// / `cache_hit` / `detect_target_label` are pub(super) and imported directly by
// the sibling transfer-orchestration (and cargo_target_dir) modules.
mod formatting;
use formatting::{estimate_local_time_ms, format_duration_ms};
// Capacity-aware worker selection for the remote-failure fallback loop (FIX 1):
// on a worker-fault build failure `run_exec` retries on a bigger / higher-RAM
// worker rather than flooding the local orchestrator. The pure ranking lives in
// this submodule so it is unit-testable without a live daemon.
mod fallback_selection;
use fallback_selection::{build_capacity_snapshots, pick_bigger_worker};
fn is_test_kind(kind: Option<CompilationKind>) -> bool {
matches!(
kind,
Some(CompilationKind::CargoTest | CompilationKind::CargoNextest | CompilationKind::BunTest)
)
}
#[allow(dead_code)]
fn emit_first_run_message(worker: &SelectedWorker, remote_ms: u64, local_ms: Option<u64>) {
let divider = "----------------------------------------";
let remote = format_duration_ms(Duration::from_millis(remote_ms));
eprintln!();
eprintln!("{}", divider);
eprintln!("First remote build complete!");
eprintln!();
if let Some(local_ms) = local_ms {
let local = format_duration_ms(Duration::from_millis(local_ms));
eprintln!(
"Your build ran on '{}' in {} (local estimate ~{}).",
worker.id, remote, local
);
} else {
eprintln!("Your build ran on '{}' in {}.", worker.id, remote);
}
eprintln!("RCH will run silently in the background from now on.");
eprintln!();
eprintln!("To see build activity: rch status --jobs");
eprintln!("To disable this message: rch config set first_run_complete true");
eprintln!("{}", divider);
eprintln!();
}
/// Process a hook request and return the output.
async fn process_hook(input: HookInput) -> HookOutput {
// Tier 0: Only process Bash tool
if input.tool_name != "Bash" {
debug!("Non-Bash tool: {}, allowing", input.tool_name);
return HookOutput::allow();
}
let command = &input.tool_input.command;
// Mask sensitive data in debug logs (API keys, tokens, passwords)
debug!("Processing command: {}", mask_sensitive_command(command));
// Classify the command using the 5-tier system.
// Per AGENTS.md: non-compilation decisions must complete in <1ms, compilation in <5ms
// The real hook path bypasses the classification cache because hook
// invocations are one-shot even when RCH_HOOK_MODE is not set.
let classify_start = Instant::now();
let classification = crate::cache::classify_hook_command(command, classify_command);
let classification_duration = classify_start.elapsed();
let classification_duration_us = classification_duration.as_micros() as u64;
if !classification.is_compilation {
// Log non-compilation decision latency (budget: <1ms per AGENTS.md)
let duration_ms = classification_duration_us as f64 / 1000.0;
if duration_ms > 1.0 {
warn!(
"Non-compilation decision exceeded 1ms budget: {:.3}ms for '{}'",
duration_ms, command
);
} else {
debug!(
"Non-compilation decision: {:.3}ms for '{}' ({})",
duration_ms, command, classification.reason
);
}
// Issue #24, item 3: surface a hint when we decline a command that is a
// compilation command in disguise but whose pipe/redirect/background/
// subshell structure we can't safely offload. On an orchestrator with
// force_remote=true this would otherwise be an *invisible* local
// fallback (a silent rustc/cc storm). Only loads config on this rare
// path to avoid touching the hot non-compilation path.
if let Some(structure_reason) = declined_compilation_due_to_structure(command) {
let config = load_config().ok();
let force_remote = config
.as_ref()
.map(|cfg| cfg.general.force_remote)
.unwrap_or(false);
// Issue #50: `cargo build ; cargo test` and `cargo build || ...`
// compile locally. That is deliberate, but it must be VISIBLE —
// a build the agent believes was offloaded is the exact failure
// the `[RCH]` summary contract exists to prevent.
let visibility = config
.as_ref()
.map(|cfg| cfg.output.visibility)
.unwrap_or(OutputVisibility::Summary);
HookReporter::new(visibility).summary(&format!("[RCH] local ({structure_reason})"));
if force_remote {
warn!(
"⚠️ RCH: declined to offload compilation command due to shell structure \
({structure_reason}) while force_remote=true — running LOCALLY: '{command}'. \
Run it as a bare command (no unsupported pipe/subshell) so it can be offloaded."
);
} else {
debug!(
"RCH: declined to offload compilation command due to shell structure \
({structure_reason}): '{command}'"
);
}
}
return HookOutput::allow();
}
let config = match load_config() {
Ok(cfg) => cfg,
Err(e) => {
warn!("Failed to load config: {}, allowing local execution", e);
return HookOutput::allow();
}
};
let reporter = HookReporter::new(config.output.visibility);
if !config.general.enabled {
debug!("RCH disabled via config, allowing local execution");
return HookOutput::allow();
}
// Per-project overrides (bd-1vzb)
//
// - force_local: always allow local execution for compilation commands (skip daemon + transfer)
// - force_remote: always attempt remote execution when safe (bypass confidence threshold)
//
// Conflicting flags should be caught by config validation, but handle defensively here.
if config.general.force_local && config.general.force_remote {
warn!(
"Invalid config: both general.force_local and general.force_remote are set; allowing local execution"
);
reporter.summary("[RCH] local (invalid config: force_local+force_remote)");
return HookOutput::allow();
}
if config.general.force_local {
debug!("RCH force_local enabled, allowing local execution");
reporter.summary("[RCH] local (force_local)");
return HookOutput::allow();
}
// Log compilation decision latency (budget: <5ms per AGENTS.md)
let duration_ms = classification_duration_us as f64 / 1000.0;
if duration_ms > 5.0 {
warn!(
"Compilation decision exceeded 5ms budget: {:.3}ms",
duration_ms
);
}
info!(
"Compilation detected: {:?} (confidence: {:.2}, classified in {:.3}ms)",
classification.kind, classification.confidence, duration_ms
);
reporter.verbose(&format!(
"[RCH] compile {:?} (confidence {:.2})",
classification.kind, classification.confidence
));
// Check confidence threshold
let confidence_threshold = if config.general.force_remote {
reporter.verbose("[RCH] force_remote enabled: bypassing confidence threshold");
0.0
} else {
config.compilation.confidence_threshold
};
if classification.confidence < confidence_threshold {
debug!(
"Confidence {:.2} below threshold {:.2}, allowing local execution",
classification.confidence, confidence_threshold
);
reporter.summary("[RCH] local (confidence below threshold)");
return HookOutput::allow();
}
// Check execution allowlist (bd-785w)
// Commands not in the allowlist fail-open to local execution
if let Some(kind) = classification.kind {
let command_base = kind.command_base();
if !config.execution.is_allowed(command_base) {
debug!(
"Command base '{}' not in execution allowlist, allowing local execution",
command_base
);
reporter.summary(&format!(
"[RCH] local (command '{}' not in execution.allowlist)",
command_base
));
return HookOutput::allow();
}
}
// CRITICAL: Return immediately with delegated command to avoid hook timeout.
//
// Claude Code hooks have a tight timeout budget (~50-100ms). The full remote
// compilation pipeline (daemon query + rsync + SSH + rsync back) takes 3+ seconds.
// If we do that work here, the hook times out and Claude Code ignores our response.
//
// Solution: Return immediately with `rch exec -- <command>`. The hook completes
// in <10ms, and the actual remote compilation happens when Claude Code executes
// the modified command.
//
// For compound commands like "cd /path && cargo build", we preserve the prefix
// and only wrap the compilation part: "cd /path && rch exec -- cargo build"
info!(
"Delegating compilation to rch exec (classification: {:?}, compound: {})",
classification.kind,
classification.command_prefix.is_some()
);
reporter.verbose("[RCH] delegating to rch exec...");
let modified_command = if let (Some(prefix), Some(extracted)) = (
&classification.command_prefix,
&classification.extracted_command,
) {
// Compound command: the classifier has already wrapped every earlier
// compilation segment inside `prefix` (issue #50); wrap the final one.
// Build-looking segments it could NOT offload stay local — say so.
for segment in rch_common::compound_local_compilation_segments(command) {
reporter.summary(&format!(
"[RCH] local (compound segment not offloadable: '{segment}')"
));
}
format!("{}rch exec -- {}", prefix, extracted)
} else {
// Simple command: wrap the entire command
format!("rch exec -- {}", command)
};
// A compound rewrite keeps the user's own prefix; auto-approving it would
// bypass the user's permission rules for that prefix (bd-08ele). A simple
// `rch exec -- <command>` stays approved: offloading it is rch's purpose.
if classification.command_prefix.is_some() {
HookOutput::rewrite_with_permission_check(modified_command)
} else {
HookOutput::allow_with_modified_command(modified_command)
}
}
fn selection_cancelled_before_start(response: &SelectionResponse) -> bool {
response.worker.is_none()
&& response.build_id.is_none()
&& matches!(&response.reason, SelectionReason::SelectionError(reason) if reason == "job_cancelled_before_start")
}
#[allow(dead_code)]
#[allow(clippy::too_many_arguments)] // Pipeline wiring favors explicit params.
async fn handle_selection_response(
response: SelectionResponse,
command: &str,
config: &rch_common::RchConfig,
reporter: &HookReporter,
toolchain: Option<&ToolchainInfo>,
classification_kind: Option<CompilationKind>,
project: &str,
estimated_cores: u32,
) -> HookOutput {
if selection_cancelled_before_start(&response) {
reporter.summary("[RCH] cancelled before remote admission");
return HookOutput::allow_with_modified_command("exit 130".to_owned());
}
// Check if a worker was assigned
let Some(worker) = response.worker else {
// No worker available - graceful fallback to local execution
warn!(
"⚠️ RCH: No remote workers available ({}), executing locally",
response.reason
);
reporter.summary(&format!("[RCH] local ({})", response.reason));
return HookOutput::allow();
};
info!(
"Selected worker: {} at {}@{} ({} slots remaining after reservation, speed {:.1})",
worker.id, worker.user, worker.host, worker.slots_available, worker.speed_score
);
reporter.verbose(&format!(
"[RCH] selected {}@{} ({} slots remaining after reservation, speed {:.1})",
worker.user, worker.host, worker.slots_available, worker.speed_score
));
let invocation_cwd = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
let command_tokens = parse_command_tokens(command, reporter);
let forwarded_cargo_target_dir = resolve_forwarded_cargo_target_dir(
classification_kind,
&invocation_cwd,
reporter,
command_tokens.as_deref(),
);
let remote_command = rewrite_cargo_target_dir_command_for_remote(
command,
command_tokens.as_deref(),
forwarded_cargo_target_dir.as_ref(),
reporter,
);
// Execute remote compilation pipeline
let topology_policy = config.path_topology.to_policy();
let remote_start = Instant::now();
let result = execute_remote_compilation(
&worker,
&remote_command,
config.transfer.clone(),
config.environment.allowlist.clone(),
forwarded_cargo_target_dir,
&config.compilation,
toolchain,
classification_kind,
reporter,
&config.general.socket_path,
config.output.color_mode,
response.build_id,
None,
None,
&topology_policy,
None,
false,
&[],
&[],
config.remediation.pooled_target.reaper_pooled_idle_hours,
config.remediation.pooled_target.store_base.as_deref(),
)
.await;
let remote_elapsed = remote_start.elapsed();
// Ownership/receipt loss after execution is not a terminal completion.
let retain_unconfirmed_ownership = result
.as_ref()
.err()
.is_some_and(is_remote_execution_unconfirmed);
let release_exit_code = result
.as_ref()
.map(|ok| ok.exit_code)
.unwrap_or(EXIT_BUILD_ERROR);
// Add total elapsed time to the timing breakdown
let release_timing = result.as_ref().ok().map(|ok| {
let mut timing = ok.timing.clone();
timing.total = Some(remote_elapsed);
timing
});
if retain_unconfirmed_ownership {
warn!(
"Remote completion unconfirmed; retaining worker {} ownership",
worker.id
);
} else if let Err(e) = release_worker(
&config.general.socket_path,
&worker.id,
estimated_cores,
response.build_id,
Some(release_exit_code),
None,
None,
release_timing.as_ref(),
None,
)
.await
{
warn!("Failed to release worker slots: {}", e);
}
match result {
Ok(result) => {
// RABS B002: every COMPLETED offloaded invocation (success,
// test failure, build error alike) joins the corpus. Off the
// SLO path via spawn_blocking; fail-open inside.
{
let command_for_corpus = command.to_string();
let cwd_for_corpus = invocation_cwd.clone();
let exit_code = result.exit_code;
let duration_ms = result.duration_ms;
tokio::task::spawn_blocking(move || {
record_invocation(
&command_for_corpus,
&cwd_for_corpus,
classification_kind,
exit_code,
duration_ms,
);
});
}
if result.exit_code == 0 {
// Command succeeded remotely - replace with no-op for transparency
// The agent already saw output via stderr, artifacts are local
// Using allow+modified_command makes this completely transparent to the agent
info!("Remote compilation succeeded, replacing with no-op for transparency");
reporter.summary(&format!(
"[RCH] remote {} ({})",
worker.id,
format_duration_ms(remote_elapsed)
));
// Record successful build for cache affinity
let is_test = classification_kind
.map(|kind| kind.is_test_command())
.unwrap_or(false);
if let Err(e) =
record_build(&config.general.socket_path, &worker.id, project, is_test).await
{
warn!("Failed to record build: {}", e);
}
// Record timing for future gating decisions (bd-mnhp: spawn_blocking for file I/O)
let project_for_timing = project.to_string();
let duration = result.duration_ms;
tokio::task::spawn_blocking(move || {
record_build_timing(&project_for_timing, classification_kind, duration, true);
});
if !config.output.first_run_complete {
let local_estimate =
estimate_local_time_ms(result.duration_ms, worker.speed_score);
emit_first_run_message(&worker, result.duration_ms, local_estimate);
if let Err(e) = crate::config::set_first_run_complete(true) {
warn!("Failed to persist first_run_complete: {}", e);
}
}
// Replace original command with a no-op - agent thinks command ran locally
HookOutput::allow_with_modified_command("true")
} else if result.deadline_triggered {
reporter.summary(&format!(
"[RCH] remote {} exceeded its configured deadline (exit {}); not retrying",
worker.id, result.exit_code
));
HookOutput::deny(format!(
"Remote compilation exceeded its configured deadline (exit {}). Do not rerun locally; adjust the deadline before retrying.",
result.exit_code
))
} else if is_toolchain_failure(&result.stderr, result.exit_code) {
// Toolchain failure - fall back to local execution
warn!(
"Remote toolchain failure detected (exit {}), falling back to local",
result.exit_code
);
reporter.summary(&format!("[RCH] local (toolchain missing on {})", worker.id));
HookOutput::allow()
} else {
// Command failed remotely - still deny to prevent re-execution
// The agent saw the error output via stderr
//
// Exit code semantics:
// - 101: Test failures (cargo test ran but tests failed)
// - 1: Build/compilation error
// - 128+N: Process killed by signal N
let exit_code = result.exit_code;
// Check for topology-specific Cargo workspace inheritance before
// falling back to generic build/test failure text.
if let Some(workspace_failure) =
detect_cargo_workspace_inheritance_failure(&result.stderr, exit_code)
{
let error = ErrorCode::BuildCargoWorkspaceInheritance;
warn!(
"Remote Cargo workspace-inheritance failure on {} [{}]: {}",
worker.id,
error.code_string(),
workspace_failure.log_detail()
);
reporter.summary(&format!(
"[RCH] remote {} failed [{}] {}",
worker.id,
error.code_string(),
workspace_failure.summary()
));
reporter.verbose(&format!(
"[RCH] remediation [{}]: {}",
error.code_string(),
workspace_failure.remediation()
));
} else if let Some(signal) = is_signal_killed(exit_code) {
if is_cpu_capability_signal(signal) {
// bd-68hon: SIGILL on the hook path — the worker's
// CPU cannot execute this build's codegen. Quarantine
// it so later builds auto-route around it (the hook
// path has no retry loop; transparency still returns
// the exit code to the agent for THIS build).
warn!(
"Remote command killed by SIGILL on {} — worker CPU cannot execute build (likely missing ISA, e.g. AVX2); quarantining worker",
worker.id
);
if let Err(e) = disable_worker_for_fault(
&config.general.socket_path,
&worker.id,
"cpu-capability-fault (SIGILL) — likely missing ISA (e.g. AVX2)",
)
.await
{
warn!("Failed to quarantine SIGILL worker {}: {}", worker.id, e);
}
} else {
warn!(
"Remote command killed by signal {} ({}) on {}, replacing with exit code for transparency",
signal,
signal_name(signal),
worker.id
);
}
reporter.summary(&format!(
"[RCH] remote {} killed ({})",
worker.id,
signal_name(signal)
));
} else if let Some(signal) =
wrapped_cpu_capability_signal(exit_code, &result.stderr)
{
// bd-68hon (wrapped shape): a build-script/proc-macro
// SIGILL surfaces from cargo as exit 101 with the signal
// named only in the diagnostics, so it must be sniffed
// BEFORE the generic exit-101 arm below. Quarantine the
// worker so later builds route around it; exit-code
// transparency for THIS build is preserved (no retry loop
// on the hook path).
warn!(
"Remote command's build-script/proc-macro killed by {} on {} (cargo exit {}) — worker CPU cannot execute build (likely missing ISA, e.g. AVX2); quarantining worker",
signal_name(signal),
worker.id,
exit_code
);
if let Err(e) = disable_worker_for_fault(
&config.general.socket_path,
&worker.id,
"cpu-capability-fault (build-script SIGILL) — likely missing ISA (e.g. AVX2)",
)
.await
{
warn!("Failed to quarantine SIGILL worker {}: {}", worker.id, e);
}
reporter.summary(&format!(
"[RCH] remote {} build script killed ({} — CPU capability)",
worker.id,
signal_name(signal)
));
} else if exit_code == EXIT_TEST_FAILURES {
// Cargo test exit 101: tests ran but some failed
info!(
"Remote tests failed (exit 101) on {}, replacing with exit code for transparency",
worker.id
);
reporter.summary(&format!("[RCH] remote {} tests failed", worker.id));
} else if exit_code == EXIT_BUILD_ERROR {
// Build/compilation error
info!(
"Remote build error (exit 1) on {}, replacing with exit code for transparency",
worker.id
);
reporter.summary(&format!("[RCH] remote {} build error", worker.id));
} else {
// Other non-zero exit code
info!(
"Remote command failed (exit {}) on {}, replacing with exit code for transparency",
exit_code, worker.id
);
reporter.summary(&format!(
"[RCH] remote {} failed (exit {})",
worker.id, exit_code
));
}
// Still record timing for failed builds (useful for predictions)
// bd-mnhp: spawn_blocking for file I/O
let project_for_timing = project.to_string();
let duration = result.duration_ms;
tokio::task::spawn_blocking(move || {
record_build_timing(&project_for_timing, classification_kind, duration, true);
});
// Replace with exit command to preserve the exit code transparently
// Agent already saw the error output, now they see the correct exit code
HookOutput::allow_with_modified_command(format!("exit {}", exit_code))
}
}
Err(e) => {
if let Some(preflight_err) = e.downcast_ref::<DependencyPreflightFailure>() {
let evidence_summary = preflight_err.evidence_summary();
info!(
"Dependency preflight blocked remote execution [{}], falling back to local; evidence='{}'",
preflight_err.reason_code, evidence_summary
);
reporter.summary(&format!(
"[RCH] local (dependency preflight {}: {}; evidence: {})",
preflight_err.reason_code, preflight_err.remediation, evidence_summary
));
reporter.verbose(&format!(
"[RCH] dependency preflight report: {}",
preflight_err.report_json()
));
return HookOutput::allow();
}
// Check if this is a transfer skip (not a failure, just too large/slow)
if let Some(skip_err) = e.downcast_ref::<TransferError>()
&& let TransferError::TransferSkipped { reason } = skip_err
{
info!(
"Transfer skipped ({}), falling back to local execution",
reason
);
reporter.summary(&format!("[RCH] local ({})", reason));
return HookOutput::allow();
}
if classify_remote_pipeline_failure(&e)
== RemotePipelineFailurePolicy::FailClosedNoLocalFallback
{
warn!(
"Remote execution pipeline failed on {}; refusing local fallback: {:#}",
worker.id, e
);
// Issue #62: quarantine the worker when the post-timeout
// remote cleanup could not be verified (possible orphan
// holding the project's Cargo target lock).
quarantine_worker_on_unverified_timeout_cleanup(
&e,
&config.general.socket_path,
&worker.id,
reporter,
)
.await;
reporter.summary_critical(&remote_pipeline_failure_summary(&worker.id, &e));
return HookOutput::allow_with_modified_command(format!(
"exit {}",
EXIT_BUILD_ERROR
));
}
// Pipeline failed - fall back to local execution
warn!(
"Remote execution pipeline failed: {}, falling back to local",
e
);
// Issue #59: name the stalled phase and worker instead of the
// generic line, so a silent sync stall is attributable from the
// one summary line hook mode emits.
if let Some(stall) = crate::transfer::find_source_sync_stall(&e) {
reporter.summary(&format!(
"[RCH] local (source sync stalled on {} — phase {}, no output for {}s)",
worker.id,
stall.phase,
stall.silence.as_secs()
));
} else {
reporter.summary("[RCH] local (remote pipeline failed)");
}
HookOutput::allow()
}
}
}
fn command_priority_from_env(reporter: &HookReporter) -> CommandPriority {
let Ok(raw) = std::env::var("RCH_PRIORITY") else {
return CommandPriority::Normal;
};
match raw.parse::<CommandPriority>() {
Ok(value) => value,
Err(()) => {
reporter.verbose(&format!(
"[RCH] ignoring invalid RCH_PRIORITY={:?} (expected: low|normal|high)",
raw
));
CommandPriority::Normal
}
}
}
/// Convert a SelectedWorker to a WorkerConfig.
fn selected_worker_to_config(worker: &SelectedWorker) -> WorkerConfig {
// Rebuild the reserved `os:<os>` tag from the daemon-propagated declared OS
// so `WorkerPlatform::from_worker` can pick the Windows transport. Historical
// workers report no OS and keep the empty (Posix) tag set.
let tags = worker
.declared_os
.as_deref()
.map(|os| vec![rch_common::os_tag(os)])
.unwrap_or_default();
WorkerConfig {
id: worker.id.clone(),
host: worker.host.clone(),
user: worker.user.clone(),
identity_file: worker.identity_file.clone(),
total_slots: worker.slots_available,
priority: 100,
tags,
tools: Vec::new(),
}
}
#[derive(Debug, Clone)]
struct DependencyRuntimePlan {
sync_roots: Vec<PathBuf>,
fail_open_decision: Option<DependencyRuntimeFailOpenDecision>,
}
#[derive(Debug, Clone)]
struct DependencyRuntimeFailOpenDecision {
reason_code: &'static str,
remediation: &'static str,
detail: String,
}
fn text_indicates_timeout(value: &str) -> bool {
let lower = value.to_ascii_lowercase();
lower.contains("timeout") || lower.contains("timed out")
}
fn classify_dependency_runtime_fail_open(
plan: &DependencyClosurePlan,
) -> DependencyRuntimeFailOpenDecision {
let has_policy_violation = plan
.issues
.iter()
.any(|issue| issue.code == "path-policy-violation");
let has_materialization_failure = plan
.issues
.iter()
.any(|issue| issue.code == "materialization-closure-unavailable");
let has_timeout = plan
.fail_open_reason
.as_deref()
.is_some_and(text_indicates_timeout)
|| plan.issues.iter().any(|issue| {
text_indicates_timeout(&issue.message)
|| issue
.diagnostics
.iter()
.any(|diag| text_indicates_timeout(diag))
});
let (reason_code, remediation) = if has_policy_violation {
(
DEPENDENCY_PREFLIGHT_CODE_POLICY,
DEPENDENCY_PREFLIGHT_REMEDIATION_POLICY,
)
} else if has_materialization_failure {
(
DEPENDENCY_PREFLIGHT_CODE_MATERIALIZATION,
DEPENDENCY_PREFLIGHT_REMEDIATION_MATERIALIZATION,
)
} else if has_timeout {
(
DEPENDENCY_PREFLIGHT_CODE_TIMEOUT,
DEPENDENCY_PREFLIGHT_REMEDIATION_TIMEOUT,
)
} else {
(
DEPENDENCY_PREFLIGHT_CODE_UNKNOWN,
DEPENDENCY_PREFLIGHT_REMEDIATION_UNKNOWN,
)
};
let issue_codes = if plan.issues.is_empty() {
"none".to_string()
} else {
plan.issues
.iter()
.map(|issue| issue.code.clone())
.collect::<Vec<_>>()
.join(",")
};
let fail_open_reason = plan
.fail_open_reason
.as_deref()
.unwrap_or("no planner fail-open reason supplied");
let detail = format!("planner fail-open reason={fail_open_reason}; issue_codes={issue_codes}");
DependencyRuntimeFailOpenDecision {
reason_code,
remediation,
detail,
}
}
fn build_dependency_runtime_fail_open_report(
worker: &WorkerConfig,
normalized_project_root: &Path,
decision: &DependencyRuntimeFailOpenDecision,
) -> DependencyPreflightReport {
let status = if decision.reason_code == DEPENDENCY_PREFLIGHT_CODE_POLICY {
DependencyPreflightStatus::PolicyViolation
} else if decision.reason_code == DEPENDENCY_PREFLIGHT_CODE_MATERIALIZATION {
DependencyPreflightStatus::MaterializationUnavailable
} else if decision.reason_code == DEPENDENCY_PREFLIGHT_CODE_TIMEOUT {
DependencyPreflightStatus::Timeout
} else {
DependencyPreflightStatus::Unknown
};
DependencyPreflightReport {
schema_version: DEPENDENCY_PREFLIGHT_SCHEMA_VERSION,
worker: worker.id.as_str().to_string(),
verified: false,
reason_code: Some(decision.reason_code),
remediation: Some(decision.remediation),
evidence: vec![DependencyPreflightEvidence {
root: normalized_project_root.to_string_lossy().to_string(),
manifest: normalized_project_root
.join("Cargo.toml")
.to_string_lossy()
.to_string(),
required_path: normalized_project_root
.join("Cargo.toml")
.to_string_lossy()
.to_string(),
required_kind: "manifest",
status,
reason_code: decision.reason_code,
detail: decision.detail.clone(),
is_primary: true,
}],
}
}
fn should_force_local_fallback_for_runtime_fail_open(reason_code: &str) -> bool {
reason_code == DEPENDENCY_PREFLIGHT_CODE_POLICY
|| reason_code == DEPENDENCY_PREFLIGHT_CODE_MATERIALIZATION
}
fn command_uses_cargo_dependency_graph(kind: Option<CompilationKind>) -> bool {
matches!(
kind,
Some(
CompilationKind::CargoBuild
| CompilationKind::CargoCheck
| CompilationKind::CargoClippy
| CompilationKind::CargoDoc
| CompilationKind::CargoTest
| CompilationKind::CargoNextest
| CompilationKind::CargoBench
// A zig cross-build consumes the same cargo dependency graph as
// a plain build; omitting it here skipped sibling path-dep root
// sync entirely, so every zigbuild of a workspace with external
// path deps died with "failed to read <sibling>/Cargo.toml"
// (hfdt aarch64 release leg, 2026-08-06).
| CompilationKind::CargoZigbuild
)
)
}
/// bd-raobv: why a build started in `cwd` must run locally because the project
/// lies outside the configured canonical root (the worker mirror cannot place
/// it), or `None` when topology admits it. The execution path and
/// `rch diagnose` both call this so their verdicts cannot disagree.
pub(crate) fn project_topology_local_reason(
policy: &PathTopologyPolicy,
cwd: &Path,
) -> Option<String> {
normalize_project_path_with_policy(cwd, policy)
.err()
.map(|error| {
format!(
"project {} is outside canonical root {} ({error}); set [path_topology] \
canonical_root or RCH_CANONICAL_PROJECT_ROOT to offload it",
cwd.display(),
policy.canonical_root().display()
)
})
}
fn normalize_dependency_root_for_runtime(
root: &Path,
policy: &PathTopologyPolicy,
) -> Option<PathBuf> {
normalize_project_path_with_policy(root, policy)
.ok()
.map(|normalized| normalized.canonical_path().to_path_buf())
}
fn build_dependency_runtime_plan(
normalized_project_root: &Path,
kind: Option<CompilationKind>,
reporter: &HookReporter,
topology_policy: &PathTopologyPolicy,
) -> DependencyRuntimePlan {
if !command_uses_cargo_dependency_graph(kind) {
return DependencyRuntimePlan {
sync_roots: vec![normalized_project_root.to_path_buf()],
fail_open_decision: None,
};
}
let plan = build_dependency_closure_plan_with_policy(normalized_project_root, topology_policy);
if !plan.is_ready() {
if let Some(reason) = &plan.fail_open_reason {
reporter.verbose(&format!(
"[RCH] dependency closure planner fail-open: {}",
reason
));
}
for issue in &plan.issues {
reporter.verbose(&format!(
"[RCH] dependency closure issue {} ({:?}): {}",
issue.code, issue.risk, issue.message
));
}
let decision = classify_dependency_runtime_fail_open(&plan);
reporter.verbose(&format!(
"[RCH] dependency planner fail-open decision [{}]: {}",
decision.reason_code, decision.remediation
));
return DependencyRuntimePlan {
sync_roots: vec![normalized_project_root.to_path_buf()],
fail_open_decision: Some(decision),
};
}
let mut seen = std::collections::BTreeSet::<PathBuf>::new();
let mut ordered = Vec::<PathBuf>::new();
for action in &plan.sync_order {
if let Some(root) =
normalize_dependency_root_for_runtime(&action.package_root, topology_policy)
&& seen.insert(root.clone())
{
reporter.verbose(&format!(
"[RCH] dependency root {} ({:?})",
root.display(),
action.metadata.reason
));
ordered.push(root);
}
}
if ordered.is_empty() {
ordered.push(normalized_project_root.to_path_buf());
}
if !ordered.iter().any(|root| root == normalized_project_root) {
ordered.push(normalized_project_root.to_path_buf());
}
DependencyRuntimePlan {
sync_roots: ordered,
fail_open_decision: None,
}
}
fn parse_command_tokens(command: &str, reporter: &HookReporter) -> Option<Vec<String>> {
match shell_words::split(command) {
Ok(tokens) => Some(tokens),
Err(error) => {
reporter.verbose(&format!(
"[RCH] failed to parse delegated command for CARGO_TARGET_DIR forwarding: {}",
error
));
None
}
}
}
/// Map a classification kind to required runtime.
pub(crate) fn required_runtime_for_kind(kind: Option<CompilationKind>) -> RequiredRuntime {
match kind {
Some(k) => match k {
CompilationKind::CargoBuild
| CompilationKind::CargoTest
| CompilationKind::CargoCheck
| CompilationKind::CargoClippy
| CompilationKind::CargoDoc
| CompilationKind::CargoNextest
| CompilationKind::CargoBench
| CompilationKind::Rustc => RequiredRuntime::Rust,
// A zig cross-build is a cargo build, but plain `RequiredRuntime::Rust`
// is NOT a sufficient gate: `needs_zig` is only enforced by
// `assess_admissibility`, which the daemon's selection path never
// calls (only fleet smoke tests and bypass recovery do). A Rust-capable
// worker without `cargo-zigbuild` would therefore be selected and fail
// with `error: no such command: 'zigbuild'` — which is not a rustup
// message, so `is_toolchain_failure` doesn't recognize it and the
// nonzero exit is surfaced to the user verbatim instead of falling open
// to local. Zig gets its own runtime so selection gates on `has_zig()`.
CompilationKind::CargoZigbuild => RequiredRuntime::Zig,
CompilationKind::BunTest | CompilationKind::BunTypecheck => RequiredRuntime::Bun,
CompilationKind::NixBuild => RequiredRuntime::Nix,
// Go builds/tests/vets must carry the Go runtime so worker selection
// gates them to a go-capable worker via `has_go()`. Falling through to
// the `_ => None` catch-all would disable capability gating entirely and
// dispatch `go build` to a worker with no Go toolchain.
CompilationKind::GoBuild | CompilationKind::GoTest | CompilationKind::GoVet => {
RequiredRuntime::Go
}
// `tsc` / `npx tsc` run under Node.
CompilationKind::Tsc => RequiredRuntime::Node,
_ => RequiredRuntime::None,
},
None => RequiredRuntime::None,
}
}
/// Add per-worker CARGO_HOME isolation to prevent cache lock contention.
pub(crate) fn add_cargo_isolation(command: &str, worker_id: &WorkerId) -> String {
// Check if this is a cargo command that could benefit from isolation
if !command.contains("cargo") {
return command.to_string();
}
// Durable per-worker CARGO_HOME (issue #42). The old scheme created a
// unique CARGO_HOME per job and deleted it afterwards, which forced every
// job to re-download the registry and — far worse — re-clone and re-pack
// every Cargo *Git* dependency from scratch. For a large Git dependency
// that pack step alone can exceed the external build timeout, so the job
// is killed, the work is discarded, and the next attempt starts over: an
// unbounded loop that never reaches compilation.
//
// A single per-worker cache dir fixes that: registry downloads and git
// db/checkout state persist across jobs (and across timeout kills — a
// completed fetch stays completed). Concurrent jobs on the same worker
// coordinate through Cargo's own package-cache lock, which lives inside
// the (now shared) CARGO_HOME — sharing the lock is what makes concurrent
// cache access safe; the old per-job split actually defeated it.
//
// The staging base is resolved on the worker at execution time (honoring
// $TMPDIR / /data/tmp / /tmp) rather than hardcoding /tmp, so these caches
// don't eat RAM on tmpfs-/tmp hosts. `cargo_home` is therefore a shell
// expression (`${RCH_CH_BASE}/rch-cargo-cache-…`) and must be double-quoted,
// not shell-escaped, so `$RCH_CH_BASE` expands; the worker_id is sanitized
// so the basename needs no further escaping. The durable dir deliberately
// uses the `rch-cargo-cache-` prefix, NOT the legacy per-job
// `rch-cargo-home-` prefix that orphan-cleanup passes match on — this dir
// is not an orphan and must survive between jobs. Each job `touch`es the
// cache dir so its top-level mtime doubles as a liveness signal for the
// transfer-side janitor sweep (bd-wfumv): only caches unused for
// WORKER_DURABLE_CACHE_PRUNE_MAX_AGE_MINS get reaped.
//
// `CARGO_NET_GIT_FETCH_WITH_CLI` routes Cargo's git-dependency fetches
// through the git CLI when available: for pathological repositories the
// libgit2 pack path can spin at ~100% CPU for over an hour where git
// itself finishes in minutes.
let safe_worker_id = sanitize_cargo_home_token(worker_id.as_str());
let cargo_home = rch_common::remote_cargo_cache_expr(&safe_worker_id);
let quoted_cargo_home = format!("\"{cargo_home}\"");
let base_prelude = rch_common::remote_cargo_home_base_prelude();
let base_var = rch_common::RCH_CARGO_HOME_BASE_VAR;
let escaped_command = shell_escape::escape(command.into());
let script = format!(
"{base_var}=\"${{1:-}}\"; if [ -z \"${{{base_var}}}\" ]; then {base_prelude}; fi; mkdir -p {cargo_home} || exit $?; touch {cargo_home} 2>/dev/null || true; for rch_cache_dir in registry git; do if [ -L {cargo_home}/$rch_cache_dir ] && [ ! -e {cargo_home}/$rch_cache_dir ]; then (cd {cargo_home} && mkdir -p -- \"$(readlink -- \"$rch_cache_dir\")\") || exit $?; fi; done; export CARGO_HOME={cargo_home}; if command -v git >/dev/null 2>&1; then export CARGO_NET_GIT_FETCH_WITH_CLI=true; fi; sh -c {command}",
base_prelude = base_prelude,
cargo_home = quoted_cargo_home,
command = escaped_command
);
// The transfer layer may prepend `timeout ...` directly before this string.
// Running the env assignment inside an explicit shell prevents `timeout`
// from trying to exec `CARGO_HOME=...` as argv[0]. The wrapped command is
// the script's final statement, so its exit status propagates unchanged.
format!(
"sh -c {} rch-cargo-cache \"${{{base_var}:-}}\"",
shell_escape::escape(script.into())
)
}
#[cfg(test)]
mod tests;