rch 2.1.18

//! Transfer / remote-execution orchestration for the hook.
//!
//! This submodule owns the remote-build execution pipeline: `execute_remote_compilation`
//! — which syncs the project to a worker, runs the command remotely with a
//! streaming heartbeat, and retrieves artifacts back — together with the leaf
//! telemetry-forwarding helpers it drives (wrapping the remote command so it
//! piggybacks worker telemetry on stdout, and the two daemon-IPC POST helpers
//! that forward the collected `WorkerTelemetry` / `TestRunRecord` back to the
//! local daemon).
//!
//! It reaches its support layer from the parent via `use super::*`: the
//! sync-topology / dependency-manifest helpers, `HookReporter`, and the
//! `rch_common` types/consts. The offload SSH primitives now live in the sibling
//! `ssh` submodule — this module drives the remote topology preflight via
//! `ensure_worker_projects_topology`, imported explicitly below. The build
//! heartbeat (`progress_reporting`) and the repo_updater pre-sync entry point
//! (`repo_updater`) likewise live in sibling submodules and are imported below.
//!
//! `execute_remote_compilation` is `pub(super)` (its only non-test callers,
//! `run_hook`/`run_exec`, are re-exported into `hook`); `wrap_command_with_telemetry`
//! stays `pub(super)` for the hook test suite; the two daemon-IPC POST helpers
//! are private to this module.

use super::artifact_patterns::{
    artifact_delivery_kind, expected_output_glob_list, get_custom_target_artifact_patterns,
    get_project_artifact_patterns, kind_has_enumerable_output_contract,
    kind_produces_transferable_artifacts, sync_back_verified_zero_build_outputs,
    sync_back_verified_zero_package_archives,
};
use super::artifact_triple::{describe_findings, foreign_target_artifacts};
use super::cargo_output_contract::CargoOutputCapture;
use super::cargo_target_dir::{
    cargo_target_env_allowlist, cargo_target_env_overrides, default_host_target_triple,
    explicit_target_triple_for_command, remote_cargo_pooled_target_dir_name,
    remote_cargo_target_dir_name, stale_target_reap_idle_hours, target_reuse_disabled,
};
use super::daemon_ipc::urlencoding_encode;
use super::dependency_closure::{
    SyncClosureMode, SyncClosurePlanEntry, SyncRootOutcome, build_owned_sync_closure_layout,
    build_sync_closure_manifest, build_sync_closure_plan, merge_sync_result,
    verify_remote_dependency_manifests, workspace_metadata_sync_patterns,
};
use super::formatting::{cache_hit, detect_target_label, emit_job_banner, render_compile_summary};
use super::progress_reporting::{
    BuildHeartbeatLoop, BuildHeartbeatSnapshot, mark_heartbeat_progress,
};
use super::remote_result::RemoteExecutionResult;
use super::repo_updater::maybe_sync_repo_set_with_repo_updater;
use super::source_fidelity::{
    PreparedSourceContentRoot, bind_build_source_aliases, build_source_commit_env,
    capture_build_source_stamp, finalize_owned_source_content_receipt,
    finalize_source_content_receipt, guard_owned_source_content_cargo_config,
    owned_source_content_freshness_identity, prepare_source_content_root,
    reconcile_build_source_stamps, validate_owned_source_content_closure,
    verify_owned_source_content_roots, verify_source_content_roots,
};
use super::ssh::{
    acquire_clean_overlay_source_pair, acquire_remote_source_authority_lock,
    ensure_worker_projects_topology, remote_preflight_topology_policy,
    worker_projects_topology_requires_initialization,
};
use super::*;

#[path = "cargo_manifest.rs"]
mod cargo_manifest;

#[path = "retrieval_recovery.rs"]
pub(crate) mod recovery;

/// Read the durable completion receipt after the execution SSH session exits.
///
/// The read is idempotent (it only `cat`s an immutable, identity-bound file),
/// so a transient SSH failure is retried briefly instead of turning a build
/// that already finished into "completion unconfirmed", which retains its
/// source ownership and fences later builds on that worker. Only errors are
/// retried; an absent receipt is evidence and is returned at once.
async fn read_completion_with_retry(
    pipeline: &TransferPipeline,
    worker: &WorkerConfig,
) -> anyhow::Result<Option<i32>> {
    retry_completion_probe(Duration::from_secs(1), async || {
        pipeline.read_recovery_completion(worker).await
    })
    .await
}

/// Up to three probe attempts, sleeping `backoff` then twice that between
/// failures (1 s and 2 s in production).
async fn retry_completion_probe(
    backoff: Duration,
    mut probe: impl AsyncFnMut() -> anyhow::Result<Option<i32>>,
) -> anyhow::Result<Option<i32>> {
    const ATTEMPTS: u64 = 3;
    let mut attempt = 1;
    loop {
        match probe().await {
            Err(error) if attempt < ATTEMPTS => {
                warn!(attempt, %error, "completion probe failed; retrying");
                tokio::time::sleep(backoff * u32::try_from(attempt).unwrap_or(1)).await;
                attempt += 1;
            }
            other => return other,
        }
    }
}

/// Keep ownership failures distinct from a confirmed pre-workload setup refusal.
fn confirm_source_pair_execution(
    lock: Option<&mut super::ssh::RemoteSourceAuthorityLock>,
    exit_code: i32,
) -> anyhow::Result<()> {
    if !(0..255).contains(&exit_code) {
        return Err(anyhow::anyhow!(
            "SSH execution exit {exit_code} does not prove remote completion"
        )
        .context(crate::transfer::RemoteExecutionUnconfirmed));
    }
    if let Some(lock) = lock {
        lock.ensure_execution_finished(exit_code)
            .context(crate::transfer::RemoteExecutionUnconfirmed)?;
    }
    Ok(())
}

/// A holder can disappear while durable completion is being read. Recheck the
/// actual guard before turning a setup refusal into permission to fail over.
pub(super) fn ensure_remote_process_setup_after_completion(
    pipeline: &TransferPipeline,
    result: &rch_common::CommandResult,
    source_pair_lock: Option<&mut super::ssh::RemoteSourceAuthorityLock>,
) -> anyhow::Result<()> {
    confirm_source_pair_execution(source_pair_lock, result.exit_code)?;
    pipeline.ensure_remote_process_setup(result)
}

/// A recovery request can interrupt collection, never execution. Keep the same
/// session and authority guards outside this boundary across the one explicit
/// retry, and wait for transport cancellation to reap its receiver first.
async fn retrieve_with_live_recovery<T>(
    pipeline: &TransferPipeline,
    lease: Option<&DurableLeaseWriter>,
    mut retrieve: impl AsyncFnMut(&TransferPipeline) -> anyhow::Result<T>,
) -> anyhow::Result<T> {
    // Retrieval futures are large; heap-pin them (bd-uz82c).
    let Some(lease) = lease else {
        return Box::pin(retrieve(pipeline)).await;
    };
    let evidence = lease.snapshot();
    if evidence.recovery.is_none() || evidence.identity.remote_build_id.is_none() {
        return Box::pin(retrieve(pipeline)).await;
    }
    let receipt = default_job_lease_directory()
        .join(format!("{}.recover", evidence.identity.local_wrapper_id));
    let started = tokio::time::Instant::now();
    for attempt in 0..2 {
        let (cancel, requested) = tokio::sync::watch::channel(false);
        let controlled = pipeline.clone().with_retrieval_control(requested, started);
        let result = {
            let mut retrieval = Box::pin(retrieve(&controlled));
            let mut poll = tokio::time::interval(Duration::from_millis(250));
            poll.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
            loop {
                tokio::select! {
                    biased;
                    result = &mut retrieval => break Some(result),
                    _ = poll.tick(), if attempt == 0 => {
                        if consume_recovery_request(&receipt, &evidence.identity) {
                            info!(wrapper_id = %evidence.identity.local_wrapper_id, "Interrupting artifact collection for same-job recovery");
                            let _ = cancel.send(true);
                            let result = retrieval.await;
                            if result.as_ref().is_err_and(|error| error.is::<crate::transfer::RetrievalCancelled>()) {
                                break None;
                            }
                            // A raced completion (or failure) is authoritative;
                            // only acknowledged cancellation authorizes retry.
                            break Some(result);
                        }
                    }
                }
            }
        };
        if let Some(result) = result {
            return result;
        }
        info!(wrapper_id = %evidence.identity.local_wrapper_id, "Retrying artifact collection after receiver reaped; command not replayed");
    }
    unreachable!("second collection attempt cannot consume another request")
}

fn consume_recovery_request(path: &Path, identity: &JobIdentity) -> bool {
    let Ok(bytes) = std::fs::read(path) else {
        return false;
    };
    if serde_json::from_slice::<JobIdentity>(&bytes).ok().as_ref() != Some(identity) {
        return false;
    }
    // Retain evidence and avoid overwriting any previous acknowledgment. The
    // wrapper is the sole consumer; producers atomically write the same identity.
    let consumed = path.with_extension(format!("recover-consumed-{}", uuid::Uuid::new_v4()));
    match std::fs::rename(path, consumed) {
        Ok(()) => true,
        Err(error) => {
            warn!(%error, "Could not consume same-job recovery request");
            false
        }
    }
}

fn clean_overlay_cargo_policy_failure(
    root: &Path,
    worker: &str,
    detail: String,
) -> DependencyPreflightFailure {
    DependencyPreflightFailure::from_report(super::dependency_closure::DependencyPreflightReport {
        schema_version: DEPENDENCY_PREFLIGHT_SCHEMA_VERSION,
        worker: worker.to_owned(),
        verified: false,
        reason_code: Some(DEPENDENCY_PREFLIGHT_CODE_POLICY),
        remediation: Some(
            "Clean-overlay requires selected Cargo inputs. Bind each external sibling Git root with --dependency-base PATH=REV; non-sibling layouts are refused.",
        ),
        evidence: vec![DependencyPreflightEvidence {
            root: root.display().to_string(),
            manifest: "Cargo.toml".to_owned(),
            required_path: root.display().to_string(),
            required_kind: "selected_cargo_sources",
            status: super::dependency_closure::DependencyPreflightStatus::PolicyViolation,
            reason_code: DEPENDENCY_PREFLIGHT_CODE_POLICY,
            detail,
            is_primary: true,
        }],
    })
}

pub(super) fn source_sync_terminal_summary(
    attempts: &[crate::transfer::TransferAttemptDiagnostic],
    clean_overlay: bool,
) -> Option<String> {
    let last = attempts.last()?;
    let label = if clean_overlay {
        "clean-overlay source sync"
    } else {
        "source sync"
    };
    Some(format!(
        "[RCH] {label} failed before remote Cargo execution after {}/{} attempts; remote Cargo was not started: {}",
        attempts.len(),
        last.max_attempts,
        last.detail
    ))
}

/// Per-segment callback for the streaming source sync (issue #59): forwards
/// rsync output to the rich progress UI (when enabled) and marks
/// build-heartbeat forward progress so the daemon can distinguish a live
/// transfer from a stalled one while the build phase is still `sync_up`.
/// Every rsync output segment — including bare-`\r` `--info=progress2`
/// refreshes — counts as forward progress.
pub(super) fn sync_progress_line(
    progress: Option<&mut TransferProgress>,
    heartbeat_state: Option<&Arc<Mutex<BuildHeartbeatSnapshot>>>,
    line: &str,
) {
    if let Some(progress) = progress {
        progress.update_from_line(line);
    }
    if let Some(state) = heartbeat_state {
        mark_heartbeat_progress(state);
    }
}

pub(super) fn apply_source_sync_integrity_policy(
    pipeline: TransferPipeline,
    exact_dependency_closure_sync: bool,
    source_content_receipt: bool,
) -> TransferPipeline {
    // Exact Cargo closure sync is correctness-authoritative: rsync's default size-and-mtime
    // quick check can otherwise retain stale dependency bytes with matching metadata.
    if exact_dependency_closure_sync || source_content_receipt {
        pipeline.with_sync_checksum(true)
    } else {
        pipeline
    }
}

/// Select the Cargo families whose build and target directories can be bound
/// by the owned source route. External drivers keep their existing receipt
/// path; malformed commands within a supported family still fail admission.
fn supports_owned_source_content_cargo(kind: Option<CompilationKind>, command: &str) -> bool {
    matches!(
        kind,
        Some(
            CompilationKind::CargoBuild
                | CompilationKind::CargoTest
                | CompilationKind::CargoCheck
                | CompilationKind::CargoClippy
                | CompilationKind::CargoDoc
                | CompilationKind::CargoBench
        )
    ) && rch_common::patterns::cargo_xwin_compilation_kind(command).is_none()
}

pub(super) fn wrap_command_with_telemetry(command: &str, worker_id: &WorkerId) -> String {
    let escaped_worker = shell_escape::escape(worker_id.as_str().into());
    // Use newline instead of semicolon to ensure trailing comments in command
    // don't comment out the status capture logic.
    format!(
        "{cmd}\nstatus=$?; if command -v rch-telemetry >/dev/null 2>&1; then \
         telemetry=$(rch-telemetry collect --format json --worker-id {worker} 2>/dev/null || true); \
         if [ -n \"$telemetry\" ]; then echo '{marker}'; echo \"$telemetry\"; fi; \
         fi; exit $status",
        cmd = command,
        worker = escaped_worker,
        marker = PIGGYBACK_MARKER
    )
}

/// Derive the remote-root component for one clean-overlay execution.
///
/// Unpooled jobs use a nonce to keep even identical source snapshots in
/// separate directories. Pooled jobs instead hold an exclusive source-pair
/// lease throughout materialization, execution, retrieval, and retirement.
fn clean_overlay_remote_project_hash(
    base_commit: &str,
    overlay_fingerprint: &str,
    job_nonce: uuid::Uuid,
) -> String {
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"rch-clean-overlay-remote-root-v1\0");
    hasher.update(base_commit.as_bytes());
    hasher.update(b"\0");
    hasher.update(overlay_fingerprint.as_bytes());
    hasher.update(b"\0");
    hasher.update(job_nonce.as_bytes());
    hasher.finalize().to_hex()[..16].to_string()
}

/// Migrate away from artifacts that embed disposable source paths. Preserve
/// the pooled basename grammar used by the cache reaper.
fn clean_overlay_source_pair_pool_name(legacy_name: &str, source_base: &str) -> String {
    let prefix = legacy_name
        .rsplit_once('-')
        .map_or(legacy_name, |(prefix, _)| prefix);
    let hash = blake3::hash(
        format!(
            "rch-clean-overlay-source-pair-v1\0{legacy_name}\0{}",
            source_base.trim_end_matches('/')
        )
        .as_bytes(),
    );
    format!("{prefix}-{}", &hash.to_hex()[..32])
}

/// Bind freshness to the entire selected closure, including explicit overlays.
/// The source-pair lease and fresh-root materialization remain prerequisites;
/// this identity only permits reusing the prior source timestamp, not bytes.
fn clean_overlay_freshness_identity(spec: &CleanOverlaySpec) -> String {
    fn field(hash: &mut blake3::Hasher, bytes: &[u8]) {
        hash.update(&(bytes.len() as u64).to_le_bytes());
        hash.update(bytes);
    }
    fn selected(hash: &mut blake3::Hasher, spec: &CleanOverlaySpec) {
        field(hash, spec.base_commit().as_bytes());
        field(hash, spec.tree_object.as_bytes());
        field(hash, spec.overlay_fingerprint().as_bytes());
        hash.update(&(spec.dependencies.len() as u64).to_le_bytes());
        let mut dependencies = spec.dependencies.iter().collect::<Vec<_>>();
        dependencies.sort_by(|(left, _), (right, _)| left.cmp(right));
        for (root, dependency) in dependencies {
            field(hash, root.as_os_str().as_encoded_bytes());
            selected(hash, dependency);
        }
    }
    let mut hash = blake3::Hasher::new();
    hash.update(b"rch-clean-overlay-freshness-v1\0");
    selected(&mut hash, spec);
    hash.finalize().to_hex().to_string()
}

/// The STABLE absolute location of a pooled Cargo target store under an
/// explicit base: `<base>/<project_id>/<pooled dir name>`.
///
/// Used for two things. Issue #60: a clean-overlay run's remote root
/// is retired at teardown, so its pooled store must
/// live in a sibling path that survives. Issue #64:
/// `[remediation.pooled_target] store_base` relocates every build's pooled
/// store off the filesystem holding the project mirror. Either way the
/// `.rch-target-…-pool-…` basename is preserved so the pool-janitor / sbh GC
/// conventions (and Cargo's own CACHEDIR.TAG) still apply, and Cargo's
/// target-dir flock serializes overlapping jobs sharing the pool.
/// Where a build's pooled Cargo target store is PLACED on the worker, or
/// `None` to keep it inside the synced project mirror (the default: under the
/// worker's canonical project root, on whatever filesystem holds it).
///
/// Issue #64: `[remediation.pooled_target] store_base`, when configured, moves
/// every build's pooled store to `<store_base>/<project_id>/<name>` — how a
/// worker whose canonical root sits on a small disk keeps multi-GB warm pools
/// on a larger volume. Windows workers keep their drive-letter build base: a
/// Unix store base is not a valid path there.
///
/// Issue #60: a clean-overlay run hashes a per-command job nonce into its
/// remote root (a deliberate overlap-safety invariant) and reaps that root
/// wholesale at teardown, so a pooled store placed UNDER it could never be
/// reused. Absent a `store_base`, those runs relocate the pool under
/// `[transfer] remote_base`, where their roots already live.
fn pooled_store_base_for<'a>(
    worker_is_windows: bool,
    store_base: Option<&'a str>,
    clean_overlay: bool,
    transfer_remote_base: &'a str,
) -> Option<&'a str> {
    if worker_is_windows {
        return clean_overlay.then_some(crate::transfer::WINDOWS_DEFAULT_REMOTE_BASE);
    }
    store_base
        .map(|base| base.trim_end_matches('/'))
        .filter(|base| !base.is_empty())
        .or_else(|| clean_overlay.then(|| transfer_remote_base.trim_end_matches('/')))
}

fn stable_pooled_target_dir(remote_base: &str, project_id: &str, pooled_dir_name: &str) -> String {
    format!(
        "{}/{}/{}",
        remote_base.trim_end_matches('/'),
        project_id,
        pooled_dir_name
    )
}

async fn send_telemetry(
    socket_path: &str,
    source: TelemetrySource,
    telemetry: &WorkerTelemetry,
) -> anyhow::Result<()> {
    if !Path::new(socket_path).exists() {
        return Ok(());
    }

    let stream = match timeout(Duration::from_secs(2), UnixStream::connect(socket_path)).await {
        Ok(Ok(s)) => s,
        Ok(Err(e)) => return Err(e.into()),
        Err(_) => return Ok(()), // Timeout connecting — don't block hook
    };
    let (reader, mut writer) = stream.into_split();

    let body = telemetry.to_json()?;
    let request = format!(
        "POST /telemetry/ingest?source={}\n{}\n",
        urlencoding_encode(&source.to_string()),
        body
    );

    writer.write_all(request.as_bytes()).await?;
    writer.flush().await?;
    writer.shutdown().await?;

    let mut reader = BufReader::new(reader);
    let mut line = String::new();
    let _ = timeout(Duration::from_secs(5), reader.read_line(&mut line)).await;

    Ok(())
}

async fn send_test_run(socket_path: &str, record: &TestRunRecord) -> anyhow::Result<()> {
    if !Path::new(socket_path).exists() {
        return Ok(());
    }

    let stream = match timeout(Duration::from_secs(2), UnixStream::connect(socket_path)).await {
        Ok(Ok(s)) => s,
        Ok(Err(e)) => return Err(e.into()),
        Err(_) => return Ok(()), // Timeout connecting — don't block hook
    };
    let (reader, mut writer) = stream.into_split();

    let body = record.to_json()?;
    let request = format!("POST /test-run\n{}\n", body);

    writer.write_all(request.as_bytes()).await?;
    writer.flush().await?;
    writer.shutdown().await?;

    let mut reader = BufReader::new(reader);
    let mut line = String::new();
    let _ = timeout(Duration::from_secs(5), reader.read_line(&mut line)).await;

    Ok(())
}

/// Environment opt-out for the retrieved-artifact typing gate (GitHub #65).
///
/// An operator who deliberately wants a foreign-platform artifact in the local
/// target tree — cross-build staging, a container image assembled from a Linux
/// worker's output on a macOS controller — sets `RCH_ALLOW_FOREIGN_ARTIFACTS=1`
/// and gets the pre-#65 behaviour back for that invocation. Any value other
/// than empty / `0` / `false` / `no` / `off` (case-insensitive) opts out.
fn foreign_artifact_gate_disabled() -> bool {
    foreign_artifact_gate_disabled_from_value(std::env::var(RCH_ALLOW_FOREIGN_ARTIFACTS_ENV).ok())
}

/// Pure predicate behind [`foreign_artifact_gate_disabled`], with the env value
/// injected so it is unit-testable under `#![forbid(unsafe_code)]`.
pub(super) fn foreign_artifact_gate_disabled_from_value(value: Option<String>) -> bool {
    value.is_some_and(|value| {
        let value = value.trim().to_ascii_lowercase();
        !value.is_empty() && value != "0" && value != "false" && value != "no" && value != "off"
    })
}

/// Environment variable name for [`foreign_artifact_gate_disabled`].
pub(super) const RCH_ALLOW_FOREIGN_ARTIFACTS_ENV: &str = "RCH_ALLOW_FOREIGN_ARTIFACTS";

/// Execute a compilation command on a remote worker.
///
/// This function:
/// 1. Syncs the project to the remote worker
/// 2. Executes the command remotely with streaming output
/// 3. Retrieves build artifacts back to local
///
/// Returns the execution result including exit code and stderr.
#[allow(clippy::too_many_arguments)] // Pipeline wiring favors explicit params
pub(super) async fn execute_remote_compilation(
    worker: &SelectedWorker,
    command: &str,
    transfer_config: TransferConfig,
    env_allowlist: Vec<String>,
    forwarded_cargo_target_dir: Option<PathBuf>,
    compilation_config: &rch_common::CompilationConfig,
    toolchain: Option<&ToolchainInfo>,
    kind: Option<CompilationKind>,
    reporter: &HookReporter,
    socket_path: &str,
    color_mode: ColorMode,
    build_id: Option<u64>,
    local_wrapper_id: Option<&str>,
    durable_lease: Option<&DurableLeaseWriter>,
    topology_policy: &PathTopologyPolicy,
    clean_overlay: Option<&CleanOverlaySpec>,
    source_content_receipt: bool,
    // Declared job result directories (bd-p0yoo), repository-relative.
    // Retrieved after the command completes on ANY exit code; a directory
    // that is missing or only partially transferable fails the invocation
    // loudly instead of surfacing the job's bare exit status.
    result_dirs: &[PathBuf],
    // Resolved Layer 0 pack env pairs (bd-bqu38), forced onto the remote
    // build regardless of the ambient environment.
    layer0_env: &[(String, String)],
    // Configured `[remediation.pooled_target] reaper_pooled_idle_hours`
    // (issue #53): the transfer-start janitor prunes idle pooled target
    // stores on exactly this window so it never undercuts the reaper.
    pooled_target_prune_idle_hours: u32,
    // Configured `[remediation.pooled_target] store_base` (issue #64):
    // `None` keeps pooled stores inside the project mirror.
    pooled_target_store_base: Option<&str>,
) -> anyhow::Result<RemoteExecutionResult> {
    // The pipeline's state machine is large; held inline, debug builds moved
    // it across a 2 MiB thread stack and overflowed (bd-uz82c). Keep it on
    // the heap so every caller's future stays small.
    let outcome = Box::pin(execute_remote_compilation_inner(
        worker,
        command,
        transfer_config,
        env_allowlist,
        forwarded_cargo_target_dir,
        compilation_config,
        toolchain,
        kind,
        reporter,
        socket_path,
        color_mode,
        build_id,
        local_wrapper_id,
        durable_lease,
        topology_policy,
        clean_overlay,
        source_content_receipt,
        result_dirs,
        layer0_env,
        pooled_target_prune_idle_hours,
        pooled_target_store_base,
    ))
    .await;
    // The inner future has dropped every local holder before cleanup tries to
    // reattach it. Cancellation drains remote mutation descriptors first; once
    // execution may have started, its exact completion must resolve ownership.
    if outcome.is_err()
        && let Some(writer) = durable_lease.filter(|writer| writer.snapshot().recovery.is_some())
    {
        match recovery::cancel_preparation(writer).await {
            Ok(true) => {}
            Ok(false) => {
                return outcome
                    .map_err(|error| error.context(crate::transfer::RemoteExecutionUnconfirmed));
            }
            Err(cleanup) => {
                let detail = format!(
                    "source ownership remains recoverable with rch jobs recover {}: {cleanup:#}",
                    writer.snapshot().identity.local_wrapper_id
                );
                return outcome.map_err(|error| {
                    error
                        .context(detail)
                        .context(crate::transfer::RemoteExecutionUnconfirmed)
                });
            }
        }
    }
    outcome
}

#[allow(clippy::too_many_arguments)]
async fn execute_remote_compilation_inner(
    worker: &SelectedWorker,
    command: &str,
    transfer_config: TransferConfig,
    env_allowlist: Vec<String>,
    forwarded_cargo_target_dir: Option<PathBuf>,
    compilation_config: &rch_common::CompilationConfig,
    toolchain: Option<&ToolchainInfo>,
    kind: Option<CompilationKind>,
    reporter: &HookReporter,
    socket_path: &str,
    color_mode: ColorMode,
    build_id: Option<u64>,
    local_wrapper_id: Option<&str>,
    durable_lease: Option<&DurableLeaseWriter>,
    topology_policy: &PathTopologyPolicy,
    clean_overlay: Option<&CleanOverlaySpec>,
    source_content_receipt: bool,
    result_dirs: &[PathBuf],
    layer0_env: &[(String, String)],
    pooled_target_prune_idle_hours: u32,
    pooled_target_store_base: Option<&str>,
) -> anyhow::Result<RemoteExecutionResult> {
    let worker_config = selected_worker_to_config(worker);
    let caller_command = command;
    // The build is already registered. Source validation and fingerprinting can
    // outlast the stuck detector's deadline before any remote transfer begins.
    // Keep this RAII guard alive across preflight, including every early error;
    // the remote process-group path is attached only once planning establishes it.
    let mut heartbeat_loop = build_id.map(|id| {
        BuildHeartbeatLoop::start(
            socket_path,
            id,
            &worker_config.id,
            local_wrapper_id,
            durable_lease,
        )
    });
    if let Some(loop_ref) = heartbeat_loop.as_ref() {
        loop_ref.update_phase(
            BuildHeartbeatPhase::SyncUp,
            Some("source_validation".to_string()),
        );
        loop_ref.flush().await;
    }
    if source_content_receipt && WorkerPlatform::from_worker(&worker_config).is_windows() {
        anyhow::bail!("source-content receipts require the Unix rsync transport");
    }
    if !result_dirs.is_empty() && WorkerPlatform::from_worker(&worker_config).is_windows() {
        anyhow::bail!("--result-dir requires the Unix rsync transport; worker is Windows");
    }
    let source_content_build_id = if source_content_receipt {
        Some(build_id.ok_or_else(|| {
            anyhow::anyhow!("source-content receipt requires a durable remote build id")
        })?)
    } else {
        None
    };

    // Get current working directory and normalize it to the canonical project root.
    let project_root =
        std::env::current_dir().map_err(|e| TransferError::NoProjectRoot { source: e })?;
    let normalized_project = normalize_project_path_with_policy(&project_root, topology_policy)
        .map_err(|e| {
            anyhow::anyhow!(
                "Project path normalization failed for {}: {}",
                project_root.display(),
                e
            )
        })?;
    for decision in normalized_project.decision_trace() {
        reporter.verbose(&format!("[RCH] project path normalized: {}", decision));
    }
    let normalized_project_root = normalized_project.canonical_path().to_path_buf();

    // Windows workers have no Unix canonical/alias projects topology and cargo
    // needs drive-letter paths, so their whole remote layout lives under the
    // Windows build base and syncs via tar-over-ssh (no rsync/streaming).
    let worker_is_windows = WorkerPlatform::from_worker(&worker_config).is_windows();
    if reporter.cargo_json_stdout {
        anyhow::ensure!(
            !worker_is_windows
                && durable_lease.is_some()
                && !super::ssh::should_skip_remote_preflight(&worker_config),
            "caller Cargo JSON requires the admitted durable POSIX worker route"
        );
    }

    let go_build_environment = if kind == Some(CompilationKind::GoBuild) {
        anyhow::ensure!(
            !worker_is_windows
                && durable_lease.is_some()
                && !super::ssh::should_skip_remote_preflight(&worker_config),
            "Go output delivery requires the durable Unix worker path"
        );
        Some(
            super::artifact_patterns::direct_compiler::validate_go_build_output(
                command,
                &normalized_project_root,
            )
            .await?,
        )
    } else {
        None
    };

    let clean_overlay_cargo = clean_overlay.is_some()
        && (kind.is_some_and(|kind| kind.command_base() == "cargo")
            || classify_command(command)
                .kind
                .is_some_and(|kind| kind.command_base() == "cargo"));
    if clean_overlay_cargo && worker_is_windows {
        return Err(clean_overlay_cargo_policy_failure(
            &normalized_project_root,
            &worker_config.id.to_string(),
            "Selected Cargo configuration validation requires a Unix worker".to_owned(),
        )
        .into());
    }
    if let Some(spec) = clean_overlay
        && clean_overlay_cargo
        && let Err(error) =
            super::validate_clean_overlay_cargo_sources(&normalized_project_root, spec, command)
                .await
    {
        return Err(clean_overlay_cargo_policy_failure(
            &normalized_project_root,
            &worker_config.id.to_string(),
            format!("{error:#}"),
        )
        .into());
    }

    let exact_dependency_closure_sync =
        clean_overlay.is_none() && command_uses_cargo_dependency_graph(kind);
    // #70: the invocation/transfer basis need not be a Cargo package. Select
    // the graph entrypoint without changing the build cwd or artifact basis.
    let explicit_manifest_root = if exact_dependency_closure_sync {
        cargo_manifest::selected_manifest_root(command, &normalized_project_root, topology_policy)
            .context("cannot plan the selected Cargo manifest")?
    } else {
        None
    };
    let dependency_entry_root = explicit_manifest_root
        .as_deref()
        .unwrap_or(&normalized_project_root);
    let build_source_before = if kind.is_some_and(|kind| kind.command_base() == "cargo") {
        Some(capture_build_source_stamp(dependency_entry_root, clean_overlay).await)
    } else {
        None
    };
    let raw_sync_roots = if let Some(spec) = clean_overlay {
        let mut roots = vec![normalized_project_root.clone()];
        roots.extend(spec.dependencies.iter().map(|(root, _)| root.clone()));
        roots
    } else {
        let dependency_plan =
            build_dependency_runtime_plan(dependency_entry_root, kind, reporter, topology_policy);
        if let Some(decision) = dependency_plan.fail_open_decision.as_ref() {
            let report = build_dependency_runtime_fail_open_report(
                &worker_config,
                dependency_entry_root,
                decision,
            );
            if let Ok(report_json) = serde_json::to_string(&report) {
                reporter.verbose(&format!(
                    "[RCH] dependency planner fail-open report: {}",
                    report_json
                ));
            }
            if source_content_receipt || explicit_manifest_root.is_some() {
                warn!(
                    "Dependency planner could not prove the exact source closure on {} [{}]: refusing selected-manifest/receipt execution ({})",
                    worker_config.id, decision.reason_code, decision.remediation
                );
                reporter.verbose(&format!(
                    "[RCH] dependency planner refusal [{}]: selected-manifest and receipt modes require an exact closure — {}",
                    decision.reason_code, decision.remediation
                ));
                return Err(DependencyPreflightFailure::from_report(report).into());
            }
            if exact_dependency_closure_sync
                && should_force_local_fallback_for_runtime_fail_open(decision.reason_code)
            {
                warn!(
                    "Dependency planner fail-open on {} [{}]: refusing remote Cargo execution and falling back local ({})",
                    worker_config.id, decision.reason_code, decision.remediation
                );
                reporter.verbose(&format!(
                    "[RCH] dependency planner fail-open [{}]: exact dependency closure required, forcing local fallback — {}",
                    decision.reason_code, decision.remediation
                ));
                return Err(DependencyPreflightFailure::from_report(report).into());
            }
            warn!(
                "Dependency planner fail-open on {} [{}]: proceeding with primary-root-only sync ({})",
                worker_config.id, decision.reason_code, decision.remediation
            );
            reporter.verbose(&format!(
                "[RCH] dependency planner fail-open [{}]: proceeding with primary root only — {}",
                decision.reason_code, decision.remediation
            ));
        }
        dependency_plan.sync_roots
    };
    let project_id = project_id_from_path(&normalized_project_root);
    if clean_overlay.is_some() && kind.is_some_and(|kind| kind.command_base() == "cargo") {
        // Reject unsupported command grammar before claiming a persistent
        // owner or writing source. The real directory is bound once the
        // primary pipeline has resolved its managed target below.
        super::cargo_target_dir::managed_clean_overlay_cargo_build_dir(
            command,
            "/rch-managed-build-dir-validation",
        )?;
    }
    // Stable compiler-visible paths and their caches form one leased pair.
    // Windows lacks the POSIX source lease; retain invocation isolation there
    // with a unique target instead of reusing path-bearing artifacts unsafely.
    let reuse_disabled = target_reuse_disabled() || (clean_overlay.is_some() && worker_is_windows);
    // The explicit source-content route can reuse compiler paths only when the
    // worker can retire a whole source tree without replacing or deleting it.
    // Probe before persisting an intent or mutating any worker path. Other Unix
    // platforms keep invocation-isolated source until their rename protocol is
    // implemented and qualified; a Linux capability failure is a refusal.
    let owned_source_content_pair = if source_content_receipt
        && clean_overlay.is_none()
        && exact_dependency_closure_sync
        && supports_owned_source_content_cargo(kind, command)
        && !reuse_disabled
    {
        anyhow::ensure!(
            durable_lease.is_some()
                && !super::ssh::should_skip_remote_preflight(&worker_config)
                && forwarded_cargo_target_dir.is_some(),
            "owned source-content staging requires a durable Unix Cargo target route"
        );
        let probe = format!(
            "set -eu; if [ \"$(uname -s)\" = Linux ]; then {}; \
             else printf '%s\\n' RCH_SOURCE_RETENTION_LEGACY_PLATFORM; fi",
            TransferPipeline::remote_tree_retention_capability_command(),
        );
        let output = super::ssh::run_offload_ssh_command_with_stdin(
            &worker_config,
            "sh -s",
            probe.as_bytes(),
            Duration::from_secs(30),
        )
        .await?;
        anyhow::ensure!(
            output.status.success() && output.stderr.is_empty(),
            "owned source retention admission failed on {} (status {:?}): {}",
            worker_config.id,
            output.status.code(),
            String::from_utf8_lossy(&output.stderr).trim(),
        );
        match output.stdout.as_slice() {
            b"RCH_SOURCE_RETENTION_CAPABLE\n" => true,
            b"RCH_SOURCE_RETENTION_LEGACY_PLATFORM\n" => false,
            _ => anyhow::bail!("owned source retention admission returned an invalid envelope"),
        }
    } else {
        false
    };
    if owned_source_content_pair {
        // A managed build directory must win over ambient Cargo settings and
        // cannot be added safely to arbitrary shell grammar.
        super::cargo_target_dir::managed_clean_overlay_cargo_build_dir(
            command,
            "/rch-managed-build-dir-validation",
        )?;
    }
    let mut source_pair_pool = (clean_overlay.is_some()
        && forwarded_cargo_target_dir.is_some()
        && !reuse_disabled)
        .then(|| {
            let legacy = clean_overlay_source_pair_pool_name(
                &remote_cargo_pooled_target_dir_name(
                    &worker_config.id,
                    &normalized_project_root,
                    toolchain,
                    command,
                ),
                &transfer_config.remote_base,
            );
            if let Some(spec) = clean_overlay.filter(|spec| !spec.dependencies.is_empty()) {
                let mut layout = blake3::Hasher::new();
                layout.update(b"rch-selected-sibling-layout-v1\0");
                layout.update(legacy.as_bytes());
                for root in &raw_sync_roots {
                    layout.update(b"\0");
                    layout.update(root.file_name().unwrap_or_default().as_encoded_bytes());
                }
                debug_assert!(spec.primary_directory.is_some());
                format!(
                    ".rch-target-{}-pool-{}",
                    worker_config.id,
                    &layout.finalize().to_hex()[..32]
                )
            } else {
                legacy
            }
        });
    let project_hash = if let Some(pool) = source_pair_pool.as_ref() {
        format!("paired-{}", &blake3::hash(pool.as_bytes()).to_hex()[..32])
    } else if let Some(spec) = clean_overlay {
        clean_overlay_remote_project_hash(
            spec.base_commit(),
            spec.overlay_fingerprint(),
            uuid::Uuid::new_v4(),
        )
    } else {
        compute_project_hash_with_dependency_roots_and_policy(
            &normalized_project_root,
            &raw_sync_roots,
            topology_policy,
        )
    };
    // Bind receipt provenance to this invocation. The legacy receipt route
    // also uses it for its isolated path; the owned pair instead preserves
    // compiler paths under a durable exclusive source/target lease.
    let project_hash = if let Some(proof_build_id) = source_content_build_id {
        blake3::hash(
            format!(
                "rch.source_content_remote_root.v1\0{}\0{}\0{}",
                project_hash, proof_build_id, worker_config.id
            )
            .as_bytes(),
        )
        .to_hex()
        .to_string()
    } else {
        project_hash
    };
    let mut sync_plan = if clean_overlay.is_some() {
        // Explicit committed roots are the whole authority. The ordinary
        // planner reads ambient ancestor manifests and must not widen this set.
        raw_sync_roots
            .iter()
            .map(|root| SyncClosurePlanEntry {
                local_root: root.clone(),
                project_id: project_id_from_path(root),
                root_hash: project_hash.clone(),
                remote_root: String::new(),
                is_primary: root == &normalized_project_root,
                mode: SyncClosureMode::Full,
            })
            .collect()
    } else {
        build_sync_closure_plan(
            &raw_sync_roots,
            &normalized_project_root,
            &project_hash,
            topology_policy,
        )
    };
    // Ordinary Cargo invocations target shared canonical worker paths. Capture
    // those logical authorities before any proof/overlay/Windows relocation so
    // overlapping primary projects that share a path dependency take the same
    // remote lock. Source-content and clean-overlay runs already own isolated
    // source roots and do not need this mutable-authority guard.
    let recovery_identity = uuid::Uuid::new_v4().simple().to_string();
    let mut overlay_remote_root: Option<String> = None;
    if owned_source_content_pair {
        let layout = build_owned_sync_closure_layout(
            &sync_plan,
            topology_policy,
            &transfer_config.remote_base,
            &remote_cargo_pooled_target_dir_name(
                &worker_config.id,
                &normalized_project_root,
                toolchain,
                command,
            ),
        )?;
        source_pair_pool = Some(layout.pooled_target_name);
        overlay_remote_root = Some(layout.source_root);
        sync_plan = layout.plan;
    } else if let Some(proof_build_id) = source_content_build_id {
        let proof_base = format!(
            "{}/source-content-{}-{}",
            transfer_config.remote_base.trim_end_matches('/'),
            proof_build_id,
            &project_hash[..project_hash.len().min(16)]
        );
        for entry in &mut sync_plan {
            let relative = entry
                .local_root
                .strip_prefix(topology_policy.canonical_root())
                .with_context(|| {
                    format!(
                        "source-content root {} is outside canonical topology {}",
                        entry.local_root.display(),
                        topology_policy.canonical_root().display()
                    )
                })?;
            let relative = relative.to_str().ok_or_else(|| {
                anyhow::anyhow!(
                    "source-content remote relative path is not UTF-8: {}",
                    relative.display()
                )
            })?;
            if relative.is_empty() || relative.chars().any(char::is_control) {
                anyhow::bail!("source-content remote relative path is invalid");
            }
            entry.remote_root = format!("{proof_base}/{relative}");
            entry.root_hash = blake3::hash(
                format!(
                    "rch.source_content_root_hash.v1\0{}\0{}\0{}",
                    entry.root_hash, proof_build_id, relative
                )
                .as_bytes(),
            )
            .to_hex()
            .to_string();
        }
    }
    if let Some(spec) = clean_overlay {
        anyhow::ensure!(
            sync_plan.iter().filter(|entry| entry.is_primary).count() == 1,
            "clean-overlay requires exactly one primary root"
        );
        let remote_base = transfer_config.remote_base.trim_end_matches('/');
        let container = format!("{remote_base}/{project_id}/{project_hash}");
        for entry in &mut sync_plan {
            entry.remote_root = if spec.primary_directory.is_some() {
                let directory = entry
                    .local_root
                    .file_name()
                    .and_then(|name| name.to_str())
                    .ok_or_else(|| anyhow::anyhow!("invalid selected root directory"))?;
                format!("{container}/{directory}")
            } else {
                container.clone()
            };
            entry.root_hash.clone_from(&project_hash);
        }
        overlay_remote_root = Some(container);
    }
    let retained_source_root = owned_source_content_pair.then(|| {
        format!(
            "{}.retained-{recovery_identity}",
            overlay_remote_root
                .as_deref()
                .expect("owned source container")
        )
    });
    let source_freshness_control = owned_source_content_pair.then(|| {
        format!(
            "{}.freshness",
            overlay_remote_root
                .as_deref()
                .expect("owned source container")
        )
    });
    // Relocate every closure root under the Windows build base so all downstream
    // remote paths (sync target, build cwd, CARGO_TARGET_DIR, manifest
    // verification) use drive-letter paths. Mirrors the clean-overlay remote_root
    // override above but applies to the whole plan. See rch#<NN>.
    if worker_is_windows {
        for entry in sync_plan.iter_mut() {
            entry.remote_root = format!(
                "{}/{}/{}",
                crate::transfer::WINDOWS_DEFAULT_REMOTE_BASE,
                entry.project_id,
                entry.root_hash
            );
        }
    }
    let sync_roots = sync_plan
        .iter()
        .map(|entry| entry.local_root.clone())
        .collect::<Vec<_>>();
    let sync_manifest = build_sync_closure_manifest(&sync_plan, &normalized_project_root);

    let output_ctx = OutputContext::detect();
    let console = RchConsole::with_context(output_ctx);
    let feedback_visible = reporter.visibility != OutputVisibility::None && !console.is_machine();
    // The Windows tar-over-ssh transport has no streaming (rsync-style) progress
    // variant, so disable progress there to force the Windows-aware
    // non-streaming sync/retrieve paths.
    let progress_enabled = output_ctx.supports_rich()
        && reporter.visibility != OutputVisibility::None
        && !worker_is_windows
        && !reporter.cargo_json_stdout;
    let remote_pgid_file = build_id.and_then(|id| {
        sync_plan
            .iter()
            .find(|entry| entry.is_primary)
            .map(|entry| TransferPipeline::remote_pgid_file_path_for_root(&entry.remote_root, id))
    });
    if let Some(loop_ref) = heartbeat_loop.as_ref() {
        loop_ref.set_remote_pgid_file(remote_pgid_file);
        loop_ref.update_phase(BuildHeartbeatPhase::SyncUp, Some("sync_start".to_string()));
        loop_ref.flush().await;
    }

    if feedback_visible {
        emit_job_banner(&console, output_ctx, worker, build_id);
    }

    info!(
        "Starting remote compilation pipeline for {} (hash: {})",
        project_id, project_hash
    );
    reporter.verbose(&format!(
        "[RCH] dependency sync roots planned: {}",
        sync_plan.len()
    ));
    for (idx, entry) in sync_plan.iter().enumerate() {
        reporter.verbose(&format!(
            "[RCH] dependency sync root {}/{}: {}",
            idx + 1,
            sync_plan.len(),
            entry.local_root.display()
        ));
    }
    match serde_json::to_string(&sync_manifest) {
        Ok(manifest_json) => {
            reporter.verbose(&format!(
                "[RCH] dependency sync manifest: {}",
                manifest_json
            ));
            info!(
                "Prepared dependency sync manifest for {} roots",
                sync_manifest.entries.len()
            );
        }
        Err(err) => warn!("Failed to serialize dependency sync manifest: {}", err),
    }
    reporter.verbose(&format!(
        "[RCH] sync start (project {} on {})",
        project_id, worker_config.id
    ));

    let remote_cap = compilation_config.timeout_for_kind(kind);
    let command_timeout = if compilation_config.external_timeout_enabled() {
        remote_cap + std::time::Duration::from_secs(30)
    } else {
        remote_cap
    };
    // Ensure deterministic remote topology before any repo synchronization.
    // bd-8iwkm/bd-gc0ze: the ownership sweep inside is scoped to this
    // dispatch's closure remote roots so its cost tracks the trees about to
    // be written, not the entire multi-GB mirror tree.
    let ownership_scan_roots: Vec<PathBuf> = sync_plan
        .iter()
        .map(|entry| PathBuf::from(entry.remote_root.as_str()))
        .collect();
    let remote_topology_policy = remote_preflight_topology_policy(
        topology_policy,
        clean_overlay.is_some() && !worker_is_windows,
        source_content_receipt && !worker_is_windows,
        &transfer_config.remote_base,
        &ownership_scan_roots,
    )?;
    // bd-fbtws: reject live checkouts before ownership repair or repo-updater
    // convergence can mutate them, not merely when rsync starts. The receiver
    // repeats the guard immediately before pruning/upload under its grant.
    if !worker_is_windows && !super::ssh::should_skip_remote_preflight(&worker_config) {
        let guard = sync_plan
            .iter()
            .map(|entry| {
                format!(
                    "{} || exit 73;\n",
                    crate::transfer::live_checkout_sync_guard(&entry.remote_root)
                )
            })
            .collect::<String>();
        let output = super::ssh::run_offload_ssh_command_with_stdin(
            &worker_config,
            "sh -s",
            guard.as_bytes(),
            Duration::from_secs(30),
        )
        .await?;
        anyhow::ensure!(
            output.status.success(),
            "source destination safety check refused on {} (status {:?}): {}",
            worker_config.id,
            output.status.code(),
            String::from_utf8_lossy(&output.stderr).trim(),
        );
    }
    // Build transfer pipelines with color mode, command timeout, and compilation kind.
    // When the in-session watchdog is active it enforces the real build cap
    // remotely (same timeout_for_kind value). Give the local SSH stream a grace
    // margin over that cap so a genuine remote group-kill propagates as exit
    // 137 instead of losing the race to a local "SSH command timed out" (#20).
    let mut effective_env_allowlist =
        cargo_target_env_allowlist(&env_allowlist, forwarded_cargo_target_dir.is_some());
    let build_source_aliases = build_source_before
        .as_ref()
        .map(|_| build_source_commit_env(|key| std::env::var(key).ok()));
    if build_source_aliases.is_some() {
        // Alias cleanup/restoration belongs immediately around the caller's
        // command, after worker environment assembly. Empty forwarded values
        // would suppress valid Cargo `[env]` defaults from selected config.
        effective_env_allowlist
            .retain(|key| !rch_common::BUILD_COMMIT_ENV_VARS.contains(&key.trim()));
    }
    let cargo_env_overrides = cargo_target_env_overrides(forwarded_cargo_target_dir.as_deref());
    // Remote target-dir name for the forwarded-CARGO_TARGET_DIR sync. By default
    // this is a STABLE pooled name keyed on (project, toolchain, triple, profile,
    // features) so independent jobs with identical dimensions REUSE the same warm
    // remote incremental cache instead of cold-recompiling into a unique-per-job
    // dir. `RCH_DISABLE_TARGET_REUSE=1` restores the legacy unique-per-job name.
    let remote_cargo_target_dir_name_override = forwarded_cargo_target_dir.as_ref().map(|_| {
        if reuse_disabled {
            reporter.verbose(
                "[RCH] remote target-dir reuse disabled (explicit opt-out or Windows clean-overlay); using unique-per-job dir",
            );
            remote_cargo_target_dir_name(build_id, &worker_config.id)
        } else {
            let name = source_pair_pool.clone().unwrap_or_else(|| remote_cargo_pooled_target_dir_name(
                &worker_config.id,
                &normalized_project_root,
                toolchain,
                command,
            ));
            reporter.verbose(&format!(
                "[RCH] remote target-dir reuse active; pooled dir {name}"
            ));
            name
        }
    });
    let pooled_store_base = pooled_store_base_for(
        worker_is_windows,
        pooled_target_store_base,
        clean_overlay.is_some() || owned_source_content_pair,
        &transfer_config.remote_base,
    );
    let pooled_target_dir_override = if reuse_disabled {
        None
    } else {
        pooled_store_base
            .zip(remote_cargo_target_dir_name_override.as_ref())
            .map(|(base, name)| {
                let stable = stable_pooled_target_dir(base, &project_id, name);
                let stable = if owned_source_content_pair {
                    super::ssh::canonical_source_authority_root(&stable)
                } else {
                    stable
                };
                reporter.verbose(&format!(
                    "[RCH] pooled target store placed outside the project mirror: {stable}"
                ));
                stable
            })
    };
    if owned_source_content_pair {
        let target = Path::new(
            pooled_target_dir_override
                .as_deref()
                .context("owned source staging requires a persistent target outside its source")?,
        );
        for source in [
            overlay_remote_root.as_deref(),
            retained_source_root.as_deref(),
            source_freshness_control.as_deref(),
        ]
        .into_iter()
        .flatten()
        {
            let source = Path::new(source);
            anyhow::ensure!(
                !target.starts_with(source) && !source.starts_with(target),
                "owned source, retained trees, and freshness metadata must be separate from the target pool"
            );
        }
    }
    let durable_sources =
        !worker_is_windows && !super::ssh::should_skip_remote_preflight(&worker_config);
    // Initializing a shared alias changes the physical meaning of every path
    // beneath it. Discover that need without mutation, then include both
    // topology roots in the same saved grant before creating either path.
    // Healthy topology needs no parent grant, so sibling builds stay parallel.
    let initialize_topology = if durable_sources {
        worker_projects_topology_requires_initialization(&worker_config, &remote_topology_policy)
            .await?
    } else {
        false
    };
    let mut mutable_source_authority_roots: Vec<String> = sync_plan
        .iter()
        .map(|entry| entry.remote_root.clone())
        .collect();
    if initialize_topology {
        for root in [
            remote_topology_policy.canonical_root(),
            remote_topology_policy.alias_root(),
        ] {
            mutable_source_authority_roots.push(
                root.to_str()
                    .context("worker topology root is not a UTF-8 path")?
                    .to_owned(),
            );
        }
    }
    if source_pair_pool.is_some()
        && let Some(root) = overlay_remote_root.as_ref()
    {
        // Include the complete pair in the durable claim. Its already-owned
        // kernel lock is omitted by the source grant using the live pair guard.
        mutable_source_authority_roots.push(root.clone());
    }
    // Retirement and freshness metadata are outside the reusable source tree.
    // Their exact siblings belong to this same saved grant before any write.
    if let Some(root) = retained_source_root.as_ref() {
        mutable_source_authority_roots.push(root.clone());
    }
    if let Some(root) = source_freshness_control.as_ref() {
        mutable_source_authority_roots.push(root.clone());
    }
    if let Some(primary) = sync_plan.iter().find(|entry| entry.is_primary) {
        mutable_source_authority_roots.push(pooled_target_dir_override.clone().unwrap_or_else(
            || {
                format!(
                    "{}/{}",
                    primary.remote_root,
                    remote_cargo_target_dir_name_override
                        .as_deref()
                        .unwrap_or(".rch-target")
                )
            },
        ));
    }
    // Record and lock the canonical spelling: a trailing-slash project path
    // used to reach the lease verbatim and was then unlockable and
    // unrecoverable (bd-4d1hs).
    for root in &mut mutable_source_authority_roots {
        *root = super::ssh::canonical_source_authority_root(root);
    }
    mutable_source_authority_roots.sort();
    mutable_source_authority_roots.dedup();
    if worker_is_windows {
        mutable_source_authority_roots.clear();
    }
    let source_identity = durable_sources.then_some(recovery_identity.as_str());
    let planned_pair = source_pair_pool.as_ref().map(|_| {
        (
            overlay_remote_root
                .clone()
                .expect("paired source container"),
            recovery_identity.clone(),
        )
    });
    if durable_sources {
        recovery::RecoverySession::begin(
            durable_lease.context(
                "remote source ownership requires an admitted durable job lease; use rch exec",
            )?,
            &worker_config,
            mutable_source_authority_roots.clone(),
            planned_pair.clone(),
            overlay_remote_root.clone(),
            retained_source_root.clone(),
            transfer_config.clone(),
            normalized_project_root.clone(),
            recovery_identity.clone(),
        )?;
    }
    let mut source_pair_lock = if durable_sources {
        if let Some((root, token)) = planned_pair.as_ref() {
            reporter.verbose("[RCH] waiting for the owned source/target pair; same-pool jobs serialize through retrieval and source retirement");
            Some(
                acquire_clean_overlay_source_pair(&worker_config, root, token, command_timeout)
                    .await?,
            )
        } else {
            None
        }
    } else {
        None
    };
    let mut source_authority_lock = if mutable_source_authority_roots.is_empty()
        || super::ssh::should_skip_remote_preflight(&worker_config)
    {
        None
    } else {
        reporter.verbose(&format!(
            "[RCH] waiting for {} remote source-authority lock(s) on {}",
            mutable_source_authority_roots.len(),
            worker_config.id
        ));
        let guard = acquire_remote_source_authority_lock(
            &worker_config,
            &mutable_source_authority_roots,
            source_pair_lock.as_mut(),
            &recovery_identity,
            command_timeout,
        )
        .await?;
        reporter.verbose(&format!(
            "[RCH] acquired remote source-authority locks on {}",
            worker_config.id
        ));
        Some(guard)
    };
    // Ownership repair changes source-tree metadata too. It must wait for the
    // complete grant, including ancestor/descendant exclusion, just like sync.
    ensure_worker_projects_topology(
        &worker_config,
        reporter,
        &remote_topology_policy,
        &ownership_scan_roots,
        source_identity,
        initialize_topology,
    )
    .await?;
    // Hold source and output authorities through retrieval, not only execution.
    // Best-effort repo convergence for ordinary multi-repo dependency graphs.
    // Both isolated modes already name their complete staged source roots.
    // Converging canonical worker checkouts would mutate paths outside this
    // job's source grant and cannot prepare any of its selected destinations.
    if clean_overlay.is_none() && !source_content_receipt {
        maybe_sync_repo_set_with_repo_updater(
            &worker_config,
            &sync_roots,
            reporter,
            source_identity,
        )
        .await?;
    } else {
        reporter
            .verbose("[RCH] isolated sources selected; canonical repository convergence skipped");
    }

    let mut primary_pipeline: Option<TransferPipeline> = None;
    let mut aggregate_sync_result: Option<SyncResult> = None;
    let mut prepared_source_roots: Vec<PreparedSourceContentRoot> = Vec::new();

    // Step 1: Sync project to remote
    info!("Syncing project to worker {}...", worker_config.id);
    let mut upload_progress = if progress_enabled {
        Some(TransferProgress::upload(
            output_ctx,
            "Syncing workspace closure",
            reporter.visibility == OutputVisibility::None,
        ))
    } else {
        None
    };
    // Issue #59: the streaming sync path is used for every non-Windows worker
    // (not only when the rich progress UI is enabled) so rsync's output feeds
    // BOTH the silence-based stall detector inside the transfer layer and the
    // build heartbeat here — a live sync stays observably alive to the daemon
    // while the phase is still `sync_up`.
    let sync_streaming = !worker_is_windows;
    let sync_heartbeat_state = heartbeat_loop
        .as_ref()
        .map(BuildHeartbeatLoop::shared_state);
    let mut root_outcomes: Vec<(SyncClosurePlanEntry, SyncRootOutcome)> = Vec::new();
    for entry in &sync_plan {
        let root_overlay = clean_overlay
            .map(|spec| {
                if entry.is_primary {
                    Ok(spec)
                } else {
                    spec.dependencies
                        .iter()
                        .find(|(root, _)| root == &entry.local_root)
                        .map(|(_, selected)| selected)
                        .ok_or_else(|| {
                            anyhow::anyhow!(
                                "unbound clean-overlay dependency root {}",
                                entry.local_root.display()
                            )
                        })
                }
            })
            .transpose()?;
        let mut root_pipeline = TransferPipeline::new(
            entry.local_root.clone(),
            entry.project_id.clone(),
            entry.root_hash.clone(),
            transfer_config.clone(),
        )
        .with_color_mode(color_mode)
        .with_command_timeout(command_timeout)
        .with_compilation_config(compilation_config.clone())
        .with_compilation_kind(kind)
        .with_remote_path_override(entry.remote_root.clone())
        .with_worker_platform(WorkerPlatform::from_worker(&worker_config))
        .with_build_id(build_id)
        .with_pooled_target_prune_idle_hours(pooled_target_prune_idle_hours);
        if let Some(identity) = source_identity {
            root_pipeline = root_pipeline.with_source_authority(identity.to_owned())?;
        }
        if owned_source_content_pair {
            // Acquiring the pair proved that the entire container was absent.
            // A new source tree never needs a deleting sync or a cache sweep.
            // No-delta verification still detects surplus selected entries.
            root_pipeline = root_pipeline.with_owned_source_materialization();
        }
        if let Some(spec) = root_overlay {
            root_pipeline = root_pipeline
                .with_sync_include_patterns(clean_overlay_include_patterns(
                    &entry.local_root,
                    spec.overlay_paths(),
                )?)
                .with_sync_delete(false)
                .with_sync_checksum(true);
            if let Some(stable_pool) = pooled_target_dir_override.as_ref() {
                // Every selected root must become newer than the same cache
                // whose dep-info may refer to any of its source files.
                root_pipeline =
                    root_pipeline.with_remote_cargo_target_dir_override(stable_pool.clone());
            }
        }
        if entry.mode == SyncClosureMode::WorkspaceMetadata {
            root_pipeline =
                root_pipeline.with_sync_include_patterns(workspace_metadata_sync_patterns());
            root_pipeline = root_pipeline.with_env_allowlist(effective_env_allowlist.clone());
            if !layer0_env.is_empty() {
                root_pipeline = root_pipeline.with_layer0_env(layer0_env.to_vec());
            }
        }
        if entry.is_primary {
            root_pipeline = root_pipeline.with_env_allowlist(effective_env_allowlist.clone());
            if let Some(overrides) = cargo_env_overrides.as_ref() {
                root_pipeline = root_pipeline.with_env_overrides(overrides.clone());
            }
            if let Some(name) = remote_cargo_target_dir_name_override.as_ref() {
                root_pipeline = root_pipeline.with_remote_cargo_target_dir_name(name.clone());
            }
            if let Some(stable_pool) = pooled_target_dir_override.as_ref() {
                root_pipeline =
                    root_pipeline.with_remote_cargo_target_dir_override(stable_pool.clone());
            }
        }
        root_pipeline = apply_source_sync_integrity_policy(
            root_pipeline,
            exact_dependency_closure_sync,
            source_content_receipt,
        );

        let prepared_source_root = if source_content_receipt {
            Some(
                prepare_source_content_root(prepared_source_roots.len(), entry, &root_pipeline)
                    .await?,
            )
        } else {
            None
        };

        if let Some(spec) = root_overlay {
            reporter.verbose(&format!(
                "[RCH] clean-overlay transfer estimator bypassed for immutable base {}",
                spec.base_commit()
            ));
        } else if exact_dependency_closure_sync || source_content_receipt {
            reporter.verbose(&format!(
                "[RCH] exact source sync required; bypassing transfer estimator for {}",
                entry.local_root.display()
            ));
        } else if let Some(skip_reason) = root_pipeline.should_skip_transfer(&worker_config).await {
            info!(
                "Transfer estimation indicates skip for {}: {} (worker {})",
                entry.local_root.display(),
                skip_reason,
                worker_config.id
            );
            reporter.verbose(&format!(
                "[RCH] skip transfer for {}: {}",
                entry.local_root.display(),
                skip_reason
            ));
            if entry.is_primary {
                // Primary root skip is fatal — cannot build without the main project.
                return Err(TransferError::TransferSkipped {
                    reason: skip_reason,
                }
                .into());
            }
            root_outcomes.push((
                entry.clone(),
                SyncRootOutcome::Skipped {
                    reason: skip_reason,
                },
            ));
            continue;
        }

        reporter.verbose(&format!(
            "[RCH] syncing dependency root {} to remote {}",
            entry.local_root.display(),
            entry.remote_root.as_str()
        ));
        let sync_attempt = if let Some(spec) = root_overlay {
            spec.verify_archive_attributes(&entry.local_root).await?;
            spec.verify_overlay_unchanged(&entry.local_root)?;
            let base_materialization = match root_pipeline
                .materialize_git_archive(&worker_config, &entry.local_root, spec.base_commit())
                .await
            {
                Ok(materialization) => materialization,
                Err(error) => {
                    if let Some(history) =
                        error.downcast_ref::<crate::transfer::TransferAttemptsExhausted>()
                    {
                        for attempt in &history.attempts {
                            reporter.verbose(&format!(
                                "[RCH] clean-overlay base transfer attempt {}/{} {} before remote Cargo execution: {}",
                                attempt.attempt,
                                attempt.max_attempts,
                                attempt.outcome,
                                attempt.detail
                            ));
                        }
                        if let Some(summary) = source_sync_terminal_summary(&history.attempts, true)
                        {
                            reporter.summary(&summary);
                        }
                    }
                    return Err(error.context(
                        "clean-overlay base transfer failed before remote Cargo execution",
                    ));
                }
            };
            for attempt in &base_materialization.attempts {
                reporter.verbose(&format!(
                    "[RCH] clean-overlay base transfer attempt {}/{} {} before remote Cargo execution: {}",
                    attempt.attempt, attempt.max_attempts, attempt.outcome, attempt.detail
                ));
            }
            let base_result = base_materialization.sync_result;
            spec.verify_archive_attributes(&entry.local_root).await?;
            let result = if spec.is_base_only() {
                base_result
            } else {
                let overlay_result = if sync_streaming {
                    root_pipeline
                        .sync_to_remote_streaming(&worker_config, |line| {
                            sync_progress_line(
                                upload_progress.as_mut(),
                                sync_heartbeat_state.as_ref(),
                                line,
                            );
                        })
                        .await?
                } else {
                    root_pipeline.sync_to_remote(&worker_config).await?
                };
                merge_sync_result(&base_result, &overlay_result)
            };
            spec.verify_overlay_unchanged(&entry.local_root)?;
            if source_pair_pool.is_some() {
                root_pipeline
                    .refresh_clean_overlay_source(
                        &worker_config,
                        &clean_overlay_freshness_identity(
                            clean_overlay.expect("root overlay has a selected closure"),
                        ),
                        overlay_remote_root.as_deref().context(
                            "clean-overlay freshness requires its owned retirement container",
                        )?,
                    )
                    .await?;
            }
            Ok(result)
        } else if sync_streaming {
            root_pipeline
                .sync_to_remote_streaming(&worker_config, |line| {
                    sync_progress_line(
                        upload_progress.as_mut(),
                        sync_heartbeat_state.as_ref(),
                        line,
                    );
                })
                .await
        } else {
            root_pipeline.sync_to_remote(&worker_config).await
        };
        match sync_attempt {
            Ok(root_sync_result) => {
                aggregate_sync_result = Some(match &aggregate_sync_result {
                    Some(existing) => merge_sync_result(existing, &root_sync_result),
                    None => root_sync_result,
                });
                if entry.is_primary {
                    primary_pipeline = Some(root_pipeline);
                }
                if let Some(prepared) = prepared_source_root {
                    prepared_source_roots.push(prepared);
                }
                root_outcomes.push((entry.clone(), SyncRootOutcome::Synced));
            }
            Err(e) => {
                if entry.is_primary
                    || durable_sources
                    || clean_overlay.is_some()
                    || exact_dependency_closure_sync
                    || source_content_receipt
                {
                    // Any failed owned transfer may still arrive remotely.
                    // Cancel and drain this token before another preparation;
                    // never continue into execution behind a late old writer.
                    if let Some(history) =
                        e.downcast_ref::<crate::transfer::TransferAttemptsExhausted>()
                    {
                        for attempt in &history.attempts {
                            reporter.verbose(&format!(
                                "[RCH] source sync attempt {}/{} {} before remote Cargo execution: {}",
                                attempt.attempt,
                                attempt.max_attempts,
                                attempt.outcome,
                                attempt.detail
                            ));
                        }
                        if let Some(summary) =
                            source_sync_terminal_summary(&history.attempts, false)
                        {
                            reporter.summary(&summary);
                        }
                    } else {
                        reporter.summary(&format!(
                            "[RCH] source sync failed before remote Cargo execution; remote Cargo was not started: {e}"
                        ));
                    }
                    return Err(e);
                }
                // Dependency root failure is non-fatal (fail-open for deps).
                warn!(
                    "Dependency root sync failed for {} (non-fatal): {}",
                    entry.local_root.display(),
                    e
                );
                reporter.verbose(&format!(
                    "[RCH] dependency root sync failed (fail-open): {} — {}",
                    entry.local_root.display(),
                    e
                ));
                root_outcomes.push((
                    entry.clone(),
                    SyncRootOutcome::Failed {
                        error: e.to_string(),
                    },
                ));
            }
        }
    }

    // Emit structured partial-sync diagnostics when any dependency roots had issues.
    let failed_count = root_outcomes
        .iter()
        .filter(|(_, o)| !matches!(o, SyncRootOutcome::Synced))
        .count();
    if failed_count > 0 {
        warn!(
            "Partial sync: {}/{} closure roots had issues (build continues with available roots)",
            failed_count,
            sync_plan.len()
        );
        for (entry, outcome) in &root_outcomes {
            match outcome {
                SyncRootOutcome::Synced => {}
                SyncRootOutcome::Skipped { reason } => {
                    info!(
                        "  dependency root skipped: {} — {}",
                        entry.local_root.display(),
                        reason
                    );
                }
                SyncRootOutcome::Failed { error } => {
                    info!(
                        "  dependency root failed: {} — {}",
                        entry.local_root.display(),
                        error
                    );
                }
            }
        }
    }
    let sync_result = aggregate_sync_result
        .ok_or_else(|| anyhow::anyhow!("dependency sync produced no transfer result"))?;
    let pipeline = primary_pipeline.ok_or_else(|| {
        anyhow::anyhow!(
            "dependency sync did not include primary project root {}",
            normalized_project_root.display()
        )
    })?;
    let owned_command = if owned_source_content_pair {
        Some(validate_owned_source_content_closure(
            &prepared_source_roots,
            overlay_remote_root
                .as_deref()
                .context("missing owned source container")?,
            &normalized_project_root,
            explicit_manifest_root.as_deref(),
            command,
            topology_policy,
        )?)
    } else {
        None
    };
    let owned_freshness_identity = owned_source_content_pair
        .then(|| owned_source_content_freshness_identity(&prepared_source_roots))
        .transpose()?;
    let managed_target = (clean_overlay_cargo || owned_source_content_pair)
        .then(|| pipeline.remote_cargo_target_dir());
    let command_plan = super::cargo_target_dir::ManagedCargoCommand::new(
        owned_command.as_deref().unwrap_or(command),
        managed_target.as_deref(),
    )?;
    // Capture output semantics BEFORE build-dir binding and source stamping.
    // Both inject quoted Cargo configuration, which is deliberately outside
    // the classifier's publication grammar. They must not widen archive-only
    // delivery or disable its zero-archive gate (#86, bd-3kskq).
    let policy_command = if owned_source_content_pair {
        caller_command
    } else {
        command_plan.policy_command()
    };
    let owned_target_command = if owned_source_content_pair {
        Some(super::cargo_target_dir::bind_owned_cargo_target_dir(
            command_plan.execution_command(),
            managed_target
                .as_deref()
                .context("owned Cargo has no granted target")?,
        )?)
    } else {
        None
    };
    let command = owned_target_command
        .as_deref()
        .unwrap_or_else(|| command_plan.execution_command());
    let mut recovery_session =
        if !worker_is_windows && !super::ssh::should_skip_remote_preflight(&worker_config) {
            durable_lease
                .map(|writer| {
                    recovery::RecoverySession::prepare(
                        writer,
                        &worker_config,
                        &pipeline,
                        mutable_source_authority_roots.clone(),
                        source_pair_lock
                            .as_ref()
                            .and_then(|lock| lock.pair_token())
                            .map(|token| {
                                (
                                    overlay_remote_root.clone().expect("paired root"),
                                    token.to_owned(),
                                )
                            }),
                        overlay_remote_root.clone(),
                        retained_source_root.clone(),
                        transfer_config.clone(),
                        normalized_project_root.clone(),
                        forwarded_cargo_target_dir.as_deref(),
                        kind,
                        policy_command,
                        result_dirs,
                        recovery_identity.clone(),
                    )
                })
                .transpose()?
        } else {
            None
        };
    if reporter.cargo_json_stdout {
        recovery_session
            .as_mut()
            .context("caller Cargo JSON has no admitted recovery session")?
            .request_caller_json(Path::new(&pipeline.remote_path()))?;
    }
    let pipeline = match recovery_session.as_ref() {
        Some(session) => session.completion_pipeline(pipeline),
        None => pipeline,
    };
    // Only the durable POSIX path can retain complete, identity-bound Cargo
    // records across a wrapper disconnect. Validate the caller's original
    // command before adding the execution-only JSON option to a managed one.
    let cargo_output_capture = recovery_session
        .as_ref()
        .and_then(|_| CargoOutputCapture::for_command(kind, policy_command));
    if reporter.cargo_json_stdout {
        anyhow::ensure!(
            cargo_output_capture
                .as_ref()
                .is_some_and(CargoOutputCapture::caller_json_supported),
            "caller Cargo JSON lost its admitted named-output capture"
        );
    }
    let captured_command = cargo_output_capture
        .as_ref()
        .map(|capture| capture.execution_command(command));
    let command = captured_command.as_deref().unwrap_or(command);
    info!(
        "Sync complete: {} files, {} bytes in {}ms",
        sync_result.files_transferred, sync_result.bytes_transferred, sync_result.duration_ms
    );
    // Opportunistically reclaim *abandoned* per-job target dirs for this project
    // on the chosen worker. Only siblings with no file activity past the threshold
    // are removed, so any dir still in active use is preserved and this never races
    // a concurrent build on the same project. The heavy removal is detached on the
    // worker (a backgrounded rm); only a quick SSH dispatch is awaited here.
    // Best-effort; gated to the forwarded-CARGO_TARGET_DIR mode that makes per-job dirs.
    if clean_overlay.is_none() && !owned_source_content_pair && forwarded_cargo_target_dir.is_some()
    {
        // Cheap, current-project-only reap: only this build's own repo dir is
        // swept for abandoned sibling per-job dirs. The durable cross-project
        // GC (every repo under the worker's sync-root) now runs OFF this
        // per-dispatch path in the background daemon sweep
        // (`rchd::stale_target_reap`), so this stays a single `cd` + glob loop.
        pipeline
            .reap_stale_sibling_per_job_target_dirs(&worker_config, stale_target_reap_idle_hours())
            .await;
    }
    reporter.verbose(&format!(
        "[RCH] sync done: {} files, {} bytes in {}ms",
        sync_result.files_transferred, sync_result.bytes_transferred, sync_result.duration_ms
    ));
    if let Some(progress) = &mut upload_progress {
        progress.apply_summary(sync_result.bytes_transferred, sync_result.files_transferred);
        progress.finish();
    }
    if let Some(loop_ref) = heartbeat_loop.as_ref() {
        loop_ref.update_phase(
            BuildHeartbeatPhase::Execute,
            Some("remote_exec_start".to_string()),
        );
        loop_ref.flush().await;
    }

    if exact_dependency_closure_sync {
        // Verify package authorities even when their transfers were collapsed
        // into a manifest-less repository basis. Use the actual remote mapping
        // and sync outcomes rather than inventing a root Cargo.toml (#70).
        let manifest_outcomes;
        let preflight_outcomes = if explicit_manifest_root.is_some() {
            manifest_outcomes = cargo_manifest::manifest_preflight_outcomes(
                &root_outcomes,
                &raw_sync_roots,
                topology_policy,
            )?;
            &manifest_outcomes
        } else {
            &root_outcomes
        };
        verify_remote_dependency_manifests(
            &worker_config,
            preflight_outcomes,
            reporter,
            source_identity,
        )
        .await?;
    }

    if source_content_build_id.is_some() {
        if prepared_source_roots.len() != sync_plan.len() {
            anyhow::bail!(
                "source-content proof prepared {} of {} planned roots",
                prepared_source_roots.len(),
                sync_plan.len()
            );
        }
        // Admission check immediately before Cargo opens the isolated source
        // tree. The same roots are verified again after Cargo exits, and only
        // then is the single receipt emitted.
        if owned_source_content_pair {
            source_authority_lock
                .as_mut()
                .context("owned source freshness requires its full source grant")?
                .ensure_held()?;
            source_pair_lock
                .as_mut()
                .context("owned source freshness requires its source/target pair")?
                .ensure_held()?;
            verify_owned_source_content_roots(
                &worker_config,
                &prepared_source_roots,
                source_identity,
                false,
            )
            .await?;
            let source_container = overlay_remote_root
                .as_deref()
                .context("owned source freshness has no source container")?;
            let control = source_freshness_control
                .as_deref()
                .context("owned source freshness has no granted metadata directory")?;
            // One ledger covers the whole materialized closure. Refreshing
            // metadata ancestors and their descendants separately can assign
            // conflicting timestamps to the same source file.
            pipeline
                .clone()
                .with_remote_path_override(source_container.to_owned())
                .refresh_owned_source_closure(
                    &worker_config,
                    owned_freshness_identity
                        .as_deref()
                        .context("owned source freshness has no captured identity")?,
                    source_container,
                    control,
                    &recovery_identity,
                )
                .await?;
            verify_owned_source_content_roots(
                &worker_config,
                &prepared_source_roots,
                source_identity,
                true,
            )
            .await?;
        } else {
            verify_source_content_roots(&worker_config, &prepared_source_roots, source_identity)
                .await?;
        }
    }
    if let Some(lock) = source_authority_lock.as_mut() {
        lock.ensure_held()?;
    }
    if let Some(lock) = source_pair_lock.as_mut() {
        lock.ensure_held()?;
    }

    let stamped_command = if let Some(before) = build_source_before.as_deref() {
        let after = capture_build_source_stamp(dependency_entry_root, clean_overlay).await;
        let stamp = reconcile_build_source_stamps(before, &after);
        Some(super::cargo_target_dir::bind_build_source_stamp(
            command, &stamp,
        )?)
    } else {
        None
    };
    // Only execution consumes the stamp. `policy_command` still names the
    // original contract already persisted in the recovery recipe above.
    let command = stamped_command.as_deref().unwrap_or(command);
    let guarded_go_command = if let Some(environment) = go_build_environment.as_ref() {
        Some(
            super::artifact_patterns::direct_compiler::go_build_execution_command(
                command,
                environment,
            )
            .context("Go build lost its validated file output contract")?,
        )
    } else {
        None
    };
    let command = guarded_go_command.as_deref().unwrap_or(command);

    // Step 2: Execute command remotely with streaming output
    // Mask sensitive data (API keys, tokens, passwords) before logging
    let masked_command = mask_sensitive_command(command);
    info!("Executing command remotely: {}", masked_command);
    reporter.verbose(&format!("[RCH] exec start: {}", masked_command));

    // Capture stderr for toolchain failure detection
    //
    // `std::env::set_var` is unsafe in Rust 2024, but reading env is fine. For streaming,
    // we need shared mutable state across stdout/stderr callbacks; use `Rc<RefCell<_>>`
    // to avoid borrow-checker conflicts between the two closures.
    use std::cell::{Cell, RefCell};
    use std::rc::Rc;

    let stderr_capture_cell = Rc::new(RefCell::new(String::new()));
    let deadline_triggered = Rc::new(Cell::new(false));

    struct CompileUiState {
        progress: Option<CompilationProgress>,
        output: String,
        output_truncated: bool,
        crates_compiled: Option<u32>,
        warnings: Option<u32>,
    }
    let use_compile_progress = progress_enabled
        && matches!(
            kind,
            Some(
                CompilationKind::CargoBuild
                    | CompilationKind::CargoCheck
                    | CompilationKind::CargoClippy
                    | CompilationKind::CargoDoc
                    | CompilationKind::CargoBench
            )
        );
    let ui_state = Rc::new(RefCell::new(CompileUiState {
        progress: if use_compile_progress {
            Some(CompilationProgress::new(
                output_ctx,
                worker_config.id.as_str().to_string(),
                reporter.visibility == OutputVisibility::None,
            ))
        } else {
            None
        },
        output: String::new(),
        output_truncated: false,
        crates_compiled: None,
        warnings: None,
    }));

    // Add per-worker CARGO_HOME isolation to prevent cache lock contention
    let guarded_command = if owned_source_content_pair {
        Some(guard_owned_source_content_cargo_config(
            command,
            &prepared_source_roots,
        )?)
    } else {
        clean_overlay_cargo
            .then(|| super::cargo_target_dir::guard_clean_overlay_cargo_config(command))
            .transpose()?
    };
    let guarded_command = guarded_command.as_deref().unwrap_or(command);
    let alias_bound_command = build_source_aliases
        .as_ref()
        .map(|aliases| bind_build_source_aliases(guarded_command, aliases));
    let isolated_command = add_cargo_isolation(
        alias_bound_command.as_deref().unwrap_or(guarded_command),
        &worker_config.id,
    );

    // Stream stdout/stderr to our stderr so the agent sees the output
    let command_with_telemetry = wrap_command_with_telemetry(&isolated_command, &worker_config.id);
    let ui_state_stdout = Rc::clone(&ui_state);
    let ui_state_stderr = Rc::clone(&ui_state);
    let stderr_capture_stderr = Rc::clone(&stderr_capture_cell);
    let deadline_triggered_stderr = Rc::clone(&deadline_triggered);
    let deadline_pipeline = pipeline.clone();
    let process_setup_marker = pipeline.remote_process_setup_marker();
    let heartbeat_state_stdout = heartbeat_loop
        .as_ref()
        .map(BuildHeartbeatLoop::shared_state);
    let heartbeat_state_stderr = heartbeat_loop
        .as_ref()
        .map(BuildHeartbeatLoop::shared_state);
    let mut suppress_telemetry = false;
    let defer_cargo_stdout = reporter.cargo_json_stdout;

    if let Some(session) = recovery_session.as_mut() {
        session.starting_execution()?;
    }
    // Heap-pinned: the streaming execution future is large (bd-uz82c).
    let result = Box::pin(pipeline.execute_remote_streaming(
        &worker_config,
        &command_with_telemetry,
        toolchain,
        move |line| {
            if suppress_telemetry {
                return;
            }
            if line.trim() == PIGGYBACK_MARKER {
                suppress_telemetry = true;
                return;
            }
            if let Some(state) = heartbeat_state_stdout.as_ref() {
                mark_heartbeat_progress(state);
            }
            // Do not publish worker paths or a successful terminal record
            // before delivery. The complete supervisor log supplies stdout
            // later; the streamer's bounded preview is not artifact authority.
            if defer_cargo_stdout {
                return;
            }
            if cargo_output_capture
                .as_ref()
                .is_some_and(|capture| capture.suppress_stdout_line(line))
            {
                return;
            }

            let mut state = ui_state_stdout.borrow_mut();
            if let Some(progress) = state.progress.as_mut() {
                progress.update_from_line(line);
                if !state.output_truncated {
                    const MAX_OUTPUT_BYTES: usize = 256 * 1024;
                    if state.output.len() + line.len() <= MAX_OUTPUT_BYTES {
                        state.output.push_str(line);
                    } else {
                        state.output_truncated = true;
                    }
                }
            } else {
                // Write stdout lines to stderr (hook stdout is for protocol)
                eprint!("{}", line);
            }
        },
        move |line| {
            if deadline_pipeline.is_deadline_marker(line) {
                deadline_triggered_stderr.set(true);
                return;
            }
            if line.trim_end_matches(['\r', '\n']) == process_setup_marker {
                return;
            }
            if let Some(state) = heartbeat_state_stderr.as_ref() {
                mark_heartbeat_progress(state);
            }
            // Write stderr lines to stderr and capture for analysis
            let mut state = ui_state_stderr.borrow_mut();
            if let Some(progress) = state.progress.as_mut() {
                progress.update_from_line(line);
                if !state.output_truncated {
                    const MAX_OUTPUT_BYTES: usize = 256 * 1024;
                    if state.output.len() + line.len() <= MAX_OUTPUT_BYTES {
                        state.output.push_str(line);
                    } else {
                        state.output_truncated = true;
                    }
                }
            } else {
                eprint!("{}", line);
            }
            drop(state);

            stderr_capture_stderr.borrow_mut().push_str(line);
        },
    ))
    .await
    .context(crate::transfer::RemoteExecutionUnconfirmed)?;

    // The execution SSH session is separate from the holder. A healthy holder
    // cannot prove Cargo stopped after that transport was lost.
    confirm_source_pair_execution(source_pair_lock.as_mut(), result.exit_code)?;

    if let Some(session) = recovery_session.as_mut() {
        async {
            let completed = read_completion_with_retry(&pipeline, &worker_config)
                .await?
                .context(
                    "SSH exit has no exact durable completion evidence; use jobs recover, never replay",
                )?;
            anyhow::ensure!(
                completed == result.exit_code,
                "SSH status disagrees with durable completion"
            );
            session.completed(completed)?;
            Ok::<(), anyhow::Error>(())
        }
        .await
        .context(crate::transfer::RemoteExecutionUnconfirmed)?;
    }

    // A setup refusal authorizes failover only after the execution and source
    // ownership checks above have established that this attempt has finished.
    if let Err(error) =
        ensure_remote_process_setup_after_completion(&pipeline, &result, source_pair_lock.as_mut())
    {
        if error.is::<crate::transfer::RemoteProcessSetupUnavailable>()
            && let Some(session) = recovery_session.as_mut()
        {
            // This typed refusal follows an exact completion and proves the
            // workload never started. Finish source ownership before allowing
            // the caller's existing worker-failover path to select another job.
            session.returned(result.exit_code)?;
            drop(source_authority_lock.take());
            drop(source_pair_lock.take());
            session.retire_returned().await?;
        }
        return Err(error);
    }

    let stderr_capture = std::mem::take(&mut *stderr_capture_cell.borrow_mut());

    if result.success()
        && let Some(spec) = clean_overlay.filter(|spec| !spec.dependencies.is_empty())
    {
        for entry in &sync_plan {
            let selected = if entry.is_primary {
                spec
            } else {
                &spec
                    .dependencies
                    .iter()
                    .find(|(root, _)| root == &entry.local_root)
                    .ok_or_else(|| anyhow::anyhow!("missing completed dependency binding"))?
                    .1
            };
            reporter.summary_critical(&format!(
                "[RCH] clean-overlay root receipt: local={} remote={} commit={} tree={} overlay-fingerprint={}",
                entry.local_root.display(), entry.remote_root, selected.base_commit,
                selected.tree_object, selected.overlay_fingerprint,
            ));
        }
    }

    info!(
        "Remote command finished: exit={} in {}ms",
        result.exit_code, result.duration_ms
    );
    reporter.verbose(&format!(
        "[RCH] exec done: exit={} in {}ms",
        result.exit_code, result.duration_ms
    ));

    if let Some(proof_build_id) = source_content_build_id {
        let receipt = if owned_source_content_pair {
            finalize_owned_source_content_receipt(
                &worker_config,
                proof_build_id,
                caller_command,
                result.exit_code,
                &prepared_source_roots,
                source_identity,
            )
            .await?
        } else {
            finalize_source_content_receipt(
                &worker_config,
                proof_build_id,
                command,
                result.exit_code,
                &prepared_source_roots,
                source_identity,
            )
            .await?
        };
        if owned_source_content_pair {
            let source_container = overlay_remote_root
                .as_deref()
                .context("owned source verification has no source container")?;
            // A pre-execution ledger alone cannot qualify a warm cache. If
            // execution or its post-proof is interrupted, the next owner must
            // advance all source timestamps beyond any incomplete artifacts.
            pipeline
                .clone()
                .with_remote_path_override(source_container.to_owned())
                .mark_owned_source_closure_verified(
                    &worker_config,
                    owned_freshness_identity
                        .as_deref()
                        .context("owned source verification has no captured identity")?,
                    source_container,
                    source_freshness_control
                        .as_deref()
                        .context("owned source verification has no granted metadata directory")?,
                    &recovery_identity,
                )
                .await?;
        }
        reporter.summary_critical(&format!(
            "[RCH] source content receipt: {}",
            receipt.canonical_json()?
        ));
    }

    {
        let mut state = ui_state.borrow_mut();

        let mut progress_stats = None;
        if let Some(progress) = state.progress.as_mut() {
            progress_stats = Some((progress.crates_compiled(), progress.warnings()));
            if result.success() {
                progress.finish();
            } else {
                let message = stderr_capture
                    .lines()
                    .find(|line| !line.trim().is_empty())
                    .unwrap_or("remote compilation failed");
                progress.finish_error(message);
            }
        }
        if let Some((crates_compiled, warnings)) = progress_stats {
            state.crates_compiled = Some(crates_compiled);
            state.warnings = Some(warnings);
        }

        if use_compile_progress && !result.success() && !state.output.is_empty() {
            eprintln!("{}", state.output);
            if state.output_truncated {
                eprintln!("[RCH] output truncated (increase buffer if needed)");
            }
        }
    }

    let mut artifacts_result: Option<SyncResult> = None;
    let mut artifacts_failed = false;
    let mut cargo_evidence_terminal = false;
    if reporter.cargo_json_stdout {
        recovery_session
            .as_mut()
            .context("caller Cargo diagnostics have no admitted recovery session")?
            .collect_cargo_stderr(&pipeline, &worker_config)
            .await?;
    }
    if result.success()
        && let Some(session) = recovery_session.as_mut()
        && let Some(remote_root) = session.cargo_artifact_remote_root().map(str::to_owned)
    {
        let evidence = if reporter.cargo_json_stdout {
            pipeline
                .read_cargo_stdout_evidence(&worker_config, &remote_root, 0)
                .await
        } else {
            pipeline
                .read_cargo_artifact_evidence(&worker_config, &remote_root)
                .await
        };
        let installed =
            evidence.and_then(|evidence| session.install_complete_cargo_evidence(&evidence));
        if let Err(error) = installed {
            // A completed rejection cannot improve on retry. Lost transport
            // and local journal failures retain the exact attempt for recovery.
            // Neither case permits glob-only publication or compiler replay.
            artifacts_failed = true;
            cargo_evidence_terminal = error.is::<crate::transfer::CargoArtifactEvidenceRejected>()
                || error.is::<recovery::CargoOutputContractRejected>();
            if cargo_evidence_terminal {
                eprintln!(
                    "[RCH] requested Cargo executable evidence is invalid: {error:#}; \
                     no build artifacts were published; delivery failed"
                );
            } else {
                eprintln!(
                    "[RCH] cannot establish the requested Cargo executable set: {error:#}; \
                     no build artifacts were published; use jobs recover for this attempt"
                );
            }
        }
    }
    if reporter.cargo_json_stdout && !result.success() {
        let evidence = pipeline
            .read_cargo_stdout_evidence(&worker_config, &pipeline.remote_path(), result.exit_code)
            .await?;
        recovery_session
            .as_mut()
            .context("failed Cargo stdout has no admitted recovery session")?
            .retain_cargo_stdout(
                &evidence.stdout,
                &evidence.remote_root,
                evidence
                    .remote_project_root
                    .as_deref()
                    .context("caller Cargo JSON has no verified source root")?,
                result.exit_code,
            )?;
    }
    // Per-file evidence from the phase that carries the build's `target/`
    // outputs, for the zero-build-output loud-failure gate (bd-mpbav): the
    // matched-file manifest and the rsync-reported matched regular-file count.
    // `custom_target_basis` records which path basis the manifest uses (paths
    // relative to the remote target dir vs `target/`-prefixed project-root
    // paths); `expected_output_patterns` feeds the failure message.
    let mut retrieval_manifest: Vec<String> = Vec::new();
    let mut retrieval_matched_regular: Option<u32> = None;
    let mut retrieval_custom_target_basis = false;
    let mut expected_output_patterns: Vec<String> = Vec::new();
    // Local directory the retrieval manifest's paths are relative to, so the
    // #65 executable-typing gate can open the files that were actually placed.
    let mut retrieval_local_base: Option<PathBuf> = None;
    // Step 3: Retrieve artifacts
    if result.success() && !artifacts_failed {
        if let Some(loop_ref) = heartbeat_loop.as_ref() {
            loop_ref.update_phase(
                BuildHeartbeatPhase::SyncDown,
                Some("artifact_sync_start".to_string()),
            );
            loop_ref.flush().await;
        }
        // Project-root artifact retrieval. When a custom CARGO_TARGET_DIR is
        // forwarded, the build's `target/` outputs are retrieved exclusively by
        // the custom-target phase below; the project-root phase must not carry
        // `target/`-prefixed patterns, or it re-materializes stale worker-side
        // `<project>/target/` residue onto the local project-root filesystem the
        // custom target dir exists to protect, and a failed stale-residue pull
        // spuriously fails an otherwise-complete build (rch#30). For cargo
        // build/doc/rustc the filtered list is empty, so the phase is skipped.
        let mut artifact_patterns = match recovery_session
            .as_ref()
            .filter(|session| session.has_native_output_contract())
        {
            // The persisted contract determines whether native retrieval is
            // required, including explicit outputs underneath target/.
            Some(session) => session.cargo_artifact_patterns("project")?,
            None => get_project_artifact_patterns(
                kind,
                Some(policy_command),
                forwarded_cargo_target_dir.is_some(),
            ),
        };
        if !artifact_patterns.is_empty() {
            if let Some(session) = recovery_session.as_ref() {
                artifact_patterns = session.cargo_artifact_patterns("project")?;
            }
            let retrieval_pipeline = match recovery_session.as_ref() {
                Some(session) => session.staging_pipeline("project", &pipeline)?,
                None => pipeline.clone(),
            };
            info!("Retrieving build artifacts...");
            reporter.verbose("[RCH] artifacts: retrieving...");
            let heartbeat_state_download = heartbeat_loop
                .as_ref()
                .map(BuildHeartbeatLoop::shared_state);
            let mut download_progress = if progress_enabled {
                Some(TransferProgress::download(
                    output_ctx,
                    "Retrieving artifacts",
                    reporter.visibility == OutputVisibility::None,
                ))
            } else {
                None
            };

            let retrieval = retrieve_with_live_recovery(
                &retrieval_pipeline,
                durable_lease,
                async |retrieval_pipeline| {
                    if let Some(progress) = &mut download_progress {
                        retrieval_pipeline
                            .retrieve_artifacts_streaming(
                                &worker_config,
                                &artifact_patterns,
                                |line| {
                                    progress.update_from_line(line);
                                    if let Some(state) = heartbeat_state_download.as_ref() {
                                        mark_heartbeat_progress(state);
                                    }
                                },
                            )
                            .await
                    } else {
                        retrieval_pipeline
                            .retrieve_artifacts(&worker_config, &artifact_patterns)
                            .await
                    }
                },
            )
            .await;
            let retrieval = match retrieval {
                Ok(artifact_result) => match recovery_session.as_mut() {
                    Some(session) => session.publish("project").await.map(|()| artifact_result),
                    None => Ok(artifact_result),
                },
                Err(error) => Err(error),
            };

            match retrieval {
                Ok(artifact_result) => {
                    info!(
                        "Artifacts retrieved: {} files, {} bytes in {}ms",
                        artifact_result.stats.files_transferred,
                        artifact_result.stats.bytes_transferred,
                        artifact_result.stats.duration_ms
                    );
                    reporter.verbose(&format!(
                        "[RCH] artifacts done: {} files, {} bytes in {}ms",
                        artifact_result.stats.files_transferred,
                        artifact_result.stats.bytes_transferred,
                        artifact_result.stats.duration_ms
                    ));
                    if let Some(progress) = &mut download_progress {
                        progress.apply_summary(
                            artifact_result.stats.bytes_transferred,
                            artifact_result.stats.files_transferred,
                        );
                        progress.finish();
                    }
                    // Default-root basis: this phase carries the `target/`
                    // outputs whenever no custom target dir is forwarded, so
                    // its manifest is the zero-output gate's evidence.
                    if forwarded_cargo_target_dir.is_none() {
                        retrieval_manifest = artifact_result.manifest_regular_files;
                        retrieval_matched_regular = artifact_result.matched_regular_files;
                        retrieval_custom_target_basis = false;
                        expected_output_patterns = expected_output_glob_list(&artifact_patterns);
                        retrieval_local_base = Some(project_root.clone());
                    }
                    artifacts_result = Some(match artifacts_result.take() {
                        Some(existing) => merge_sync_result(&existing, &artifact_result.stats),
                        None => artifact_result.stats,
                    });
                }
                Err(e) => {
                    artifacts_failed = true;

                    // Extract rsync exit code from error message if present
                    let error_str = e.to_string();
                    let rsync_exit_code = error_str.find("exit code").and_then(|_| {
                        error_str
                            .split("exit code")
                            .nth(1)
                            .and_then(|s| s.split(':').next())
                            .and_then(|s| {
                                s.trim()
                                    .trim_start_matches("Some(")
                                    .trim_end_matches(')')
                                    .parse()
                                    .ok()
                            })
                    });

                    // Create structured warning (bd-1q3p)
                    let warning = ArtifactRetrievalWarning::new(
                        worker_config.id.as_str(),
                        artifact_patterns.clone(),
                        &error_str,
                        rsync_exit_code,
                    );

                    warn!("Failed to retrieve artifacts: {}", e);

                    // Show detailed warning in verbose mode or when not in machine mode
                    if !console.is_machine() {
                        reporter.verbose(&warning.format_warning());
                    } else {
                        // For machine mode, output JSON warning
                        debug!("Artifact retrieval warning (JSON): {}", warning.to_json());
                        reporter.verbose("[RCH] artifacts failed (continuing)");
                    }

                    if let Some(progress) = &mut download_progress {
                        progress.finish_error(&e.to_string());
                    }
                    // Continue anyway - compilation succeeded
                }
            }
        } // end: project-root artifact retrieval (skipped when patterns empty)

        if let Some(local_target_dir) = forwarded_cargo_target_dir.as_ref() {
            let remote_target_path = pipeline.remote_cargo_target_dir();
            let mut custom_patterns = if recovery_session
                .as_ref()
                .is_some_and(recovery::RecoverySession::has_native_output_contract)
            {
                // Native compiler outputs belong to the project phase even
                // when the caller forwards an unrelated CARGO_TARGET_DIR.
                Vec::new()
            } else {
                get_custom_target_artifact_patterns(kind, Some(policy_command))
            };
            if custom_patterns.is_empty() {
                reporter.verbose(&format!(
                    "[RCH] custom target dir sync skipped for {} after command with no target artifacts",
                    local_target_dir.display()
                ));
            } else {
                if let Some(session) = recovery_session.as_ref() {
                    custom_patterns = session.cargo_artifact_patterns("target")?;
                }
                let target_pipeline = TransferPipeline::new(
                    local_target_dir.clone(),
                    project_id_from_path(local_target_dir),
                    compute_project_hash_with_dependency_roots_and_policy(
                        local_target_dir,
                        &[],
                        topology_policy,
                    ),
                    transfer_config.clone(),
                )
                .with_color_mode(color_mode)
                .with_command_timeout(command_timeout)
                .with_compilation_config(compilation_config.clone())
                .with_compilation_kind(kind)
                .with_remote_path_override(remote_target_path.clone())
                .with_worker_platform(WorkerPlatform::from_worker(&worker_config));
                let target_pipeline = match recovery_session.as_ref() {
                    Some(session) => session.staging_pipeline("target", &target_pipeline)?,
                    None => target_pipeline,
                };

                let mut target_progress = if progress_enabled {
                    Some(TransferProgress::download(
                        output_ctx,
                        "Syncing custom CARGO_TARGET_DIR artifacts",
                        reporter.visibility == OutputVisibility::None,
                    ))
                } else {
                    None
                };

                let target_retrieval = retrieve_with_live_recovery(
                    &target_pipeline,
                    durable_lease,
                    async |target_pipeline| {
                        if let Some(progress) = &mut target_progress {
                            let heartbeat_state_target = heartbeat_loop
                                .as_ref()
                                .map(BuildHeartbeatLoop::shared_state);
                            target_pipeline
                                .retrieve_artifacts_streaming(
                                    &worker_config,
                                    &custom_patterns,
                                    |line| {
                                        progress.update_from_line(line);
                                        if let Some(state) = heartbeat_state_target.as_ref() {
                                            mark_heartbeat_progress(state);
                                        }
                                    },
                                )
                                .await
                        } else {
                            target_pipeline
                                .retrieve_artifacts(&worker_config, &custom_patterns)
                                .await
                        }
                    },
                )
                .await;
                let target_retrieval = match target_retrieval {
                    Ok(target_result) => match recovery_session.as_mut() {
                        Some(session) => session.publish("target").await.map(|()| target_result),
                        None => Ok(target_result),
                    },
                    Err(error) => Err(error),
                };

                match target_retrieval {
                    Ok(target_result) => {
                        info!(
                            "Custom CARGO_TARGET_DIR artifacts retrieved: {} files, {} bytes in {}ms",
                            target_result.stats.files_transferred,
                            target_result.stats.bytes_transferred,
                            target_result.stats.duration_ms
                        );
                        reporter.verbose(&format!(
                            "[RCH] custom target dir sync done: {} -> {} ({} files, {} bytes in {}ms)",
                            remote_target_path,
                            local_target_dir.display(),
                            target_result.stats.files_transferred,
                            target_result.stats.bytes_transferred,
                            target_result.stats.duration_ms
                        ));
                        if let Some(progress) = &mut target_progress {
                            progress.apply_summary(
                                target_result.stats.bytes_transferred,
                                target_result.stats.files_transferred,
                            );
                            progress.finish();
                        }
                        // The custom-target phase exclusively carries the
                        // build's `target/` outputs under a forwarded
                        // CARGO_TARGET_DIR, so its manifest is the
                        // zero-output gate's evidence (paths relative to the
                        // target-dir sync root).
                        retrieval_manifest = target_result.manifest_regular_files;
                        retrieval_matched_regular = target_result.matched_regular_files;
                        retrieval_custom_target_basis = true;
                        expected_output_patterns = expected_output_glob_list(&custom_patterns);
                        retrieval_local_base = Some(local_target_dir.clone());
                        artifacts_result = Some(match artifacts_result.take() {
                            Some(existing) => merge_sync_result(&existing, &target_result.stats),
                            None => target_result.stats,
                        });
                    }
                    Err(e) => {
                        artifacts_failed = true;
                        warn!("Failed to sync custom CARGO_TARGET_DIR artifacts: {}", e);
                        reporter.verbose(&format!(
                            "[RCH] custom target dir sync failed for {}: {}",
                            local_target_dir.display(),
                            e
                        ));
                        if let Some(progress) = &mut target_progress {
                            progress.finish_error(&e.to_string());
                        }
                    }
                }
            }
        }
    }

    // Step 3b: declared job result directories (bd-p0yoo). Unlike patterned
    // artifact sync-back above, this is NOT gated on `result.success()`:
    // GH#27 exists precisely because non-compilation jobs (sharded tests,
    // fuzzers, mutation testing) produce their valuable output on FAILURE
    // exits too. Each directory is pulled as an explicit rsync source, so a
    // directory the job never created is a hard error rather than a silent
    // zero-file success. Any failure overrides the surfaced exit code below
    // bd-uoh4x: capture per-dir collection outcomes for the machine envelope.
    let mut result_dir_failures: Vec<String> = Vec::new();
    let mut exec_dir_stats: Vec<ExecResultDirStat> = Vec::new();
    if !result_dirs.is_empty() {
        if let Some(loop_ref) = heartbeat_loop.as_ref() {
            loop_ref.update_phase(
                BuildHeartbeatPhase::SyncDown,
                Some("result_dir_sync_start".to_string()),
            );
            loop_ref.flush().await;
        }
        for dir in result_dirs {
            let phase_name = format!("result:{}", dir.display());
            let result_pipeline = match recovery_session.as_ref() {
                Some(session) => session.staging_pipeline(&phase_name, &pipeline)?,
                None => pipeline.clone(),
            };
            match retrieve_with_live_recovery(
                &result_pipeline,
                durable_lease,
                async |result_pipeline| {
                    result_pipeline
                        .retrieve_result_dir(&worker_config, dir)
                        .await
                },
            )
            .await
            {
                Ok(retrieved) => {
                    if let Some(session) = recovery_session.as_mut() {
                        session.publish(&phase_name).await?;
                    }
                    reporter.verbose(&format!(
                        "[RCH] result dir '{}': {} files, {} bytes",
                        dir.display(),
                        retrieved.files_transferred,
                        retrieved.bytes_transferred
                    ));
                    exec_dir_stats.push(ExecResultDirStat {
                        path: dir.display().to_string(),
                        files: u64::from(retrieved.files_transferred),
                        bytes: retrieved.bytes_transferred,
                        status: "ok".to_string(),
                    });
                }
                Err(e) => {
                    warn!(
                        "Declared result dir '{}' could not be retrieved from {}: {}",
                        dir.display(),
                        worker_config.id,
                        e
                    );
                    result_dir_failures.push(format!("{}: {}", dir.display(), e));
                    exec_dir_stats.push(ExecResultDirStat {
                        path: dir.display().to_string(),
                        files: 0,
                        bytes: 0,
                        status: "collection_failed".to_string(),
                    });
                }
            }
        }
    }

    // Step 4: Extract and forward telemetry (piggybacked in stdout)
    let extraction = extract_piggybacked_telemetry(&result.stdout);
    if let Some(error) = extraction.extraction_error {
        warn!("Telemetry extraction failed: {}", error);
    }
    if let Some(telemetry) = extraction.telemetry
        && let Err(e) = send_telemetry(socket_path, TelemetrySource::Piggyback, &telemetry).await
    {
        warn!("Failed to forward telemetry to daemon: {}", e);
    }

    if is_test_kind(kind)
        && let Some(kind) = kind
    {
        let record = TestRunRecord::new(
            project_id.clone(),
            worker_config.id.as_str().to_string(),
            command.to_string(),
            kind,
            result.exit_code,
            result.duration_ms,
        );
        if let Err(e) = send_test_run(socket_path, &record).await {
            warn!("Failed to forward test run telemetry: {}", e);
        }
    }

    let (crates_compiled, output_snapshot) = {
        let state = ui_state.borrow();
        (state.crates_compiled, state.output.clone())
    };

    if feedback_visible {
        render_compile_summary(
            &console,
            output_ctx,
            worker,
            build_id,
            &sync_result,
            result.duration_ms,
            artifacts_result.as_ref(),
            artifacts_failed,
            cache_hit(&sync_result),
            result.success(),
        );
    }

    if result.success() {
        let artifacts_summary = artifacts_result.as_ref().map(|artifact| ArtifactSummary {
            files: u64::from(artifact.files_transferred),
            bytes: artifact.bytes_transferred,
        });
        let target_label = detect_target_label(command, &output_snapshot);

        let summary = CelebrationSummary::new(project_id.clone(), result.duration_ms)
            .worker(worker_config.id.as_str())
            .crates_compiled(crates_compiled)
            .artifacts(artifacts_summary)
            .cache_hit(Some(cache_hit(&sync_result)))
            .target(target_label)
            .quiet(reporter.visibility == OutputVisibility::None);

        CompletionCelebration::new(summary).record_and_render(output_ctx);
    }

    // Construct per-phase timing breakdown
    let timing = CommandTimingBreakdown {
        sync_up: Some(Duration::from_millis(sync_result.duration_ms)),
        exec: Some(Duration::from_millis(result.duration_ms)),
        sync_down: artifacts_result
            .as_ref()
            .map(|ar| Duration::from_millis(ar.duration_ms)),
        ..Default::default()
    };

    if let Some(loop_ref) = heartbeat_loop.take() {
        let detail = if result.success() {
            Some("build_complete".to_string())
        } else {
            Some(format!("build_exit_{}", result.exit_code))
        };
        loop_ref.finish(BuildHeartbeatPhase::Finalize, detail).await;
    }

    // GitHub #65: a worker is chosen for capacity, not platform, so an
    // UNPINNED build dispatched to a foreign-OS worker returns executables the
    // caller's host cannot exec. rsync reports success, the executable bit is
    // set, and the mismatch only surfaces as `exec format error` — possibly
    // hours later, after intervening gates have consumed the artifact. Type the
    // retrieved outputs against the triple the caller's build was actually for.
    // Evidence-only: an unrecognized triple, an unreadable file, or unknown
    // magic yields no findings and never fails a build.
    let pinned_triple = explicit_target_triple_for_command(command);
    let expected_triple = pinned_triple
        .clone()
        .unwrap_or_else(default_host_target_triple);
    // Build-only test/bench outputs belong to the caller. They require the
    // same transfer-failure, metadata-only and foreign-target checks as builds;
    // the execution kind remains unchanged for telemetry and remote execution.
    let artifact_kind = artifact_delivery_kind(kind, Some(policy_command));
    // Only the kinds whose contract is "materialize the caller's runnable
    // outputs under the cargo target tree" are typed. Test/bench/coverage kinds
    // retrieve reports and instrumented trees whose binaries are the WORKER's
    // by design (`target/llvm-cov-target/…`), and unclassified commands prove
    // nothing about what they wrote — matching the zero-output gate's
    // deliberately narrow scope.
    let foreign_artifacts = match retrieval_local_base.as_ref() {
        Some(base)
            if result.success()
                && !artifacts_failed
                && kind_has_enumerable_output_contract(artifact_kind)
                && !foreign_artifact_gate_disabled() =>
        {
            foreign_target_artifacts(
                base,
                &retrieval_manifest,
                retrieval_custom_target_basis,
                &expected_triple,
                pinned_triple.as_deref(),
            )
        }
        _ => Vec::new(),
    };

    // Loud, fatal sync-back failure (issue #19 Fix 1). A remote compile that
    // SUCCEEDED but whose artifacts never came back leaves the local build
    // incomplete — no binary/lib where the agent expects one. Reporting exit 0
    // here is a silent footgun: the agent believes the build succeeded and the
    // missing artifact only surfaces much later. So when the compile succeeded,
    // artifact retrieval failed, AND this kind actually produces transferable
    // artifacts, surface a PROMINENT stderr error and return a non-zero,
    // build-failure-class exit code. The retrieval layer already turns exit-0
    // partial transfers into `TransferError::SyncFailed` (transfer.rs); this
    // propagates that as a non-zero hook exit instead of swallowing it.
    let exit_code = if !result_dir_failures.is_empty() {
        // Declared job result directories (bd-p0yoo) could not be retrieved.
        // This outranks the job's own exit status: a run whose declared
        // outputs are absent or partial is untrustworthy no matter what the
        // command reported, and GH#27 requires reliable result return
        // INCLUDING on nonzero exits. Same loud-fatal treatment as the
        // artifact sync-back failure below (issue #19 Fix 1 precedent).
        let code = ErrorCode::BuildArtifactMissing;
        // stderr, not just `warn!`: this MUST reach the operator/agent even
        // when tracing is silenced. stderr is the diagnostics stream.
        eprintln!(
            "[RCH] {} job on {} exited {} but declared result directories could \
         not be retrieved — the invocation's outputs are INCOMPLETE: {}. \
         Treating as a transfer failure (exit {EXIT_ARTIFACT_TRANSFER_FAILED}).",
            code.code_string(),
            worker_config.id,
            result.exit_code,
            result_dir_failures.join("; "),
        );
        warn!(
            "Result-dir retrieval failed on {} after remote exit {}; \
         returning exit {} so the caller knows declared outputs are missing",
            worker_config.id, result.exit_code, EXIT_ARTIFACT_TRANSFER_FAILED
        );
        EXIT_ARTIFACT_TRANSFER_FAILED
    } else if result.success()
        && artifacts_failed
        && kind_produces_transferable_artifacts(artifact_kind)
    {
        let code = ErrorCode::BuildArtifactMissing;
        // stderr, not just `warn!`: this MUST reach the operator/agent even when
        // tracing is silenced. stderr is the diagnostics stream (AGENTS.md).
        eprintln!(
            "[RCH] {} remote compile on {} SUCCEEDED but build artifacts could not be \
             retrieved — the local build is INCOMPLETE (expected binaries/libraries are \
             missing). Treating as a build failure (exit {EXIT_ARTIFACT_TRANSFER_FAILED}); \
             re-run to rebuild, or check connectivity to the worker.",
            code.code_string(),
            worker_config.id,
        );
        warn!(
            "Artifact transfer failed after a successful remote compile on {} [{}]; \
             returning exit {} so the caller knows the local build is incomplete",
            worker_config.id,
            code.code_string(),
            EXIT_ARTIFACT_TRANSFER_FAILED
        );
        EXIT_ARTIFACT_TRANSFER_FAILED
    } else if result.success()
        && !artifacts_failed
        && (sync_back_verified_zero_build_outputs(
            &retrieval_manifest,
            retrieval_matched_regular,
            artifact_kind,
            retrieval_custom_target_basis,
        ) || sync_back_verified_zero_package_archives(
            retrieval_matched_regular,
            policy_command,
        ))
    {
        // bd-mpbav loud failure, layer B: the sync-back SUCCEEDED (unlike the
        // issue-#19 arm above) yet matched ZERO build outputs — every matched
        // file was loose target metadata or cache state, or package verification
        // returned no archive files at all. The classic cause is
        // an output directory the include patterns don't cover (a custom
        // cargo profile's `target/<profile>/` before layer A added its globs,
        // or any future pattern gap): the remote binary exists, rsync happily
        // pulls 4 metadata files, exits 0, and the LOCAL artifact silently
        // stays the previous build's. Surfacing exit 0 here would let an
        // agent benchmark or ship a stale binary. Same class of loud-fatal
        // treatment as issue #19 Fix 1, with its own error code so the two
        // hazards are distinguishable in logs (E326 vs E309).
        let code = ErrorCode::BuildArtifactSyncEmpty;
        let expected = expected_output_patterns.join(", ");
        // stderr, not just `warn!`: this MUST reach the operator/agent even
        // when tracing is silenced. stderr is the diagnostics stream (AGENTS.md).
        eprintln!(
            "[RCH] {} remote compile on {} SUCCEEDED but the artifact sync-back \
             matched ZERO build outputs ({} of {} matched file(s): {}) — the local \
             build is INCOMPLETE and any existing local artifacts may be STALE. \
             Expected outputs under: {}. Treating as a build failure \
             (exit {EXIT_ARTIFACT_TRANSFER_FAILED}); re-run to retry the sync, or \
             build locally before trusting any binary.",
            code.code_string(),
            worker_config.id,
            retrieval_manifest.len(),
            retrieval_matched_regular.unwrap_or(0),
            retrieval_manifest.join(", "),
            expected,
        );
        warn!(
            "Artifact sync-back matched zero build outputs after a successful remote \
             compile on {} [{}]; returning exit {} so the caller knows local \
             artifacts may be stale",
            worker_config.id,
            code.code_string(),
            EXIT_ARTIFACT_TRANSFER_FAILED
        );
        EXIT_ARTIFACT_TRANSFER_FAILED
    } else if !foreign_artifacts.is_empty() {
        // GitHub #65 loud failure: the sync-back succeeded, but at least one
        // retrieved executable is a container this host cannot run. Surfacing
        // exit 0 would publish an unrunnable binary into a shared target
        // directory where every later existence/executable-bit check stays
        // green. rsync has already overwritten the local artifact by this
        // point, so the recovery instruction is explicit: rebuild.
        let code = ErrorCode::BuildArtifactForeignTarget;
        // stderr, not just `warn!`: this MUST reach the operator/agent even
        // when tracing is silenced. stderr is the diagnostics stream (AGENTS.md).
        eprintln!(
            "[RCH] {} remote compile on {} SUCCEEDED but returned executables for the \
             WRONG PLATFORM: this build targets {} and the retrieved artifact(s) are \
             {}. Offending file(s): {}. The local target directory now holds \
             unrunnable binaries. Treating as a build failure \
             (exit {EXIT_ARTIFACT_TRANSFER_FAILED}); rebuild for this host, pin the \
             build with `--target {}`, or restrict this project to a same-platform \
             worker (set RCH_WORKER, or declare `os = \"<name>\"` for the worker in \
             workers.toml). Set {RCH_ALLOW_FOREIGN_ARTIFACTS_ENV}=1 to accept \
             foreign-platform artifacts deliberately.",
            code.code_string(),
            worker_config.id,
            expected_triple,
            foreign_artifacts
                .first()
                .map_or("of another platform", |f| f.found.label()),
            describe_findings(&foreign_artifacts),
            expected_triple,
        );
        warn!(
            "Retrieved artifacts do not match the requesting host's target triple {} \
             after a successful remote compile on {} [{}]; returning exit {}",
            expected_triple,
            worker_config.id,
            code.code_string(),
            EXIT_ARTIFACT_TRANSFER_FAILED
        );
        EXIT_ARTIFACT_TRANSFER_FAILED
    } else {
        result.exit_code
    };
    let retrieval_complete =
        (!artifacts_failed || cargo_evidence_terminal) && result_dir_failures.is_empty();
    if retrieval_complete && let Some(session) = recovery_session.as_mut() {
        session.returned(exit_code)?;
    }
    let cargo_stdout = if reporter.cargo_json_stdout {
        let session = recovery_session
            .as_mut()
            .context("caller Cargo JSON has no recovery session")?;
        if exit_code == 0 {
            Some(session.caller_cargo_json().await?)
        } else if !result.success() {
            Some(session.failed_cargo_stdout()?)
        } else {
            // The compiler succeeded but delivery failed. Its original
            // success record stays retained, never advertised as local output.
            None
        }
    } else {
        None
    };

    // Retire the materialized source after delivery. Owned source-content
    // pairs move to their exact retained sibling. The durable grant remains
    // outstanding on failure so recovery must finish the same retirement
    // before a subsequent owner can write this source/target pair.
    if retrieval_complete && let Some(overlay_remote_root) = overlay_remote_root.as_deref() {
        if let Some(lock) = source_authority_lock.as_mut() {
            lock.ensure_held()?;
        }
        if let Some(lock) = source_pair_lock.as_mut() {
            lock.ensure_held()?;
        }
        let retirement = if let Some(session) = recovery_session.as_mut() {
            session.retire_source_tree(&pipeline).await
        } else {
            anyhow::ensure!(
                retained_source_root.is_none(),
                "owned source retention requires its persisted recovery session"
            );
            pipeline
                .reap_remote_tree(&worker_config, overlay_remote_root)
                .await
        };
        match retirement {
            Ok(()) => {
                if let Some(retained) = retained_source_root.as_deref() {
                    reporter.verbose(&format!("[RCH] source tree retained at {retained}"));
                } else {
                    reporter.verbose(&format!(
                        "[RCH] clean-overlay remote root {overlay_remote_root} reaped"
                    ));
                }
            }
            Err(error) => {
                return Err(
                    error.context("remote tree retirement remains pending; use jobs recover")
                );
            }
        }
    }
    if retrieval_complete
        && overlay_remote_root.is_none()
        && let Some(session) = recovery_session.as_mut()
    {
        session.tree_retired()?;
    }
    if retrieval_complete && let Some(lock) = source_pair_lock.take() {
        // The explicit acknowledgment also verifies that the root is gone.
        // Never release this lease from an error/Drop path: a remote process
        // or transfer may still be alive after its client disconnects.
        lock.release().await?;
        if let Some(session) = recovery_session.as_mut() {
            session.pair_released()?;
        }
    }
    if retrieval_complete && let Some(lock) = source_authority_lock.take() {
        lock.release().await?;
        if let Some(session) = recovery_session.as_mut() {
            session.sources_released()?;
        }
    }
    if retrieval_complete && let Some(session) = recovery_session.as_mut() {
        session.retire_returned().await?;
    }

    if clean_overlay_cargo
        && result.exit_code == 113
        && stderr_capture.contains("RCH-E413 clean-overlay unselected Cargo configuration")
    {
        return Err(clean_overlay_cargo_policy_failure(
            &normalized_project_root,
            &worker_config.id.to_string(),
            "Worker refused unselected Cargo home or ancestor configuration before starting Cargo"
                .to_owned(),
        )
        .into());
    }

    let mut disk_roots: Vec<String> = sync_plan
        .iter()
        .map(|entry| entry.remote_root.clone())
        .collect();
    disk_roots.push(pipeline.remote_cargo_target_dir());
    disk_roots.sort();
    disk_roots.dedup();
    Ok(RemoteExecutionResult {
        deadline_triggered: exit_code == 137 && deadline_triggered.get(),
        exit_code,
        stderr: stderr_capture,
        duration_ms: result.duration_ms,
        timing,
        result_dirs: exec_dir_stats,
        disk_roots,
        cargo_stdout,
    })
}

#[cfg(test)]
mod tests {
    use super::clean_overlay_remote_project_hash;
    use super::clean_overlay_source_pair_pool_name;
    use super::foreign_artifact_gate_disabled_from_value;
    use super::{CleanOverlaySpec, clean_overlay_freshness_identity};

    #[test]
    fn owned_source_content_cargo_family_preserves_external_driver_receipts() {
        use rch_common::patterns::{CompilationKind, classify_command};

        let _guard = rch_common::test_guard!();
        for (command, expected_kind, supported) in [
            ("cargo build", Some(CompilationKind::CargoBuild), true),
            ("cargo b", Some(CompilationKind::CargoBuild), true),
            ("cargo test", Some(CompilationKind::CargoTest), true),
            ("cargo t", Some(CompilationKind::CargoTest), true),
            ("cargo check", Some(CompilationKind::CargoCheck), true),
            ("cargo c", Some(CompilationKind::CargoCheck), true),
            ("cargo clippy", Some(CompilationKind::CargoClippy), true),
            ("cargo doc", Some(CompilationKind::CargoDoc), true),
            ("cargo bench", Some(CompilationKind::CargoBench), true),
            ("cargo run", Some(CompilationKind::CargoBuild), true),
            ("cargo package", Some(CompilationKind::CargoBuild), true),
            (
                "cargo publish --dry-run",
                Some(CompilationKind::CargoBuild),
                true,
            ),
            (
                "env NOTE=xwin cargo +nightly build",
                Some(CompilationKind::CargoBuild),
                true,
            ),
            (
                "cargo --config=xwin build",
                Some(CompilationKind::CargoBuild),
                true,
            ),
            (
                "cargo --config xwin check",
                Some(CompilationKind::CargoCheck),
                true,
            ),
            (
                "cargo nextest run",
                Some(CompilationKind::CargoNextest),
                false,
            ),
            ("cargo-nextest nextest run", None, false),
            (
                "cargo zigbuild",
                Some(CompilationKind::CargoZigbuild),
                false,
            ),
            (
                "cargo-zigbuild zigbuild",
                Some(CompilationKind::CargoZigbuild),
                false,
            ),
            ("cargo xwin build", Some(CompilationKind::CargoBuild), false),
            ("cargo xwin check", Some(CompilationKind::CargoCheck), false),
            (
                "cargo xwin clippy",
                Some(CompilationKind::CargoClippy),
                false,
            ),
            ("cargo-xwin build", Some(CompilationKind::CargoBuild), false),
            (
                "cargo-xwin xwin check",
                Some(CompilationKind::CargoCheck),
                false,
            ),
            (
                "cargo-xwin clippy",
                Some(CompilationKind::CargoClippy),
                false,
            ),
            (
                "cargo xwin check '--target=x86_64-pc-windows-msvc'",
                Some(CompilationKind::CargoCheck),
                false,
            ),
        ] {
            let classification = classify_command(command);
            assert_eq!(classification.kind, expected_kind, "{command}");
            assert_eq!(
                super::supports_owned_source_content_cargo(classification.kind, command),
                supported,
                "{command}"
            );
        }

        // Family selection must not turn a managed-directory refusal into a
        // fallback to the legacy receipt route.
        for command in [
            "cargo build --build-dir=/unmanaged",
            "cargo --unknown-global build",
        ] {
            let classification = classify_command(command);
            assert_eq!(classification.kind, Some(CompilationKind::CargoBuild));
            assert!(super::supports_owned_source_content_cargo(
                classification.kind,
                command
            ));
            assert!(
                super::super::cargo_target_dir::managed_clean_overlay_cargo_build_dir(
                    command,
                    "/rch-managed-build-dir-validation",
                )
                .is_err(),
                "{command}"
            );
        }
    }

    /// A finished build must not be stranded by one transient SSH failure on
    /// its receipt read; a persistent failure still surfaces after 3 tries,
    /// and an absent receipt is returned at once (it is evidence, not noise).
    #[tokio::test]
    async fn completion_probe_retries_transport_errors_only() {
        use std::cell::Cell;
        let fast = std::time::Duration::from_millis(1);
        let calls = Cell::new(0);
        let flaky = super::retry_completion_probe(fast, async || {
            calls.set(calls.get() + 1);
            if calls.get() < 3 {
                anyhow::bail!("completion probe SSH failed")
            } else {
                Ok(Some(0))
            }
        })
        .await;
        assert_eq!(flaky.unwrap(), Some(0));
        assert_eq!(calls.get(), 3);

        calls.set(0);
        let down = super::retry_completion_probe(fast, async || {
            calls.set(calls.get() + 1);
            anyhow::bail!("completion probe SSH failed")
        })
        .await;
        assert!(down.is_err());
        assert_eq!(calls.get(), 3);

        calls.set(0);
        let absent = super::retry_completion_probe(fast, async || {
            calls.set(calls.get() + 1);
            Ok(None)
        })
        .await;
        assert_eq!(absent.unwrap(), None);
        assert_eq!(calls.get(), 1);
    }

    #[test]
    fn source_pair_freshness_identity_binds_overlays_and_canonical_dependency_closure() {
        let leaf = CleanOverlaySpec {
            base_commit: "a".repeat(40),
            tree_object: "b".repeat(40),
            overlay_paths: Vec::new(),
            overlay_fingerprint: "c".repeat(64),
            dependencies: Vec::new(),
            primary_directory: None,
        };
        let mut source = leaf.clone();
        source.dependencies = vec![("/a".into(), leaf.clone()), ("/b".into(), leaf)];
        let expected = clean_overlay_freshness_identity(&source);
        source.dependencies.reverse();
        assert_eq!(clean_overlay_freshness_identity(&source), expected);
        let mut changed = source.clone();
        changed.dependencies[0].1.overlay_fingerprint = "d".repeat(64);
        assert_ne!(clean_overlay_freshness_identity(&changed), expected);
        changed = source.clone();
        changed.dependencies[0].1.base_commit = "e".repeat(40);
        assert_ne!(clean_overlay_freshness_identity(&changed), expected);
        changed = source.clone();
        changed.overlay_fingerprint = "f".repeat(64);
        assert_ne!(clean_overlay_freshness_identity(&changed), expected);
        changed = source;
        changed.dependencies[0].0 = "/different".into();
        assert_ne!(clean_overlay_freshness_identity(&changed), expected);
    }

    #[test]
    fn source_pair_pool_migrates_legacy_artifacts_and_keeps_pool_isolation() {
        let legacy = ".rch-target-worker-pool-0123456789abcdef0123456789abcdef";
        let paired = clean_overlay_source_pair_pool_name(legacy, "/tmp/rch");
        assert_ne!(paired, legacy);
        assert_eq!(
            paired,
            clean_overlay_source_pair_pool_name(legacy, "/tmp/rch/")
        );
        assert_ne!(
            paired,
            clean_overlay_source_pair_pool_name(legacy, "/data/rch")
        );
        assert!(paired.starts_with(".rch-target-worker-pool-"));
        let hash = paired.rsplit_once('-').unwrap().1;
        assert_eq!(hash.len(), 32);
        assert!(hash.bytes().all(|byte| byte.is_ascii_hexdigit()));
        assert_ne!(
            paired,
            clean_overlay_source_pair_pool_name(
                ".rch-target-worker-pool-fedcba9876543210fedcba9876543210",
                "/tmp/rch"
            )
        );
        assert_ne!(
            paired,
            clean_overlay_source_pair_pool_name(
                ".rch-target-other-pool-0123456789abcdef0123456789abcdef",
                "/tmp/rch"
            )
        );
    }

    #[test]
    fn foreign_artifact_gate_is_on_unless_explicitly_disabled() {
        // Absent or falsey: the gate stays armed.
        assert!(!foreign_artifact_gate_disabled_from_value(None));
        for value in ["", "  ", "0", "false", "FALSE", "no", "off", "Off"] {
            assert!(
                !foreign_artifact_gate_disabled_from_value(Some(value.to_string())),
                "{value:?} must not disable the foreign-artifact gate"
            );
        }
        // Any other value is an explicit opt-out.
        for value in ["1", "true", "yes", "on", "please"] {
            assert!(
                foreign_artifact_gate_disabled_from_value(Some(value.to_string())),
                "{value:?} must disable the foreign-artifact gate"
            );
        }
    }

    #[test]
    fn clean_overlay_concurrent_jobs_use_distinct_remote_roots() {
        let base = "0123456789abcdef0123456789abcdef01234567";
        let first = std::thread::spawn(move || {
            clean_overlay_remote_project_hash(
                base,
                "first-dirty-overlay-fingerprint",
                uuid::Uuid::from_u128(1),
            )
        });
        let second = std::thread::spawn(move || {
            clean_overlay_remote_project_hash(
                base,
                "second-conflicting-overlay-fingerprint",
                uuid::Uuid::from_u128(2),
            )
        });

        let first_root = first.join().expect("first clean-overlay job panicked");
        let second_root = second.join().expect("second clean-overlay job panicked");
        assert_ne!(
            first_root, second_root,
            "concurrent clean-overlay jobs must never share a remote root"
        );

        let fixed_nonce = uuid::Uuid::from_u128(3);
        assert_ne!(
            clean_overlay_remote_project_hash(base, "first-dirty-overlay-fingerprint", fixed_nonce,),
            clean_overlay_remote_project_hash(
                "fedcba9876543210fedcba9876543210fedcba98",
                "first-dirty-overlay-fingerprint",
                fixed_nonce,
            ),
            "the immutable base must be part of the remote-root identity"
        );
        assert_ne!(
            clean_overlay_remote_project_hash(base, "first-dirty-overlay-fingerprint", fixed_nonce,),
            clean_overlay_remote_project_hash(
                base,
                "second-conflicting-overlay-fingerprint",
                fixed_nonce,
            ),
            "the dirty overlay fingerprint must be part of the remote-root identity"
        );
    }

    /// Issue #60 regression: the pooled target store must resolve to the SAME
    /// absolute path for two clean-overlay executions whose remote roots
    /// differ only by job nonce — and that path must sit OUTSIDE both
    /// (teardown-reaped) roots, while keeping the `.rch-target-…-pool-…`
    /// naming GC conventions rely on.
    #[test]
    fn clean_overlay_pooled_target_dir_is_stable_across_job_nonces() {
        use super::stable_pooled_target_dir;

        let base_commit = "0123456789abcdef0123456789abcdef01234567";
        let fingerprint = "same-overlay-fingerprint";
        let hash_a =
            clean_overlay_remote_project_hash(base_commit, fingerprint, uuid::Uuid::from_u128(10));
        let hash_b =
            clean_overlay_remote_project_hash(base_commit, fingerprint, uuid::Uuid::from_u128(11));
        assert_ne!(hash_a, hash_b, "job nonces must keep remote roots distinct");

        let remote_base = "/data/tmp/rch";
        let project_id = "myproject";
        let root_a = format!("{remote_base}/{project_id}/{hash_a}");
        let root_b = format!("{remote_base}/{project_id}/{hash_b}");

        let pooled_name = ".rch-target-w1-pool-0123456789abcdef0123456789abcdef";
        let pool_a = stable_pooled_target_dir(remote_base, project_id, pooled_name);
        let pool_b = stable_pooled_target_dir(remote_base, project_id, pooled_name);

        assert_eq!(
            pool_a, pool_b,
            "pool location must not vary with the job nonce"
        );
        assert!(
            !pool_a.starts_with(&root_a) && !pool_a.starts_with(&root_b),
            "pool must live outside every per-command root: {pool_a}"
        );
        assert!(
            pool_a.contains("/.rch-target-") && pool_a.contains("-pool-"),
            "pool must keep the GC-recognized .rch-target-…-pool-… naming: {pool_a}"
        );
        // Trailing-slash remote_base normalizes identically.
        assert_eq!(
            stable_pooled_target_dir("/data/tmp/rch/", project_id, pooled_name),
            pool_a
        );
    }

    /// Issue #64: the default (no `store_base`, no clean overlay) keeps the
    /// pooled store inside the project mirror — nobody's paths move until the
    /// setting is written.
    #[test]
    fn pooled_store_stays_in_the_mirror_without_a_store_base() {
        use super::pooled_store_base_for;

        assert_eq!(
            pooled_store_base_for(false, None, false, "/data/tmp/rch"),
            None
        );
    }

    /// Issue #64: a configured `store_base` governs EVERY build, not just
    /// clean-overlay runs, and outranks `[transfer] remote_base`.
    #[test]
    fn store_base_governs_ordinary_and_clean_overlay_builds() {
        use super::pooled_store_base_for;

        assert_eq!(
            pooled_store_base_for(false, Some("/bigdisk/rch-pools"), false, "/data/tmp/rch"),
            Some("/bigdisk/rch-pools")
        );
        assert_eq!(
            pooled_store_base_for(false, Some("/bigdisk/rch-pools"), true, "/data/tmp/rch"),
            Some("/bigdisk/rch-pools")
        );
        // Trailing slashes normalize, and an empty value is treated as unset.
        assert_eq!(
            pooled_store_base_for(false, Some("/bigdisk/rch-pools/"), false, "/data/tmp/rch"),
            Some("/bigdisk/rch-pools")
        );
        assert_eq!(
            pooled_store_base_for(false, Some(""), false, "/data/tmp/rch"),
            None
        );
    }

    /// Issue #60 stays intact: without a `store_base`, a clean-overlay run
    /// still relocates its pool under `[transfer] remote_base`.
    #[test]
    fn clean_overlay_without_store_base_uses_transfer_remote_base() {
        use super::pooled_store_base_for;

        assert_eq!(
            pooled_store_base_for(false, None, true, "/data/tmp/rch/"),
            Some("/data/tmp/rch")
        );
    }

    /// A Unix `store_base` is not a valid path on a Windows worker, which
    /// keeps its drive-letter build base.
    #[test]
    fn windows_workers_ignore_the_unix_store_base() {
        use super::pooled_store_base_for;

        assert_eq!(
            pooled_store_base_for(true, Some("/bigdisk/rch-pools"), false, "/data/tmp/rch"),
            None
        );
        assert_eq!(
            pooled_store_base_for(true, Some("/bigdisk/rch-pools"), true, "/data/tmp/rch"),
            Some(crate::transfer::WINDOWS_DEFAULT_REMOTE_BASE)
        );
    }

    /// The pooled path under a `store_base` keeps the GC-recognized basename
    /// and is per-project, so two repos cannot collide.
    #[test]
    fn store_base_pooled_paths_are_per_project_and_gc_recognized() {
        use super::stable_pooled_target_dir;

        let name = ".rch-target-w1-pool-0123456789abcdef0123456789abcdef";
        let a = stable_pooled_target_dir("/bigdisk/rch-pools", "repo-a", name);
        let b = stable_pooled_target_dir("/bigdisk/rch-pools", "repo-b", name);
        assert_eq!(
            a,
            "/bigdisk/rch-pools/repo-a/{name}".replace("{name}", name)
        );
        assert_ne!(a, b);
        assert!(a.contains("/.rch-target-") && a.contains("-pool-"));
    }
}