rch 2.1.0

//! 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 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
    ));
    if let Err(error) = disable_worker_for_fault(
        socket_path,
        worker_id,
        "e104-timeout-orphan-unverified: remote build process group not verified dead after client timeout; possible orphan holding the project's Cargo target lock",
    )
    .await
    {
        warn!(
            "Failed to quarantine worker {} after unverified E104 cleanup: {}",
            worker_id, error
        );
    }
}

/// 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).
    let mut input = String::new();
    {
        use tokio::io::{AsyncReadExt, stdin};
        if let Err(e) = stdin()
            .take(10 * 1024 * 1024)
            .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(_) => {
            match serde_json::to_string(&output) {
                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 normalized_parts = normalize_exec_command_parts(command_parts);
    shell_words::join(normalized_parts)
}

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,
) -> ! {
    // 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.
    record_hook_incident(&build_local_fallback_incident(
        command,
        reason,
        require_remote,
        now_unix_ms(),
    ));
    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) => {
            emit_exec_envelope(&ExecResultEnvelope {
                api_version: "1.0",
                command,
                outcome: "completed",
                location: "local",
                fallback_reason: Some(reason),
                worker_id: None,
                remote_exit_code: status.code(),
                duration_ms: None,
                timing: None,
                result_dirs: None,
                error_code: None,
            });
            std::process::exit(status.code().unwrap_or(1))
        }
        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,
) -> ! {
    if require_remote {
        // Canonical remote-required refusal (proof mode / RCH_REQUIRE_REMOTE).
        exit_with_local_fallback(command, reporter, reason, true);
    }
    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(command, reporter, reason, false)
}

/// 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`.
    GatedLocalFallback { reason: String },
}

/// Fetch daemon status, rank the remaining workers by capacity, and re-query the
/// daemon pinned to the biggest untried worker. Returns the fresh selection
/// response plus the chosen worker id, or `None` when no bigger 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);
    let chosen = pick_bigger_worker(snapshots.as_slice(), tried_workers, 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() => Some((response, chosen)),
        Ok(_) => {
            reporter.verbose(&format!(
                "[RCH] retry: bigger worker {chosen} is not currently admissible; ending retries"
            ));
            None
        }
        Err(e) => {
            reporter.verbose(&format!(
                "[RCH] retry: re-query for {chosen} failed ({e}); ending retries"
            ));
            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,
) -> IncidentEvent {
    build_recovery_terminal_incident(
        refused,
        &extract_project_name(),
        &redact_secrets(command),
        reason,
        now_ms,
    )
}

/// 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 routine local fallback (RCH-I011) is already announced by the
    // `[RCH] local (<reason>)` summary; now that every fallback is recorded,
    // a WARN here would add multi-line stderr noise to every local build.
    if event.reason_code == IncidentReasonCode::LocalFallback {
        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",
        );
    }
    let ledger = IncidentLedger::new(IncidentLedgerConfig::default());
    if let Err(e) = ledger.append(event) {
        warn!(
            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,
            );
        };

        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,
            );
        }
        if !selected_worker_is_requested(&worker.id, &current_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
        });
        let release_acknowledged = if retain_unconfirmed_ownership {
            warn!(
                "Remote completion unconfirmed; retaining build {} ownership",
                remote_build_id
            );
            false
        } else {
            match release_worker(
                &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),
            )
            .await
            {
                Ok(()) => true,
                Err(error) => {
                    warn!("Failed to release worker slots: {}", error);
                    false
                }
            }
        };
        if let Ok(result) = result.as_ref() {
            durable_lease.record_exit(result.exit_code)?;
        }

        // 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),
                        },
                    }
                } 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
                            ),
                        },
                    }
                } 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(),
                        },
                    }
                }
            }
        };

        // 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 } => {
                // 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,
                );
            }
        }
    }
}

#[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, 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};
pub(crate) use daemon_ipc::{query_daemon, 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)
    };

    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; 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;