rchd 2.1.20

//! Build cancellation orchestrator with deterministic state machine (bd-1yt6).
//!
//! Provides bounded escalation (SIGTERM → remote kill → SIGKILL),
//! deterministic cleanup (slots, history, events), and per-worker
//! cancellation debt for reliability integration.

mod batch;

use crate::DaemonContext;
use crate::api::{CancelAllBuildsResponse, CancelBuildResponse, CancelledBuildInfo};
use crate::events::EventBus;
use crate::workers::WorkerEndpointSnapshot;
use rch_common::{
    BuildCancellationMetadata, BuildCancellationWorkerHealth, WorkerId, WorkerStatus,
};
use serde::Serialize;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use tracing::{debug, warn};

fn duration_millis_u64(duration: Duration) -> u64 {
    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}

// ── Cancel Reason ────────────────────────────────────────────────────────

/// Why a build cancellation was initiated.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CancelReason {
    /// Explicit user/agent request.
    User,
    /// Build exceeded its timeout.
    Timeout,
    /// Stuck detector determined build is stuck.
    StuckDetector,
    /// Build was evicted from the queue.
    QueueEviction,
}

impl std::fmt::Display for CancelReason {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::User => write!(f, "user"),
            Self::Timeout => write!(f, "timeout"),
            Self::StuckDetector => write!(f, "stuck_detector"),
            Self::QueueEviction => write!(f, "queue_eviction"),
        }
    }
}

// ── Cancellation State ───────────────────────────────────────────────────

/// State machine for an individual build cancellation lifecycle.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CancellationState {
    /// Cancel has been requested but not yet acted on.
    Requested,
    /// SIGTERM sent to local hook process, waiting for grace period.
    TermSent,
    /// SSH kill sent to remote worker process.
    RemoteKillSent,
    /// Escalated to SIGKILL locally after remote kill failed.
    Escalated,
    /// Cancellation completed: slots released, history updated, event emitted.
    Completed,
    /// Termination was not confirmed; active history and reservations are retained.
    Failed,
}

impl std::fmt::Display for CancellationState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Requested => write!(f, "requested"),
            Self::TermSent => write!(f, "term_sent"),
            Self::RemoteKillSent => write!(f, "remote_kill_sent"),
            Self::Escalated => write!(f, "escalated"),
            Self::Completed => write!(f, "completed"),
            Self::Failed => write!(f, "failed"),
        }
    }
}

// ── Cancellation Record ──────────────────────────────────────────────────

/// Tracks the lifecycle of a single build cancellation.
#[derive(Debug, Clone, Serialize)]
pub struct CancellationRecord {
    pub build_id: u64,
    pub worker_id: String,
    /// Immutable admitted coordinates; current inventory is not execution ownership.
    #[serde(skip)]
    worker_endpoint: Option<WorkerEndpointSnapshot>,
    pub state: CancellationState,
    pub reason: CancelReason,
    #[serde(skip)]
    pub requested_at: Instant,
    #[serde(skip)]
    pub completed_at: Option<Instant>,
    pub escalation_count: u32,
    pub remote_kill_attempted: bool,
    pub cleanup_ok: bool,
    /// Slots originally held by the build (captured at cancel time).
    pub slots: u32,
    pub slots_released: u32,
    pub hook_pid: u32,
    /// Boot+start evidence binding the recorded PID to its owning wrapper.
    #[serde(skip)]
    pub hook_process_identity: Option<String>,
    pub remote_pgid_file: Option<String>,
    /// The reservation was abandoned (see [`provably_abandoned_unlaunched`]):
    /// without a record path it needs no remote kill; with one, a record the
    /// worker no longer has confirms that nothing runs under it.
    #[serde(skip)]
    pub abandoned_unlaunched: bool,
    /// Current coordinates of `worker_id`, used only to ask an abandoned
    /// legacy build's worker for its record (see [`legacy_probe_config`]).
    #[serde(skip)]
    legacy_probe_config: Option<rch_common::WorkerConfig>,
}

/// Silence far past every remote lifetime cap: the in-session watchdog (30
/// min default) and the worker orphan reaper (240 min).
const ABANDONED_UNLAUNCHED_AFTER: Duration = Duration::from_secs(6 * 3600);

/// The same judgement for a build that never started its remote command (its
/// heartbeat phase is still `sync_up`). Nothing can be running remotely for
/// it, so there is no remote lifetime cap to wait out once the wrapper is
/// provably gone. A wrapper SIGKILLed in sync_up otherwise held its slots for
/// the full 6h (bd-axhoi: 4 slots on vmi1152480 for 4.6h, the stuck
/// detector's cancellation unable to confirm a remote kill for a build that
/// had no remote process).
const ABANDONED_BEFORE_REMOTE_AFTER: Duration = Duration::from_secs(15 * 60);

/// A reservation whose wrapper is gone (or was never recorded) and that has
/// been silent far past every remote lifetime cap is abandoned. With no
/// process record path it can never have launched anything; with one (the
/// hook reports the path before launch), a record the worker no longer has
/// proves nothing runs under it. Demanding a confirmed remote kill for these
/// held their slots forever: 15 such ghosts held ~40 slots on one dispatcher
/// and survived restarts because ownership is durable. A lease that still
/// owns worker source keeps it: its source claim fences the worker until
/// `rch jobs recover` retires it.
fn provably_abandoned_unlaunched(
    build: &crate::history::ActiveBuildState,
    now: Instant,
    lease_owns_source: impl FnOnce(&str) -> bool,
) -> bool {
    let silence = if build.remote_command_started() {
        ABANDONED_UNLAUNCHED_AFTER
    } else {
        ABANDONED_BEFORE_REMOTE_AFTER
    };
    now.saturating_duration_since(build.started_at_mono) >= silence
        && now.saturating_duration_since(build.last_heartbeat_mono) >= silence
        && wrapper_process_state(build.hook_pid, build.hook_process_identity.as_deref())
            == WrapperProcessState::Exited
        && !build
            .local_wrapper_id
            .as_deref()
            .is_some_and(lease_owns_source)
}

/// Whether this dispatcher's client lease for `wrapper` still owns worker
/// source. An unreadable lease fails closed (treated as owning).
fn client_lease_owns_source(wrapper: &str) -> bool {
    let path =
        rch_common::job_identity::default_job_lease_directory().join(format!("{wrapper}.json"));
    match std::fs::read(&path) {
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => false,
        Err(_) => true,
        Ok(bytes) => serde_json::from_slice::<rch_common::job_identity::DurableJobLease>(&bytes)
            .map_or(true, |lease| {
                crate::api::lease_owns_unretired_source(&lease)
            }),
    }
}

/// Ownership snapshots written before admitted endpoints were recorded
/// recover their builds without one, so the remote probe could never run and
/// the stuck detector retried an abandoned build every tick, forever
/// (bd-ilds6: two such builds held a slot each for 30h on one dispatcher,
/// across a restart). Only such an abandoned build may ask its worker ID's
/// current endpoint. That answer never becomes execution ownership: cleanup
/// and feedback still see no endpoint.
async fn legacy_probe_config(
    ctx: &DaemonContext,
    record: &CancellationRecord,
) -> Option<rch_common::WorkerConfig> {
    if record.worker_endpoint.is_some() || !record.abandoned_unlaunched {
        return None;
    }
    let worker = ctx.pool.get(&WorkerId::new(&record.worker_id)).await?;
    Some(worker.endpoint_snapshot().await.config)
}

/// A recycled PID proves that the original process has exited; it never
/// authorizes signalling the current occupant. Missing observation is distinct.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum WrapperProcessState {
    Running,
    Exited,
    Unverified,
}

/// Only precise native identities are authoritative. Old Darwin ps display
/// strings cannot distinguish same-second PID reuse and remain unverified.
#[cfg(test)]
fn well_formed_process_identity(identity: &str) -> bool {
    rch_common::process_identity::ProcessIdentity::from_record(identity).is_some()
}

fn wrapper_process_state(pid: u32, expected: Option<&str>) -> WrapperProcessState {
    if pid == 0 {
        return WrapperProcessState::Exited;
    }
    if pid <= 1 || i32::try_from(pid).is_err() {
        return WrapperProcessState::Unverified;
    }
    use rch_common::process_identity::{OwnerPresence, ProcessIdentity, owner_presence};
    let expected = expected.and_then(ProcessIdentity::from_record);
    match owner_presence(pid, expected.as_ref()) {
        OwnerPresence::Live => WrapperProcessState::Running,
        OwnerPresence::Absent => WrapperProcessState::Exited,
        OwnerPresence::Unknown => WrapperProcessState::Unverified,
    }
}

#[derive(Debug, Clone, Serialize)]
struct CancellationWorkerHealthSnapshot {
    status: String,
    speed_score: f64,
    used_slots: u32,
    available_slots: u32,
    pressure_state: String,
    pressure_reason_code: String,
}

fn cancellation_operation_id(build_id: u64) -> String {
    format!("cancel-{build_id}")
}

fn worker_status_label(status: WorkerStatus) -> &'static str {
    match status {
        WorkerStatus::Healthy => "healthy",
        WorkerStatus::Degraded => "degraded",
        WorkerStatus::Unreachable => "unreachable",
        WorkerStatus::Draining => "draining",
        WorkerStatus::Drained => "drained",
        WorkerStatus::Disabled => "disabled",
    }
}

fn worker_health_for_history(
    snapshot: &CancellationWorkerHealthSnapshot,
) -> BuildCancellationWorkerHealth {
    BuildCancellationWorkerHealth {
        status: snapshot.status.clone(),
        speed_score: snapshot.speed_score,
        used_slots: snapshot.used_slots,
        available_slots: snapshot.available_slots,
        pressure_state: snapshot.pressure_state.clone(),
        pressure_reason_code: snapshot.pressure_reason_code.clone(),
    }
}

fn push_decision_stage(path: &mut Vec<&'static str>, stage: &'static str) {
    if path.last().copied() != Some(stage) {
        path.push(stage);
    }
}

fn cancellation_decision_path(record: &CancellationRecord) -> Vec<&'static str> {
    let mut path = vec!["requested"];

    let force_path = record.remote_kill_attempted
        && record.escalation_count == 0
        && matches!(
            record.state,
            CancellationState::Completed | CancellationState::Failed | CancellationState::Escalated
        );

    if force_path {
        push_decision_stage(&mut path, "escalated");
        push_decision_stage(&mut path, "remote_kill_sent");
    } else {
        push_decision_stage(&mut path, "term_sent");
        if record.remote_kill_attempted {
            push_decision_stage(&mut path, "remote_kill_sent");
        }
        if record.escalation_count > 1 || matches!(record.state, CancellationState::Escalated) {
            push_decision_stage(&mut path, "escalated");
        }
    }

    let terminal = match record.state {
        CancellationState::Completed => "completed",
        CancellationState::Failed => "failed",
        CancellationState::Requested => "requested",
        CancellationState::TermSent => "term_sent",
        CancellationState::RemoteKillSent => "remote_kill_sent",
        CancellationState::Escalated => "escalated",
    };
    push_decision_stage(&mut path, terminal);

    path
}

fn cancellation_escalation_stage(record: &CancellationRecord) -> &'static str {
    if record.escalation_count > 1 || matches!(record.state, CancellationState::Escalated) {
        "sigkill"
    } else if record.remote_kill_attempted {
        "remote_kill"
    } else {
        "term"
    }
}

/// Sending a signal is not evidence that both sides of a remote build stopped.
fn cancellation_terminal_state(local_stopped: bool, remote_stopped: bool) -> CancellationState {
    if local_stopped && remote_stopped {
        CancellationState::Completed
    } else {
        CancellationState::Failed
    }
}

// ── Configuration ────────────────────────────────────────────────────────

/// Policy knobs for the cancellation orchestrator.
#[derive(Debug, Clone)]
pub struct CancellationConfig {
    /// How long to wait after SIGTERM before escalating.
    pub grace_period: Duration,
    /// How long to wait for SIGKILL to take effect.
    pub kill_timeout: Duration,
    /// Timeout for SSH kill command to remote worker.
    pub remote_kill_timeout: Duration,
    /// Maximum number of non-forced escalation stages attempted.
    pub max_escalations: u32,
    /// Overall timeout for termination stages, including forced cancellation.
    /// Accounting and failure reporting follow without freeing unconfirmed work.
    pub cleanup_timeout: Duration,
}

impl Default for CancellationConfig {
    fn default() -> Self {
        Self {
            grace_period: Duration::from_secs(5),
            kill_timeout: Duration::from_secs(3),
            remote_kill_timeout: Duration::from_secs(10),
            max_escalations: 3,
            cleanup_timeout: Duration::from_secs(15),
        }
    }
}

// ── Per-Worker Debt Tracker ──────────────────────────────────────────────

/// Tracks cancellation frequency per worker for reliability integration.
/// All counters use timestamped vectors pruned to the DEBT_WINDOW so stale
/// events do not permanently inflate a worker's cancellation debt.
#[derive(Debug, Clone, Default)]
struct WorkerCancelStats {
    /// Recent cancellation timestamps (within window).
    recent_cancellations: Vec<Instant>,
    /// Recent escalation timestamps (within window).
    recent_escalations: Vec<Instant>,
    /// Recent cleanup failure timestamps (within window).
    recent_cleanup_failures: Vec<Instant>,
    /// Total builds observed (for rate computation).
    total_builds: u64,
}

const DEBT_WINDOW: Duration = Duration::from_secs(300); // 5 minutes

// ── Orchestrator ─────────────────────────────────────────────────────────

/// Drives build cancellations through a deterministic state machine with
/// bounded escalation and cleanup guarantees. Clones share admission and debt.
#[derive(Clone)]
pub struct CancellationOrchestrator {
    config: CancellationConfig,
    /// Short, synchronous map operations only; never held across an await.
    /// A synchronous lock lets an attempt release its claim in Drop.
    active: Arc<Mutex<HashMap<u64, CancellationRecord>>>,
    /// Per-worker cancellation debt tracking.
    worker_stats: Arc<RwLock<HashMap<String, WorkerCancelStats>>>,
    /// Bound bulk work across concurrent requests and orchestrator clones.
    bulk_permits: Arc<tokio::sync::Semaphore>,
    /// Event bus for structured event emission.
    events: EventBus,
}

/// Owns only the in-flight claim, never the build reservation. Dropping an
/// interrupted operation must permit a retry, not certify remote termination.
struct CancellationAttempt {
    build_id: u64,
    active: Arc<Mutex<HashMap<u64, CancellationRecord>>>,
}

impl Drop for CancellationAttempt {
    fn drop(&mut self) {
        self.active
            .lock()
            .unwrap_or_else(|error| error.into_inner())
            .remove(&self.build_id);
    }
}

impl CancellationOrchestrator {
    /// Create a new orchestrator with the given config and event bus.
    pub fn new(config: CancellationConfig, events: EventBus) -> Self {
        Self {
            config,
            active: Arc::new(Mutex::new(HashMap::new())),
            worker_stats: Arc::new(RwLock::new(HashMap::new())),
            bulk_permits: Arc::new(tokio::sync::Semaphore::new(batch::MAX_BULK_CANCELLATIONS)),
            events,
        }
    }

    /// Admit one cancellation and await its result. Once admitted, the daemon
    /// task owns termination AND finalization; cancelling this wait must not
    /// abandon remote cleanup or interrupt a claimed history record's release.
    /// This survives caller cancellation, not shutdown of the daemon runtime.
    pub async fn cancel_build(
        &self,
        ctx: &DaemonContext,
        build_id: u64,
        reason: CancelReason,
        force: bool,
    ) -> CancelBuildResponse {
        let (mut record, project_id, attempt) = {
            let mut active = self
                .active
                .lock()
                .unwrap_or_else(|error| error.into_inner());
            if let Some(existing) = active.get(&build_id) {
                return CancelBuildResponse {
                    status: "cancelling".to_string(),
                    build_id,
                    worker_id: Some(existing.worker_id.clone()),
                    project_id: ctx
                        .history
                        .active_build(build_id)
                        .map(|build| build.project_id),
                    message: Some(format!(
                        "Cancellation already in progress (state: {})",
                        existing.state
                    )),
                    slots_released: existing.slots_released,
                };
            }
            // Resolve history after acquiring admission, not before an async
            // lock wait during which a prior cancellation may have finalized it.
            let Some(build) = ctx.history.active_build(build_id) else {
                return CancelBuildResponse {
                    status: "error".to_string(),
                    build_id,
                    worker_id: None,
                    project_id: None,
                    message: Some("Build not found or already completed".to_string()),
                    slots_released: 0,
                };
            };
            let abandoned_unlaunched =
                provably_abandoned_unlaunched(&build, Instant::now(), client_lease_owns_source);
            let record = CancellationRecord {
                build_id,
                worker_id: build.worker_id,
                worker_endpoint: build.worker_endpoint,
                state: CancellationState::Requested,
                reason,
                requested_at: Instant::now(),
                completed_at: None,
                escalation_count: 0,
                remote_kill_attempted: false,
                cleanup_ok: true,
                slots: build.slots,
                slots_released: 0,
                hook_pid: build.hook_pid,
                hook_process_identity: build.hook_process_identity.clone(),
                abandoned_unlaunched,
                remote_pgid_file: build.remote_pgid_file,
                legacy_probe_config: None,
            };
            active.insert(build_id, record.clone());
            (
                record,
                build.project_id,
                CancellationAttempt {
                    build_id,
                    active: Arc::clone(&self.active),
                },
            )
        };

        let owner = self.clone();
        let context = ctx.clone();
        let failed_worker = record.worker_id.clone();
        let failed_project = project_id.clone();
        // There is no await between admission and transferring the guard into
        // the owned task. Dropping a JoinHandle detaches, rather than aborts,
        // the operation. Duplicate requests still share the same admission map.
        let operation = tokio::spawn(async move {
            let _attempt = attempt;
            // A normal completion may have won before this task was scheduled.
            // Do not signal the stale hook from the admission-time snapshot.
            let Some(current) = context.history.active_build(build_id) else {
                return CancelBuildResponse {
                    status: "error".to_string(),
                    build_id,
                    worker_id: Some(record.worker_id),
                    project_id: Some(project_id),
                    message: Some("Build completed before cancellation started".to_string()),
                    slots_released: 0,
                };
            };
            // Proven exit of the original wrapper permits remote recovery.
            // A reused PID is never adopted or signalled; uncertainty still
            // retains ownership rather than treating a failed probe as exit.
            if wrapper_process_state(current.hook_pid, current.hook_process_identity.as_deref())
                == WrapperProcessState::Unverified
            {
                return CancelBuildResponse {
                    status: "failed".into(),
                    build_id,
                    worker_id: Some(record.worker_id),
                    project_id: Some(project_id),
                    message: Some(
                        "Wrapper process identity unverified; reservation retained".into(),
                    ),
                    slots_released: 0,
                };
            }
            record.hook_pid = current.hook_pid;
            record.worker_endpoint = current.worker_endpoint;
            record.remote_pgid_file = current.remote_pgid_file;
            record.hook_process_identity = current.hook_process_identity.clone();
            record.slots = current.slots;
            record.legacy_probe_config = legacy_probe_config(&context, &record).await;

            owner.events.emit(
                "cancellation_requested",
                &serde_json::json!({
                    "build_id": build_id,
                    "worker_id": record.worker_id,
                    "project_id": project_id,
                    "reason": reason,
                    "force": force,
                }),
            );
            owner.execute_cancellation(&mut record, force).await;
            // Keep finalization in this same task. Caller cancellation after
            // confirmation must not lose its slot-release/history owner.
            owner.run_cleanup(&context, &mut record).await;
            owner.record_cancellation_stats(&record).await;

            CancelBuildResponse {
                status: if record.state == CancellationState::Completed {
                    "cancelled"
                } else {
                    "failed"
                }
                .to_string(),
                build_id,
                worker_id: Some(record.worker_id),
                project_id: Some(project_id),
                message: Some(match (record.state, force) {
                    (CancellationState::Completed, true) => {
                        "Build forcefully terminated".to_string()
                    }
                    (CancellationState::Completed, false) => {
                        "Build cancellation completed".to_string()
                    }
                    _ => {
                        "Cancellation unconfirmed; active build and reservations retained for retry"
                            .to_string()
                    }
                }),
                slots_released: record.slots_released,
            }
        });
        match operation.await {
            Ok(response) => response,
            Err(error) => {
                warn!(build_id, %error, "Cancellation task ended unexpectedly");
                CancelBuildResponse {
                    status: "failed".to_string(),
                    build_id,
                    worker_id: Some(failed_worker),
                    project_id: Some(failed_project),
                    message: Some(
                        "Cancellation task ended unexpectedly; check active history before retrying"
                            .to_string(),
                    ),
                    slots_released: 0,
                }
            }
        }
    }

    /// Cancel the current snapshot with bounded, daemon-owned concurrency.
    /// Caller interruption does not abandon later builds in the snapshot.
    pub async fn cancel_all_builds(
        &self,
        ctx: &DaemonContext,
        force: bool,
    ) -> CancelAllBuildsResponse {
        batch::cancel_all_builds(self, ctx, force).await
    }

    /// One budget bounds all termination stages, including lock waits and force.
    async fn execute_cancellation(&self, record: &mut CancellationRecord, force: bool) {
        let deadline = Instant::now().checked_add(self.config.cleanup_timeout);
        if self.config.cleanup_timeout.is_zero() || deadline.is_none() {
            record.state = CancellationState::Failed;
            record.cleanup_ok = false;
            return;
        }
        let deadline = deadline.expect("checked cancellation deadline");
        let timed_out = tokio::time::timeout_at(
            tokio::time::Instant::from_std(deadline),
            self.execute_cancellation_stages(record, force, deadline),
        )
        .await
        .is_err();
        if timed_out || record.state != CancellationState::Completed {
            record.state = CancellationState::Failed;
            record.cleanup_ok = false;
            warn!(
                "Cancellation of build {} unconfirmed (timed_out={}); retaining active state",
                record.build_id, timed_out
            );
        }
    }

    /// Open the stable kernel handle before validating the recorded identity.
    /// Exit/PID reuse after validation cannot redirect pidfd_send_signal.
    #[cfg(target_os = "linux")]
    fn verified_hook_handle(record: &CancellationRecord) -> Option<std::os::fd::OwnedFd> {
        use rustix::process::{Pid, PidfdFlags, pidfd_open};

        let pid = i32::try_from(record.hook_pid).ok().filter(|pid| *pid > 1)?;
        let expected = record.hook_process_identity.as_deref()?;
        let handle = pidfd_open(Pid::from_raw(pid)?, PidfdFlags::empty()).ok()?;
        if crate::history::process_identity(record.hook_pid).as_deref() != Some(expected) {
            return None;
        }
        Some(handle)
    }

    async fn send_verified_signal(&self, record: &CancellationRecord, force: bool) -> bool {
        if wrapper_process_state(record.hook_pid, record.hook_process_identity.as_deref())
            == WrapperProcessState::Exited
        {
            return true;
        }
        #[cfg(target_os = "linux")]
        if let Some(handle) = Self::verified_hook_handle(record) {
            let signal = if force {
                rustix::process::Signal::KILL
            } else {
                rustix::process::Signal::TERM
            };
            return rustix::process::pidfd_send_signal(&handle, signal).is_ok();
        }
        #[cfg(not(target_os = "linux"))]
        let _ = force;
        warn!(
            "Skipping unverified local signal for build {}",
            record.build_id
        );
        false
    }

    async fn execute_cancellation_stages(
        &self,
        record: &mut CancellationRecord,
        force: bool,
        deadline: Instant,
    ) {
        // Losing the hook does not prove that its remote process group exited.
        // Only builds with no reservation AND no remote identity may skip SSH,
        // or an abandoned reservation that never reported a record path. An
        // abandoned build WITH a path still asks the worker, where a record
        // that is gone counts as proof of death (see `try_remote_kill`).
        let remote_required =
            record.remote_pgid_file.is_some() || (!record.abandoned_unlaunched && record.slots > 0);
        if record.abandoned_unlaunched {
            warn!(
                build_id = record.build_id,
                worker_id = %record.worker_id,
                slots = record.slots,
                "Releasing abandoned reservation that never launched remote work"
            );
            self.events.emit(
                "cancellation_abandoned_reservation_released",
                &serde_json::json!({
                    "build_id": record.build_id,
                    "worker_id": record.worker_id,
                    "slots": record.slots,
                    "hook_pid": record.hook_pid,
                }),
            );
        }

        // Force skips grace, not confirmation or the overall termination budget.
        if force {
            record.state = CancellationState::Escalated;
            if record.hook_pid > 0 {
                self.send_verified_signal(record, true).await;
            }
            let remote_stopped = !remote_required || self.try_remote_kill(record).await;
            let local_stopped = wait_for_wrapper_exit(record, self.config.kill_timeout).await;
            record.state = cancellation_terminal_state(local_stopped, remote_stopped);
            return;
        }

        // Step 1: Send SIGTERM.
        record.state = CancellationState::TermSent;
        if record.hook_pid > 0 {
            self.send_verified_signal(record, false).await;
        }

        let local_stopped = wait_for_wrapper_exit(record, self.config.grace_period).await;
        if local_stopped {
            let remote_stopped = !remote_required || self.attempt_remote_kill_stage(record).await;
            record.state = cancellation_terminal_state(true, remote_stopped);
            return;
        }

        if Instant::now() >= deadline {
            record.state = CancellationState::Failed;
            return;
        }

        // Step 2: Terminate remote work, preserving its result through SIGKILL.
        let remote_stopped = !remote_required || self.attempt_remote_kill_stage(record).await;
        if remote_stopped && wait_for_wrapper_exit(record, Duration::from_millis(500)).await {
            record.state = CancellationState::Completed;
            return;
        }

        if Instant::now() >= deadline || record.escalation_count >= self.config.max_escalations {
            record.state = CancellationState::Failed;
            return;
        }

        // Step 3: Escalate locally, but never turn a failed SSH kill into success.
        record.escalation_count += 1;
        self.events.emit(
            "cancellation_escalated",
            &serde_json::json!({
                "build_id": record.build_id,
                "worker_id": record.worker_id,
                "stage": "sigkill",
                "escalation_count": record.escalation_count,
            }),
        );

        record.state = CancellationState::Escalated;
        if record.hook_pid > 0 {
            self.send_verified_signal(record, true).await;
        }
        let local_stopped = wait_for_wrapper_exit(record, self.config.kill_timeout).await;
        record.state = cancellation_terminal_state(local_stopped, remote_stopped);
    }

    async fn attempt_remote_kill_stage(&self, record: &mut CancellationRecord) -> bool {
        if record.escalation_count >= self.config.max_escalations {
            return false;
        }
        record.escalation_count += 1;
        self.events.emit(
            "cancellation_escalated",
            &serde_json::json!({
                "build_id": record.build_id,
                "worker_id": record.worker_id,
                "stage": "remote_kill",
                "escalation_count": record.escalation_count,
            }),
        );

        record.state = CancellationState::RemoteKillSent;
        self.try_remote_kill(record).await
    }

    /// Attempt to kill the remote process on the worker via SSH.
    async fn try_remote_kill(&self, record: &mut CancellationRecord) -> bool {
        record.remote_kill_attempted = true;

        let Some(mut ssh) = build_remote_kill_command(record) else {
            warn!(
                build_id = record.build_id,
                worker_id = %record.worker_id,
                "Remote cancellation has no admitted endpoint; reservation retained"
            );
            return false;
        };
        let ssh_result = tokio::time::timeout(self.config.remote_kill_timeout, ssh.output()).await;

        match ssh_result {
            Ok(Ok(output)) => {
                let success = remote_kill_confirmed(&output, record.build_id)
                    || (record.abandoned_unlaunched
                        && remote_record_absent(&output, record.build_id));
                debug!(
                    "Remote kill for build {} on {}: success={}",
                    record.build_id, record.worker_id, success
                );
                success
            }
            Ok(Err(e)) => {
                warn!(
                    "Remote kill SSH command failed for build {}: {}",
                    record.build_id, e
                );
                false
            }
            Err(_) => {
                warn!(
                    "Remote kill timed out for build {} on {}",
                    record.build_id, record.worker_id
                );
                false
            }
        }
    }

    async fn capture_worker_health_snapshot(
        &self,
        ctx: &DaemonContext,
        worker_id: &str,
    ) -> Option<CancellationWorkerHealthSnapshot> {
        let worker = ctx.pool.get(&WorkerId::new(worker_id)).await?;
        let status = worker.status().await;
        let pressure = worker.pressure_assessment().await;

        Some(CancellationWorkerHealthSnapshot {
            status: worker_status_label(status).to_string(),
            speed_score: worker.get_speed_score(),
            used_slots: worker.used_slots(),
            available_slots: worker.available_slots().await,
            pressure_state: pressure.state.to_string(),
            pressure_reason_code: pressure.reason_code,
        })
    }

    /// Finalize confirmed cancellations; report failures without losing ownership.
    /// Failed attempts must leave the active build available to cleanup/retry.
    async fn run_cleanup(&self, ctx: &DaemonContext, record: &mut CancellationRecord) {
        let worker_id = &record.worker_id;

        // A failed attempt is not a terminal build. Taking it out of history
        // would both release unconfirmed capacity and prevent a later retry.
        let claimed_active = if record.state == CancellationState::Completed {
            ctx.history.active_build(record.build_id)
        } else {
            record.state = CancellationState::Failed;
            record.cleanup_ok = false;
            record.slots_released = 0;
            None
        };
        let mut history_ok = false;

        // 3. Build cancellation metadata and write finalized cancelled record.
        let elapsed = record.requested_at.elapsed();
        record.completed_at = Some(Instant::now());

        let decision_path = cancellation_decision_path(record);
        let escalation_stage = cancellation_escalation_stage(record);
        let operation_id = cancellation_operation_id(record.build_id);
        let cancel_origin = record.reason.to_string();
        let worker_health = self.capture_worker_health_snapshot(ctx, worker_id).await;

        if let Some(state) = claimed_active {
            let cancellation = BuildCancellationMetadata {
                operation_id: operation_id.clone(),
                origin: cancel_origin.clone(),
                reason_code: record.reason.to_string(),
                decision_path: decision_path
                    .iter()
                    .map(|stage| (*stage).to_string())
                    .collect(),
                escalation_stage: escalation_stage.to_string(),
                escalation_count: record.escalation_count,
                remote_kill_attempted: record.remote_kill_attempted,
                cleanup_ok: record.cleanup_ok,
                history_cancelled: true,
                final_state: record.state.to_string(),
                worker_health: worker_health.as_ref().map(worker_health_for_history),
            };
            match ctx.history.complete_durable(
                state.id,
                &state.worker_id,
                state.local_wrapper_id.as_deref(),
                crate::history::BuildCompletion {
                    exit_code: 130,
                    duration_ms: None,
                    bytes_transferred: None,
                    timing: None,
                    cancellation: Some(cancellation),
                },
            ) {
                Ok(Some((state, _))) => {
                    history_ok = true;
                    ctx.pool
                        .release_slots(&WorkerId::new(&state.worker_id), state.slots)
                        .await;
                    record.slots_released = state.slots;
                    if let Some(endpoint) = state.worker_endpoint.as_ref()
                        && let Some(worker) = ctx.pool.get(&endpoint.config.id).await
                    {
                        if matches!(
                            record.reason,
                            CancelReason::Timeout | CancelReason::StuckDetector
                        ) && let Some(_endpoint_guard) =
                            worker.lock_current_endpoint(endpoint).await
                        {
                            worker
                                .record_failure(Some(format!(
                                    "build cancelled by {}",
                                    record.reason
                                )))
                                .await;
                        }
                        // A cancelled build's partial cache belongs to the
                        // endpoint that ran it, including across an A -> B -> A
                        // retarget. The selector takes its cache lock first.
                        ctx.worker_selector
                            .record_bound_remote_completion(
                                &worker,
                                endpoint,
                                &state.project_id,
                                &state.command,
                                130,
                                state.remote_command_started(),
                            )
                            .await;
                    }
                }
                Ok(None) => {}
                Err(error) => {
                    warn!(
                        "Unable to persist cancellation for {}: {error}",
                        record.build_id
                    );
                    record.state = CancellationState::Failed;
                    record.cleanup_ok = false;
                }
            }
            if !cfg!(test) {
                crate::metrics::dec_active_builds("remote");
                crate::metrics::inc_build_total("cancelled", "remote");
            }
        }

        // 4. Emit completion or failure event.
        let event_name = match record.state {
            CancellationState::Completed => "cancellation_completed",
            _ => "cancellation_failed",
        };

        self.events.emit(
            event_name,
            &serde_json::json!({
                "operation_id": operation_id,
                "build_id": record.build_id,
                "worker_id": record.worker_id,
                "reason": record.reason,
                "cancel_origin": cancel_origin,
                "state": record.state,
                "decision_path": decision_path,
                "escalation_stage": escalation_stage,
                "escalation_count": record.escalation_count,
                "remote_kill_attempted": record.remote_kill_attempted,
                "slots_released": record.slots_released,
                "elapsed_ms": duration_millis_u64(elapsed),
                "cleanup_ok": record.cleanup_ok,
                "history_cancelled": history_ok,
                "active_build_retained": ctx.history.active_build(record.build_id).is_some(),
                "worker_health": worker_health,
            }),
        );

        if !history_ok && record.state == CancellationState::Completed {
            // Build was already gone from active — not necessarily an error
            // if another codepath cleaned it up.
            debug!(
                "Build {} was not in active history during cleanup (may have already completed)",
                record.build_id
            );
        }
    }

    /// Record cancellation stats for a worker (for debt computation).
    async fn record_cancellation_stats(&self, record: &CancellationRecord) {
        let mut stats = self.worker_stats.write().await;
        let entry = stats.entry(record.worker_id.clone()).or_default();

        let now = Instant::now();
        entry.recent_cancellations.push(now);

        for _ in 0..record.escalation_count {
            entry.recent_escalations.push(now);
        }
        if !record.cleanup_ok {
            entry.recent_cleanup_failures.push(now);
        }
    }

    /// Compute cancellation debt for a worker (0.0 = clean, 1.0 = saturated).
    ///
    /// Used by the reliability model as a 5th signal.
    pub async fn cancellation_debt(&self, worker_id: &str) -> f64 {
        let mut stats = self.worker_stats.write().await;
        let Some(entry) = stats.get_mut(worker_id) else {
            return 0.0; // No cancellation history → no debt.
        };

        // Prune all counters outside the window. Compare ages rather than
        // constructing a cutoff Instant, because `Instant::now() - Duration`
        // panics if the result would precede the monotonic clock's origin
        // (fresh-boot scenario on Linux where `Instant::now()` < DEBT_WINDOW).
        let now = Instant::now();
        entry
            .recent_cancellations
            .retain(|t| now.saturating_duration_since(*t) < DEBT_WINDOW);
        entry
            .recent_escalations
            .retain(|t| now.saturating_duration_since(*t) < DEBT_WINDOW);
        entry
            .recent_cleanup_failures
            .retain(|t| now.saturating_duration_since(*t) < DEBT_WINDOW);

        let recent_count = entry.recent_cancellations.len() as f64;

        // Rate component: cancellations per 5-minute window, normalized.
        // 5+ cancellations in 5 minutes → full rate debt.
        let rate_debt = (recent_count / 5.0).min(1.0);

        // Escalation component: 0.2 per recent escalation.
        let escalation_debt = (entry.recent_escalations.len() as f64 * 0.2).min(0.6);

        // Cleanup failure component: 0.3 per recent failed cleanup.
        let cleanup_debt = (entry.recent_cleanup_failures.len() as f64 * 0.3).min(0.6);

        // Weighted combination, capped at 1.0.
        (rate_debt * 0.4 + escalation_debt * 0.3 + cleanup_debt * 0.3).clamp(0.0, 1.0)
    }

    /// Get active (in-flight) cancellation records.
    pub async fn active_cancellations(&self) -> Vec<CancellationRecord> {
        self.active
            .lock()
            .unwrap_or_else(|error| error.into_inner())
            .values()
            .cloned()
            .collect()
    }

    /// Increment the total builds counter for a worker (for rate computation).
    pub async fn record_build(&self, worker_id: &str) {
        let mut stats = self.worker_stats.write().await;
        let entry = stats.entry(worker_id.to_string()).or_default();
        entry.total_builds += 1;
    }
}

// ── Process signal helpers ───────────────────────────────────────────────

// Keep the executable protocol as a shell source so tests run the exact bytes
// sent to workers, including the process-table verification after signalling.
const REMOTE_CANCELLATION_SCRIPT: &str = include_str!("cancellation_remote.sh");

/// Build a cancellation transport from immutable admission evidence. Looking
/// up today's worker ID here can send an old build's signal to a replacement,
/// so [`legacy_probe_config`] coordinates only ask whether the record is gone
/// and never signal.
fn build_remote_kill_command(record: &CancellationRecord) -> Option<tokio::process::Command> {
    let (config, admitted) = match record.worker_endpoint.as_ref() {
        Some(endpoint) => (&endpoint.config, true),
        None if record.abandoned_unlaunched => (record.legacy_probe_config.as_ref()?, false),
        None => return None,
    };
    if config.id.as_str() != record.worker_id {
        return None;
    }
    let mut ssh = tokio::process::Command::new("ssh");
    ssh.args([
        "-o",
        "StrictHostKeyChecking=no",
        "-o",
        "ConnectTimeout=5",
        "-o",
        "BatchMode=yes",
        "-i",
        &config.identity_file,
    ]);
    if let Some(options) = rch_common::ssh::identities_only_args(&config.identity_file) {
        ssh.args(options);
    }
    let script = if admitted {
        build_remote_kill_script(record.remote_pgid_file.as_deref(), record.build_id)
    } else {
        build_remote_record_probe_script(record.remote_pgid_file.as_deref(), record.build_id)
    };
    ssh.arg(format!("{}@{}", config.user, config.host))
        .arg(script)
        // The timeout owns this local SSH process even if the caller exits.
        .kill_on_drop(true);
    Some(ssh)
}

/// The absence check of `cancellation_remote.sh` without its signalling: a
/// record still present behind unadmitted coordinates keeps the reservation.
const REMOTE_RECORD_PROBE_SCRIPT: &str = r#"if [ ! -e "$1" ] && [ ! -L "$1" ]; then
    printf 'RCH_REMOTE_RECORD_ABSENT_V1:%s\n' "$2"
    exit 42
fi
exit 43
"#;

fn build_remote_record_probe_script(remote_pgid_file: Option<&str>, build_id: u64) -> String {
    let Some(remote_pgid_file) = remote_pgid_file else {
        return "exit 42".to_owned();
    };
    format!(
        "sh -c {} sh {} {build_id}",
        shell_escape::escape(std::borrow::Cow::Borrowed(REMOTE_RECORD_PROBE_SCRIPT)),
        shell_escape::escape(std::borrow::Cow::Borrowed(remote_pgid_file)),
    )
}

fn build_remote_kill_script(remote_pgid_file: Option<&str>, build_id: u64) -> String {
    let Some(remote_pgid_file) = remote_pgid_file else {
        // Matching command text cannot prove that a build's descendants exited.
        // Retain the reservation when no process-group identity was recorded.
        return "exit 42".to_owned();
    };
    let protocol = format!(
        "{}\n{}",
        rch_common::REMOTE_PROCESS_IDENTITY_SCRIPT,
        REMOTE_CANCELLATION_SCRIPT
    );
    format!(
        "sh -c {} sh {} {build_id}",
        shell_escape::escape(std::borrow::Cow::Borrowed(protocol.as_str())),
        shell_escape::escape(std::borrow::Cow::Borrowed(remote_pgid_file)),
    )
}

/// The worker no longer holds the build's process record (see
/// `cancellation_remote.sh`). Proof of death only for an abandoned build.
fn remote_record_absent(output: &std::process::Output, build_id: u64) -> bool {
    output.status.code() == Some(42)
        && output.stdout.as_slice()
            == format!("RCH_REMOTE_RECORD_ABSENT_V1:{build_id}\n").as_bytes()
}

fn remote_kill_confirmed(output: &std::process::Output, build_id: u64) -> bool {
    // Neither SSH exit zero nor partial/stale output is a termination receipt.
    output.status.success()
        && output.stdout.as_slice() == format!("RCH_REMOTE_CANCELLED_V1:{build_id}\n").as_bytes()
}

/// Use the existing safe syscall wrapper: spawning /bin/kill can block the
/// runtime before an async deadline can be polled, especially under fork load.
#[cfg(test)]
fn send_signal_to_process(pid: u32, force: bool) -> bool {
    use nix::sys::signal::{Signal, kill};
    use nix::unistd::Pid;

    let Ok(pid) = i32::try_from(pid) else {
        return false;
    };
    // Never reinterpret an invalid hook PID as a process-group/broadcast signal.
    if pid <= 1 {
        return false;
    }
    let signal = if force {
        Signal::SIGKILL
    } else {
        Signal::SIGTERM
    };
    match kill(Pid::from_raw(pid), signal) {
        Ok(()) => true,
        Err(error) => {
            debug!("Failed to send {signal} to process {pid}: {error}");
            false
        }
    }
}

// Only `wait_for_process_exit` (itself `#[cfg(test)]`) calls this; ungated it
// was dead code in the daemon binary and failed `clippy -D warnings`.
#[cfg(test)]
fn is_process_alive(pid: u32) -> bool {
    use nix::errno::Errno;
    use nix::sys::signal::kill;
    use nix::unistd::Pid;

    if pid == 0 {
        return false;
    }
    let Ok(raw_pid) = i32::try_from(pid) else {
        return false;
    };
    #[cfg(target_os = "linux")]
    if let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat"))
        && let Some((_, fields)) = stat.rsplit_once(") ")
        && matches!(fields.split_whitespace().next(), Some("Z" | "X"))
    {
        // A zombie has already exited; its parent may not have reaped it yet.
        return false;
    }
    // Permission and other probe errors are not evidence of process absence.
    !matches!(kill(Pid::from_raw(raw_pid), None), Err(Errno::ESRCH))
}

async fn wait_for_wrapper_exit(record: &CancellationRecord, budget: Duration) -> bool {
    let observe =
        || wrapper_process_state(record.hook_pid, record.hook_process_identity.as_deref());
    match observe() {
        WrapperProcessState::Exited => return true,
        WrapperProcessState::Unverified => return false,
        WrapperProcessState::Running => {}
    }
    if budget.is_zero() {
        return false;
    }
    tokio::time::timeout(budget, async {
        loop {
            match observe() {
                WrapperProcessState::Exited => return true,
                WrapperProcessState::Unverified => return false,
                WrapperProcessState::Running => {
                    tokio::time::sleep(Duration::from_millis(25)).await;
                }
            }
        }
    })
    .await
    .unwrap_or(false)
}

#[cfg(test)]
async fn wait_for_process_exit(pid: u32, budget: Duration) -> bool {
    if !is_process_alive(pid) {
        return true;
    }
    if budget.is_zero() {
        return false;
    }
    tokio::time::timeout(budget, async {
        while is_process_alive(pid) {
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
    })
    .await
    .is_ok()
}

// ── Tests ────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::benchmark_queue::BenchmarkQueue;
    use crate::benchmark_scheduler::{BenchmarkScheduler, SchedulerConfig};
    use crate::events::EventBus;
    use crate::history::BuildHistory;
    use crate::selection::WorkerSelector;
    use crate::self_test::{
        DEFAULT_RESULT_CAPACITY, DEFAULT_RUN_CAPACITY, SelfTestHistory, SelfTestService,
    };
    use crate::workers::WorkerPool;
    use chrono::Duration as ChronoDuration;
    use rch_common::SelfTestConfig;
    use std::sync::Arc;
    use std::time::Instant;

    #[test]
    fn test_duration_millis_u64_saturates() {
        assert_eq!(duration_millis_u64(Duration::from_secs(u64::MAX)), u64::MAX);
    }

    pub(super) fn test_events() -> EventBus {
        EventBus::new(64)
    }

    pub(super) fn test_config() -> CancellationConfig {
        CancellationConfig {
            grace_period: Duration::from_millis(100),
            kill_timeout: Duration::from_millis(50),
            remote_kill_timeout: Duration::from_secs(1),
            max_escalations: 3,
            cleanup_timeout: Duration::from_secs(5),
        }
    }

    fn make_test_self_test(pool: WorkerPool) -> Arc<SelfTestService> {
        let history = Arc::new(SelfTestHistory::new(
            DEFAULT_RUN_CAPACITY,
            DEFAULT_RESULT_CAPACITY,
        ));
        Arc::new(SelfTestService::new(
            pool,
            SelfTestConfig::default(),
            history,
        ))
    }

    fn make_test_alert_manager() -> Arc<crate::alerts::AlertManager> {
        Arc::new(crate::alerts::AlertManager::new(
            crate::alerts::AlertConfig::default(),
        ))
    }

    fn make_test_benchmark_trigger(
        pool: WorkerPool,
    ) -> crate::benchmark_scheduler::BenchmarkTriggerHandle {
        let telemetry = Arc::new(crate::telemetry::TelemetryStore::new(
            Duration::from_secs(300),
            None,
        ));
        let (scheduler, trigger) =
            BenchmarkScheduler::new(SchedulerConfig::default(), pool, telemetry, test_events());
        let scheduler = Arc::new(scheduler);
        tokio::spawn(scheduler.run());
        trigger
    }

    pub(super) fn make_test_context(pool: WorkerPool, history: Arc<BuildHistory>) -> DaemonContext {
        let events = test_events();
        DaemonContext {
            pool: pool.clone(),
            worker_selector: Arc::new(WorkerSelector::new()),
            history,
            telemetry: Arc::new(crate::telemetry::TelemetryStore::new(
                Duration::from_secs(300),
                None,
            )),
            benchmark_queue: Arc::new(BenchmarkQueue::new(ChronoDuration::minutes(5))),
            benchmark_trigger: make_test_benchmark_trigger(pool.clone()),
            events: events.clone(),
            self_test: make_test_self_test(pool.clone()),
            alert_manager: make_test_alert_manager(),
            repo_convergence: Arc::new(crate::repo_convergence::RepoConvergenceService::new(
                events.clone(),
            )),
            cancellation: Arc::new(CancellationOrchestrator::new(
                CancellationConfig::default(),
                events.clone(),
            )),
            started_at: Instant::now(),
            socket_path: "/tmp/test-cancel.sock".to_string(),
            version: "0.0.0-test",
            pid: std::process::id(),
            queue_timeout_secs: 300,
            bypass_store: None,
            admin_disable_store: None,
            workers_config_path: None,
            admission_barrier: Arc::new(tokio::sync::RwLock::new(false)),
        }
    }

    fn test_record(
        state: CancellationState,
        escalation_count: u32,
        remote_kill_attempted: bool,
    ) -> CancellationRecord {
        CancellationRecord {
            build_id: 42,
            worker_id: "w1".to_string(),
            worker_endpoint: None,
            state,
            reason: CancelReason::User,
            requested_at: Instant::now(),
            completed_at: None,
            escalation_count,
            remote_kill_attempted,
            cleanup_ok: true,
            slots: 1,
            slots_released: 1,
            hook_pid: 12345,
            hook_process_identity: None,
            remote_pgid_file: None,
            abandoned_unlaunched: false,
            legacy_probe_config: None,
        }
    }

    #[test]
    fn test_build_remote_kill_script_prefers_recorded_pgid_file() {
        let script = build_remote_kill_script(Some("/tmp/rch/project/.rch-run/42.pgid"), 42);
        assert!(script.contains("rch_remote_cancel"));
        assert!(script.contains("RCH_REMOTE_PROCESS_V1"));
        // Group-kill must use `-PGID` (no `--`): dash's kill builtin mishandles
        // `kill -TERM -- -PGID`, silently failing to signal the group.
        assert!(script.contains("kill -\"$1\" -\"$rch_pgid\""));
        assert!(!script.contains("kill -TERM -- -"));
        assert!(script.contains("/tmp/rch/project/.rch-run/42.pgid"));
        assert!(!script.contains("RCH_BUILD_ID=42;"));
    }

    #[test]
    fn test_build_remote_kill_script_handles_shell_special_pgid_file_path() {
        let script =
            build_remote_kill_script(Some("/tmp/rch/project dir/agent's/.rch-run/42.pgid"), 42);
        assert!(script.contains("rch_remote_cancel \"$1\" \"$2\" term"));
        assert!(script.contains("/tmp/rch/project"));
        assert!(!script.contains("pgid_file='/tmp"));

        let status = std::process::Command::new("sh")
            .arg("-n")
            .arg("-c")
            .arg(&script)
            .status()
            .expect("shell syntax check should run");
        assert!(
            status.success(),
            "remote kill script must be shell-parseable: {script}"
        );
    }

    // 1. Cancel of non-existent build → error response
    #[tokio::test]
    async fn test_cancel_nonexistent_build_returns_error() {
        let pool = WorkerPool::new();
        let history = Arc::new(BuildHistory::new(100));
        let ctx = make_test_context(pool, history);
        let orch = CancellationOrchestrator::new(test_config(), test_events());

        let resp = orch
            .cancel_build(&ctx, 999, CancelReason::User, false)
            .await;
        assert_eq!(resp.status, "error");
        assert_eq!(resp.slots_released, 0);
    }

    #[test]
    fn only_a_silent_unlaunched_reservation_with_a_gone_wrapper_is_abandoned() {
        let history = BuildHistory::new(10);
        let mut build = history.start_active_build(
            "proj".to_string(),
            "worker-a".to_string(),
            "cargo build".to_string(),
            0,
            2,
            rch_common::BuildLocation::Remote,
        );
        let now = Instant::now() + ABANDONED_UNLAUNCHED_AFTER + Duration::from_secs(60);
        let no_lease = |_: &str| false;
        assert!(
            provably_abandoned_unlaunched(&build, now, no_lease),
            "pid 0, silent, never launched"
        );

        let mut launched = build.clone();
        launched.remote_pgid_file = Some("/tmp/rch-run/p/1.pgid".to_owned());
        // The hook reports the record path before launch; the worker decides.
        assert!(
            provably_abandoned_unlaunched(&launched, now, no_lease),
            "a reported record path alone does not disqualify"
        );

        assert!(
            !provably_abandoned_unlaunched(&build, Instant::now(), no_lease),
            "not silent long enough"
        );

        let mut live_hook = build.clone();
        live_hook.hook_pid = std::process::id();
        assert!(
            !provably_abandoned_unlaunched(&live_hook, now, no_lease),
            "a live (unverifiable) wrapper keeps it"
        );

        build.local_wrapper_id = Some("rchw-owner".to_owned());
        assert!(
            !provably_abandoned_unlaunched(&build, now, |wrapper| wrapper == "rchw-owner"),
            "a lease still owning worker source keeps it"
        );
        assert!(provably_abandoned_unlaunched(&build, now, no_lease));
    }

    /// bd-ilds6: a build recovered from a pre-endpoint ownership snapshot has
    /// no admitted coordinates, so the stuck detector retried it every tick,
    /// forever. Once abandoned, its worker ID's current endpoint may prove
    /// that the record is gone, but it is never sent the signalling protocol.
    #[tokio::test]
    async fn abandoned_legacy_build_probes_current_endpoint_for_absence_only() {
        let pool = WorkerPool::new();
        let current = rch_common::WorkerConfig {
            id: WorkerId::new("w1"),
            host: "current.example".into(),
            user: "current-user".into(),
            identity_file: "/current/key".into(),
            ..rch_common::WorkerConfig::default()
        };
        pool.add_worker(current.clone()).await;
        let ctx = make_test_context(pool, Arc::new(BuildHistory::new(10)));
        let mut record = test_record(CancellationState::Requested, 0, false);
        record.remote_pgid_file = Some("/tmp/rch-run/p/42.pgid".into());

        // Not abandoned: unknown coordinates stay unknown.
        assert!(legacy_probe_config(&ctx, &record).await.is_none());
        record.legacy_probe_config = Some(current);
        assert!(build_remote_kill_command(&record).is_none());

        record.abandoned_unlaunched = true;
        record.legacy_probe_config = legacy_probe_config(&ctx, &record).await;
        let argv: Vec<_> = build_remote_kill_command(&record)
            .unwrap()
            .as_std()
            .get_args()
            .map(|arg| arg.to_string_lossy().into_owned())
            .collect();
        assert!(argv.windows(2).any(|args| args == ["-i", "/current/key"]));
        assert_eq!(argv[argv.len() - 2], "current-user@current.example");
        assert_eq!(
            argv.last().unwrap(),
            &build_remote_record_probe_script(Some("/tmp/rch-run/p/42.pgid"), 42)
        );
        assert!(!argv.last().unwrap().contains("rch_remote_cancel"));

        // An ID no longer in inventory has nothing to ask, and coordinates
        // resolved for another ID are never used.
        record.worker_id = "gone".into();
        assert!(legacy_probe_config(&ctx, &record).await.is_none());
        assert!(build_remote_kill_command(&record).is_none());
    }

    #[tokio::test]
    async fn legacy_record_probe_proves_absence_and_never_signals() {
        let root = tempfile::tempdir().unwrap();
        let path = root.path().join("group with 'quote $dollar;`tick`.pgid");
        let probe = |path: std::path::PathBuf| async move {
            // The worker's login shell runs the probe exactly as SSH passes it.
            tokio::process::Command::new("/bin/sh")
                .arg("-c")
                .arg(build_remote_record_probe_script(path.to_str(), 42))
                .stdin(std::process::Stdio::null())
                .output()
                .await
                .unwrap()
        };
        let absent = probe(path.clone()).await;
        assert!(remote_record_absent(&absent, 42), "{absent:?}");
        assert!(!remote_record_absent(&absent, 41));

        std::fs::write(&path, std::process::id().to_string()).unwrap();
        let present = probe(path.clone()).await;
        assert_eq!(present.status.code(), Some(43), "{present:?}");
        assert!(present.stdout.is_empty());
        assert!(!remote_record_absent(&present, 42));
        assert!(!remote_kill_confirmed(&present, 42));
        assert_eq!(build_remote_record_probe_script(None, 42), "exit 42");
    }

    #[test]
    fn process_identities_from_both_platforms_are_well_formed() {
        let boot = "3f1c2a9e-5b7d-4e2a-9c1f-0a1b2c3d4e5f";
        assert!(well_formed_process_identity(&format!("{boot}:123456789")));
        assert!(
            well_formed_process_identity(&format!("{boot}:darwin:1791280000:123456")),
            "macOS kernel birth retains microseconds"
        );
        assert!(!well_formed_process_identity(&format!(
            "{boot}:Mon Sep 28 10:48:17 2026"
        )));
        assert!(!well_formed_process_identity(&format!(
            "{boot}:darwin:1791280000:1000000"
        )));
        assert!(!well_formed_process_identity("not-a-uuid:123"));
        assert!(!well_formed_process_identity(&format!("{boot}:")));
        assert!(!well_formed_process_identity(&format!("{boot}:   ")));
        assert!(!well_formed_process_identity(boot));
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    #[test]
    fn native_absence_releases_legacy_owners_but_a_live_legacy_pid_is_unverified() {
        let mut child = std::process::Command::new("/bin/sleep")
            .arg("30")
            .spawn()
            .unwrap();
        let pid = child.id();
        // Reap before assertions so an unsupported test environment cannot leak
        // the fixture. Both observations still exercise production inspection.
        let current = crate::history::process_identity(pid);
        let legacy = "3f1c2a9e-5b7d-4e2a-9c1f-0a1b2c3d4e5f:Mon Sep 28 10:48:17 2026";
        let live_unknown = wrapper_process_state(pid, None);
        let live_legacy = wrapper_process_state(pid, Some(legacy));
        child.kill().unwrap();
        child.wait().unwrap();
        assert!(
            current.is_some(),
            "native birth requires a coherent process inspection namespace"
        );
        assert_eq!(live_unknown, WrapperProcessState::Unverified);
        assert_eq!(live_legacy, WrapperProcessState::Unverified);
        assert_eq!(
            wrapper_process_state(pid, None),
            WrapperProcessState::Exited
        );
        assert_eq!(
            wrapper_process_state(pid, Some(legacy)),
            WrapperProcessState::Exited
        );
    }

    /// bd-axhoi: a wrapper that died in sync_up never started anything remote,
    /// so its reservation is abandoned after the short bound, not after 6h.
    /// Once the remote command has started, the long bound still applies.
    #[test]
    fn a_build_that_never_left_sync_up_is_abandoned_after_the_short_bound() {
        let history = BuildHistory::new(10);
        let build = history.start_active_build(
            "proj".to_string(),
            "worker-a".to_string(),
            "cargo test".to_string(),
            0,
            4,
            rch_common::BuildLocation::Remote,
        );
        assert_eq!(
            build.heartbeat_phase,
            rch_common::BuildHeartbeatPhase::SyncUp
        );
        let no_lease = |_: &str| false;
        let past_short = Instant::now() + ABANDONED_BEFORE_REMOTE_AFTER + Duration::from_secs(60);
        assert!(
            provably_abandoned_unlaunched(&build, past_short, no_lease),
            "dead in sync_up: released after the short bound"
        );
        assert!(
            !provably_abandoned_unlaunched(&build, Instant::now(), no_lease),
            "not silent long enough yet"
        );

        let mut executing = build.clone();
        executing.heartbeat_phase = rch_common::BuildHeartbeatPhase::Execute;
        assert!(
            !provably_abandoned_unlaunched(&executing, past_short, no_lease),
            "a started remote command keeps the long bound"
        );
        let past_long = Instant::now() + ABANDONED_UNLAUNCHED_AFTER + Duration::from_secs(60);
        assert!(provably_abandoned_unlaunched(
            &executing, past_long, no_lease
        ));
    }

    #[tokio::test]
    async fn test_cancel_inflight_build_records_metadata() {
        let pool = WorkerPool::new();
        let history = Arc::new(BuildHistory::new(100));
        let active = history.start_active_build(
            "proj".to_string(),
            "worker-a".to_string(),
            "cargo test".to_string(),
            0,
            0,
            rch_common::BuildLocation::Remote,
        );
        let ctx = make_test_context(pool, history.clone());
        let orch = CancellationOrchestrator::new(test_config(), test_events());

        let resp = orch
            .cancel_build(&ctx, active.id, CancelReason::Timeout, false)
            .await;
        assert_eq!(resp.status, "cancelled");
        assert_eq!(resp.build_id, active.id);
        assert!(history.active_build(active.id).is_none());

        let recent = history.recent(5);
        let cancelled = recent
            .iter()
            .find(|record| record.id == active.id)
            .expect("cancelled build record should exist");
        let metadata = cancelled
            .cancellation
            .as_ref()
            .expect("cancellation metadata should be present");
        assert_eq!(metadata.origin, "timeout");
        assert_eq!(metadata.reason_code, "timeout");
        assert_eq!(metadata.operation_id, format!("cancel-{}", active.id));
        assert_eq!(metadata.final_state, "completed");
        assert!(metadata.history_cancelled);
    }

    #[tokio::test]
    async fn test_cancel_after_remote_execution_started_warms_cache_without_pin() {
        // GH #81: an interrupted build leaves its partial pool on the worker.
        let pool = WorkerPool::new();
        pool.add_worker(rch_common::WorkerConfig {
            id: WorkerId::new("worker-a"),
            ..Default::default()
        })
        .await;
        let worker = pool.get(&WorkerId::new("worker-a")).await.unwrap();
        let history = Arc::new(BuildHistory::new(100));
        let started = history
            .try_start_active_build_with_waiter(
                "proj".to_string(),
                "worker-a".to_string(),
                "cargo test".to_string(),
                0,
                None,
                0,
                rch_common::BuildLocation::Remote,
                None,
                crate::disk_pressure::DiskHeadroomAdmission::default(),
                Some(worker.endpoint_snapshot().await),
            )
            .unwrap()
            .unwrap();
        history
            .record_build_heartbeat(rch_common::BuildHeartbeatRequest {
                build_id: started.id,
                worker_id: WorkerId::new("worker-a"),
                hook_pid: None,
                local_wrapper_id: None,
                remote_pgid_file: None,
                phase: rch_common::BuildHeartbeatPhase::Execute,
                detail: None,
                progress_counter: Some(1),
                progress_percent: None,
            })
            .expect("heartbeat accepted");
        let unsynced = history
            .try_start_active_build_with_waiter(
                "unsynced".to_string(),
                "worker-a".to_string(),
                "cargo test".to_string(),
                0,
                None,
                0,
                rch_common::BuildLocation::Remote,
                None,
                crate::disk_pressure::DiskHeadroomAdmission::default(),
                Some(worker.endpoint_snapshot().await),
            )
            .unwrap()
            .unwrap();
        let ctx = make_test_context(pool, history.clone());
        let orch = CancellationOrchestrator::new(test_config(), test_events());

        for id in [started.id, unsynced.id] {
            let resp = orch
                .cancel_build(&ctx, id, CancelReason::StuckDetector, false)
                .await;
            assert_eq!(resp.status, "cancelled");
        }
        let warmth = |project: &'static str| {
            let selector = ctx.worker_selector.clone();
            async move {
                selector
                    .cache_warmth("worker-a", project, crate::selection::CacheUse::Test)
                    .await
            }
        };
        assert_eq!(warmth("proj").await, 1.0);
        assert_eq!(warmth("unsynced").await, 0.0);
        assert_eq!(ctx.worker_selector.get_pinned_worker("proj").await, None);
    }

    #[tokio::test]
    async fn cancelled_build_feedback_and_transport_follow_durable_endpoint_ownership() {
        for change in [
            "unchanged",
            "retarget",
            "aba",
            "restart",
            "removed",
            "legacy",
        ] {
            let root = tempfile::tempdir().unwrap();
            let path = root.path().join("history.jsonl");
            let original = rch_common::WorkerConfig {
                id: WorkerId::new("owned-cancellation"),
                host: "admitted.example".into(),
                user: "admitted-user".into(),
                identity_file: "/admitted/key with spaces".into(),
                total_slots: 8,
                ..rch_common::WorkerConfig::default()
            };
            let pool = WorkerPool::new();
            pool.add_worker(original.clone()).await;
            let mut worker = pool.get(&original.id).await.unwrap();
            let history = Arc::new(BuildHistory::new(100).with_persistence(path.clone()));
            let active = history
                .try_start_active_build_with_waiter(
                    "cancelled-project".into(),
                    original.id.to_string(),
                    "cargo test".into(),
                    0,
                    None,
                    2,
                    rch_common::BuildLocation::Remote,
                    None,
                    crate::disk_pressure::DiskHeadroomAdmission::default(),
                    if change == "legacy" {
                        None
                    } else {
                        Some(worker.endpoint_snapshot().await)
                    },
                )
                .unwrap()
                .unwrap();
            history
                .record_build_heartbeat(rch_common::BuildHeartbeatRequest {
                    build_id: active.id,
                    worker_id: original.id.clone(),
                    hook_pid: None,
                    local_wrapper_id: None,
                    remote_pgid_file: Some("/admitted/owned group.pgid".into()),
                    phase: rch_common::BuildHeartbeatPhase::Execute,
                    detail: None,
                    progress_counter: Some(1),
                    progress_percent: None,
                })
                .unwrap();
            let other = history.start_active_build(
                "other-project".into(),
                original.id.to_string(),
                "cargo check".into(),
                0,
                1,
                rch_common::BuildLocation::Remote,
            );
            assert!(worker.reserve_slots(3).await);
            let mut ctx = make_test_context(pool, history);
            if matches!(change, "retarget" | "aba") {
                ctx.pool
                    .add_worker(rch_common::WorkerConfig {
                        host: "replacement.example".into(),
                        user: "replacement-user".into(),
                        identity_file: "/replacement/key".into(),
                        ..original.clone()
                    })
                    .await;
                if change == "aba" {
                    ctx.pool.add_worker(original.clone()).await;
                }
            } else if matches!(change, "restart" | "removed") {
                ctx.history = Arc::new(BuildHistory::load_from_file(&path, 100).unwrap());
                ctx.pool = WorkerPool::new();
                if change == "restart" {
                    ctx.pool.add_worker(original.clone()).await;
                    worker = ctx.pool.get(&original.id).await.unwrap();
                }
                for active in ctx.history.active_builds() {
                    ctx.pool
                        .restore_recovered_slots(&original.id, active.slots)
                        .await
                        .unwrap();
                }
            }
            if change != "removed" {
                worker
                    .record_failure(Some("current endpoint evidence".into()))
                    .await;
            }
            let owned = ctx.history.active_build(active.id).unwrap();
            let mut record = test_record(CancellationState::Completed, 0, true);
            record.build_id = owned.id;
            record.worker_id = owned.worker_id;
            record.worker_endpoint = owned.worker_endpoint;
            record.remote_pgid_file = owned.remote_pgid_file;
            record.reason = CancelReason::Timeout;
            record.slots = 999;
            record.slots_released = 0;
            if change == "legacy" {
                assert!(build_remote_kill_command(&record).is_none());
            } else {
                let command = build_remote_kill_command(&record).unwrap();
                let argv: Vec<_> = command
                    .as_std()
                    .get_args()
                    .map(|arg| arg.to_string_lossy().into_owned())
                    .collect();
                assert!(
                    argv.windows(2)
                        .any(|args| args == ["-i", "/admitted/key with spaces"])
                );
                assert_eq!(
                    argv[argv.len() - 2],
                    "admitted-user@admitted.example",
                    "{change}"
                );
                assert_eq!(
                    argv.last().unwrap(),
                    &build_remote_kill_script(Some("/admitted/owned group.pgid"), active.id)
                );
                record.worker_id = "different-owner".into();
                assert!(build_remote_kill_command(&record).is_none());
                record.worker_id = original.id.to_string();
            }
            // Exercise the post-confirmation boundary; transport receipt and
            // process exit are covered by the real SSH fixtures separately.
            let orch = CancellationOrchestrator::new(test_config(), test_events());
            orch.run_cleanup(&ctx, &mut record).await;
            assert_eq!(
                record.slots_released, 2,
                "{change}: use exact durable owner slots"
            );
            if change == "removed" {
                ctx.pool.add_worker(original.clone()).await;
                worker = ctx.pool.get(&original.id).await.unwrap();
            }
            assert_eq!(worker.used_slots(), 1, "{change}");
            let failures = worker.circuit_stats().await.consecutive_failures();
            assert_eq!(
                failures,
                match change {
                    "unchanged" => 2,
                    "removed" => 0,
                    _ => 1,
                },
                "{change}"
            );
            assert_eq!(
                ctx.worker_selector
                    .cache_warmth(
                        original.id.as_str(),
                        "cancelled-project",
                        crate::selection::CacheUse::Test
                    )
                    .await,
                if change == "unchanged" { 1.0 } else { 0.0 },
                "{change}"
            );
            assert_eq!(
                ctx.worker_selector
                    .get_pinned_worker("cancelled-project")
                    .await,
                None
            );
            orch.run_cleanup(&ctx, &mut record).await;
            assert_eq!(
                worker.used_slots(),
                1,
                "{change}: duplicate cleanup released another build"
            );
            assert_eq!(
                worker.circuit_stats().await.consecutive_failures(),
                failures,
                "{change}: duplicate cleanup advanced circuit failure"
            );
            assert!(ctx.history.active_build(other.id).is_some());
            assert!(ctx.history.active_build(active.id).is_none());
            assert_eq!(ctx.history.recent(10).len(), 1);
        }
    }

    #[tokio::test]
    async fn test_cancel_after_completion_returns_error_post_completion_race() {
        let pool = WorkerPool::new();
        let history = Arc::new(BuildHistory::new(100));
        let active = history.start_active_build(
            "proj".to_string(),
            "worker-a".to_string(),
            "cargo check".to_string(),
            0,
            0,
            rch_common::BuildLocation::Remote,
        );
        let _ = history.finish_active_build(active.id, 0, None, None, None);

        let ctx = make_test_context(pool, history);
        let orch = CancellationOrchestrator::new(test_config(), test_events());
        let resp = orch
            .cancel_build(&ctx, active.id, CancelReason::User, false)
            .await;
        assert_eq!(resp.status, "error");
        assert!(
            resp.message
                .as_deref()
                .is_some_and(|message| message.contains("not found"))
        );
    }

    #[tokio::test]
    async fn test_repeated_cancel_after_completion_is_deterministic() {
        let pool = WorkerPool::new();
        let history = Arc::new(BuildHistory::new(100));
        let active = history.start_active_build(
            "proj".to_string(),
            "worker-a".to_string(),
            "cargo clippy".to_string(),
            0,
            0,
            rch_common::BuildLocation::Remote,
        );
        let ctx = make_test_context(pool, history);
        let orch = CancellationOrchestrator::new(test_config(), test_events());

        let first = orch
            .cancel_build(&ctx, active.id, CancelReason::User, false)
            .await;
        assert_eq!(first.status, "cancelled");

        let second = orch
            .cancel_build(&ctx, active.id, CancelReason::User, false)
            .await;
        assert_eq!(second.status, "error");
    }

    // 2. Double cancel (idempotent) — simulate by trying to cancel the same
    //    non-existent build twice (since we can't easily create active builds
    //    in unit tests without the full pipeline).
    #[tokio::test]
    async fn test_double_cancel_nonexistent_is_idempotent() {
        let pool = WorkerPool::new();
        let history = Arc::new(BuildHistory::new(100));
        let ctx = make_test_context(pool, history);
        let orch = CancellationOrchestrator::new(test_config(), test_events());

        let r1 = orch.cancel_build(&ctx, 42, CancelReason::User, false).await;
        let r2 = orch.cancel_build(&ctx, 42, CancelReason::User, false).await;
        assert_eq!(r1.status, "error");
        assert_eq!(r2.status, "error");
    }

    // 3. Cancellation debt computation: 0 cancellations = 0.0 debt
    #[tokio::test]
    async fn test_cancellation_debt_zero_for_unknown_worker() {
        let orch = CancellationOrchestrator::new(test_config(), test_events());
        let debt = orch.cancellation_debt("w1").await;
        assert!(debt < f64::EPSILON);
    }

    // 4. Cancellation debt increases with cancellation events
    #[tokio::test]
    async fn test_cancellation_debt_increases_with_events() {
        let orch = CancellationOrchestrator::new(test_config(), test_events());

        // Manually record some cancellation stats.
        {
            let mut stats = orch.worker_stats.write().await;
            let entry = stats.entry("w1".to_string()).or_default();
            let now = Instant::now();
            for _ in 0..5 {
                entry.recent_cancellations.push(now);
            }
            for _ in 0..3 {
                entry.recent_escalations.push(now);
            }
            entry.recent_cleanup_failures.push(now);
        }

        let debt = orch.cancellation_debt("w1").await;
        assert!(debt > 0.0);
        assert!(debt <= 1.0);
    }

    // 5. Cancellation debt is capped at 1.0
    #[tokio::test]
    async fn test_cancellation_debt_capped_at_one() {
        let orch = CancellationOrchestrator::new(test_config(), test_events());

        {
            let mut stats = orch.worker_stats.write().await;
            let entry = stats.entry("w1".to_string()).or_default();
            let now = Instant::now();
            for _ in 0..100 {
                entry.recent_cancellations.push(now);
            }
            for _ in 0..100 {
                entry.recent_escalations.push(now);
            }
            for _ in 0..100 {
                entry.recent_cleanup_failures.push(now);
            }
        }

        let debt = orch.cancellation_debt("w1").await;
        assert!((debt - 1.0).abs() < f64::EPSILON || debt <= 1.0);
    }

    // 6. CancellationConfig defaults are sensible
    #[test]
    fn test_cancellation_config_defaults() {
        let config = CancellationConfig::default();
        assert_eq!(config.grace_period, Duration::from_secs(5));
        assert_eq!(config.kill_timeout, Duration::from_secs(3));
        assert_eq!(config.remote_kill_timeout, Duration::from_secs(10));
        assert_eq!(config.max_escalations, 3);
        assert_eq!(config.cleanup_timeout, Duration::from_secs(15));
    }

    // 7. Cancel all with no active builds
    #[tokio::test]
    async fn test_cancel_all_no_active_builds() {
        let pool = WorkerPool::new();
        let history = Arc::new(BuildHistory::new(100));
        let ctx = make_test_context(pool, history);
        let orch = CancellationOrchestrator::new(test_config(), test_events());

        let resp = orch.cancel_all_builds(&ctx, false).await;
        assert_eq!(resp.status, "ok");
        assert_eq!(resp.cancelled_count, 0);
        assert!(resp.cancelled.is_empty());
    }

    // 8. CancellationState display
    #[test]
    fn test_cancellation_state_display() {
        assert_eq!(CancellationState::Requested.to_string(), "requested");
        assert_eq!(CancellationState::TermSent.to_string(), "term_sent");
        assert_eq!(
            CancellationState::RemoteKillSent.to_string(),
            "remote_kill_sent"
        );
        assert_eq!(CancellationState::Escalated.to_string(), "escalated");
        assert_eq!(CancellationState::Completed.to_string(), "completed");
        assert_eq!(CancellationState::Failed.to_string(), "failed");
    }

    // 9. CancelReason display
    #[test]
    fn test_cancel_reason_display() {
        assert_eq!(CancelReason::User.to_string(), "user");
        assert_eq!(CancelReason::Timeout.to_string(), "timeout");
        assert_eq!(CancelReason::StuckDetector.to_string(), "stuck_detector");
        assert_eq!(CancelReason::QueueEviction.to_string(), "queue_eviction");
    }

    // 10. Active cancellations list is empty by default
    #[tokio::test]
    async fn test_active_cancellations_empty() {
        let orch = CancellationOrchestrator::new(test_config(), test_events());
        let active = orch.active_cancellations().await;
        assert!(active.is_empty());
    }

    // 11. Record build increments total
    #[tokio::test]
    async fn test_record_build_increments_total() {
        let orch = CancellationOrchestrator::new(test_config(), test_events());
        orch.record_build("w1").await;
        orch.record_build("w1").await;

        let stats = orch.worker_stats.read().await;
        assert_eq!(stats["w1"].total_builds, 2);
    }

    // 12. Debt window prunes old entries
    #[tokio::test]
    async fn test_debt_prunes_old_cancellations() {
        let orch = CancellationOrchestrator::new(test_config(), test_events());

        {
            let mut stats = orch.worker_stats.write().await;
            let entry = stats.entry("w1".to_string()).or_default();
            // Add an "old" cancellation far in the past.
            // We can't set Instant directly to the past, but we can verify
            // that fresh entries produce non-zero debt.
            entry.recent_cancellations.push(Instant::now());
        }

        let debt = orch.cancellation_debt("w1").await;
        // One recent cancellation = rate_debt = 1/5 = 0.2, total ~ 0.2 * 0.4 = 0.08
        assert!(debt > 0.0);
        assert!(debt < 0.5); // Single cancel shouldn't be high.
    }

    #[test]
    fn test_cancellation_decision_path_term_only() {
        let record = test_record(CancellationState::Completed, 0, false);
        let path = cancellation_decision_path(&record);
        assert_eq!(path, vec!["requested", "term_sent", "completed"]);
        assert_eq!(cancellation_escalation_stage(&record), "term");
    }

    #[test]
    fn test_cancellation_decision_path_remote_kill() {
        let record = test_record(CancellationState::Completed, 1, true);
        let path = cancellation_decision_path(&record);
        assert_eq!(
            path,
            vec!["requested", "term_sent", "remote_kill_sent", "completed"]
        );
        assert_eq!(cancellation_escalation_stage(&record), "remote_kill");
    }

    #[test]
    fn test_cancellation_decision_path_sigkill_escalation() {
        let record = test_record(CancellationState::Completed, 2, true);
        let path = cancellation_decision_path(&record);
        assert_eq!(
            path,
            vec![
                "requested",
                "term_sent",
                "remote_kill_sent",
                "escalated",
                "completed"
            ]
        );
        assert_eq!(cancellation_escalation_stage(&record), "sigkill");
    }

    #[test]
    fn test_cancellation_decision_path_failed_before_remote_kill() {
        let record = test_record(CancellationState::Failed, 0, false);
        let path = cancellation_decision_path(&record);
        assert_eq!(path, vec!["requested", "term_sent", "failed"]);
        assert_eq!(cancellation_escalation_stage(&record), "term");
    }

    #[test]
    fn test_cancellation_decision_path_force_cancel() {
        let record = test_record(CancellationState::Completed, 0, true);
        let path = cancellation_decision_path(&record);
        assert_eq!(
            path,
            vec!["requested", "escalated", "remote_kill_sent", "completed"]
        );
        assert_eq!(cancellation_escalation_stage(&record), "remote_kill");
    }

    #[tokio::test]
    async fn test_cancellation_attempts_remote_kill_when_local_hook_exits_with_pgid_file() {
        let orch = CancellationOrchestrator::new(test_config(), test_events());
        let mut record = CancellationRecord {
            build_id: 42,
            worker_id: "missing-worker".to_string(),
            worker_endpoint: None,
            state: CancellationState::Requested,
            reason: CancelReason::User,
            requested_at: Instant::now(),
            completed_at: None,
            escalation_count: 0,
            remote_kill_attempted: false,
            cleanup_ok: true,
            slots: 1,
            slots_released: 0,
            hook_pid: 0,
            hook_process_identity: None,
            remote_pgid_file: None,
            abandoned_unlaunched: false,
            legacy_probe_config: None,
        };

        orch.execute_cancellation(&mut record, false).await;

        assert_eq!(record.state, CancellationState::Failed);
        assert!(!record.cleanup_ok);
        assert!(record.remote_kill_attempted);
        assert_eq!(record.escalation_count, 1);
        assert_eq!(
            cancellation_decision_path(&record),
            vec!["requested", "term_sent", "remote_kill_sent", "failed"]
        );
        assert_eq!(cancellation_escalation_stage(&record), "remote_kill");
    }

    #[tokio::test]
    async fn test_cancellation_without_remote_work_preserves_term_only_fast_path() {
        let orch = CancellationOrchestrator::new(test_config(), test_events());
        let mut record = CancellationRecord {
            build_id: 43,
            worker_id: "missing-worker".to_string(),
            worker_endpoint: None,
            state: CancellationState::Requested,
            reason: CancelReason::User,
            requested_at: Instant::now(),
            completed_at: None,
            escalation_count: 0,
            remote_kill_attempted: false,
            cleanup_ok: true,
            slots: 0,
            slots_released: 0,
            hook_pid: 0,
            hook_process_identity: None,
            remote_pgid_file: None,
            abandoned_unlaunched: false,
            legacy_probe_config: None,
        };

        orch.execute_cancellation(&mut record, false).await;

        assert_eq!(record.state, CancellationState::Completed);
        assert!(!record.remote_kill_attempted);
        assert_eq!(record.escalation_count, 0);
        assert_eq!(
            cancellation_decision_path(&record),
            vec!["requested", "term_sent", "completed"]
        );
        assert_eq!(cancellation_escalation_stage(&record), "term");
    }

    #[test]
    fn test_cancellation_operation_id_format() {
        assert_eq!(cancellation_operation_id(4242), "cancel-4242");
    }

    #[test]
    fn cancellation_safety_requires_both_termination_results() {
        for local in [false, true] {
            for remote in [false, true] {
                assert_eq!(
                    cancellation_terminal_state(local, remote),
                    if local && remote {
                        CancellationState::Completed
                    } else {
                        CancellationState::Failed
                    }
                );
            }
        }
    }

    #[tokio::test]
    async fn cancellation_safety_remote_failure_is_retryable_in_graceful_and_force_modes() {
        for force in [false, true] {
            let history = Arc::new(BuildHistory::new(100));
            let active = history.start_active_build(
                "unconfirmed".to_owned(),
                "missing-worker".to_owned(),
                "cargo test".to_owned(),
                0,
                1,
                rch_common::BuildLocation::Remote,
            );
            let ctx = make_test_context(WorkerPool::new(), history.clone());
            let orch = CancellationOrchestrator::new(test_config(), test_events());
            for _ in 0..2 {
                let response = orch
                    .cancel_build(&ctx, active.id, CancelReason::User, force)
                    .await;
                assert_eq!(response.status, "failed");
                assert_eq!(response.slots_released, 0);
                assert!(history.active_build(active.id).is_some());
                assert!(history.recent(10).is_empty());
                assert!(orch.active_cancellations().await.is_empty());
            }
            assert!(orch.cancellation_debt("missing-worker").await > 0.0);
        }
    }

    #[tokio::test]
    async fn cancellation_safety_deadline_preserves_real_reservations_and_exactly_once_cleanup() {
        let pool = WorkerPool::new();
        let config = rch_common::WorkerConfig::default();
        let id = config.id.clone();
        pool.add_worker(config).await;
        let worker = pool.get(&id).await.unwrap();
        assert!(worker.reserve_slots(3).await);
        let history = Arc::new(BuildHistory::new(100));
        let active = history.start_active_build(
            "reserved".to_owned(),
            id.to_string(),
            "cargo test".to_owned(),
            0,
            1,
            rch_common::BuildLocation::Remote,
        );
        let ctx = make_test_context(pool, history.clone());
        let orch = CancellationOrchestrator::new(
            CancellationConfig {
                cleanup_timeout: Duration::ZERO,
                ..test_config()
            },
            test_events(),
        );
        for force in [false, true] {
            let response = orch
                .cancel_build(&ctx, active.id, CancelReason::Timeout, force)
                .await;
            assert_eq!(response.status, "failed");
            assert_eq!(response.slots_released, 0);
            assert_eq!(worker.used_slots(), 3);
            assert!(!worker.reserve_slots(2).await);
            assert!(history.active_build(active.id).is_some());
            assert!(history.recent(10).is_empty());
        }

        // A later confirmed attempt may finalize once, without freeing the two
        // reservations belonging to other work or double-releasing on a race.
        let mut confirmed = test_record(CancellationState::Completed, 1, true);
        confirmed.build_id = active.id;
        confirmed.worker_id = id.to_string();
        confirmed.slots_released = 0;
        orch.run_cleanup(&ctx, &mut confirmed).await;
        assert_eq!(confirmed.slots_released, 1);
        assert_eq!(worker.used_slots(), 2);
        assert!(history.active_build(active.id).is_none());
        assert_eq!(history.recent(10).len(), 1);
        orch.run_cleanup(&ctx, &mut confirmed).await;
        assert_eq!(worker.used_slots(), 2);
        assert_eq!(history.recent(10).len(), 1);
    }

    #[tokio::test]
    async fn legacy_cancellation_never_uses_replacement_config_even_when_forced() {
        let pool = WorkerPool::new();
        let config = rch_common::WorkerConfig::default();
        let id = config.id.clone();
        pool.add_worker(config).await;
        let worker = pool.get(&id).await.unwrap();
        let orch = CancellationOrchestrator::new(
            CancellationConfig {
                cleanup_timeout: Duration::from_millis(40),
                remote_kill_timeout: Duration::from_secs(60),
                ..test_config()
            },
            test_events(),
        );
        for force in [false, true] {
            // The legacy build has no admitted endpoint. Even a current pool
            // entry cannot supply that missing historical identity.
            let lock = worker.config.write().await;
            let mut record = test_record(CancellationState::Requested, 0, false);
            record.worker_id = id.to_string();
            record.hook_pid = 0;
            record.slots_released = 0;
            tokio::time::timeout(
                Duration::from_secs(1),
                orch.execute_cancellation(&mut record, force),
            )
            .await
            .expect("unknown endpoint must not consult a locked replacement");
            assert_eq!(record.state, CancellationState::Failed);
            assert!(!record.cleanup_ok);
            assert!(record.remote_kill_attempted);
            drop(lock);
        }
    }

    #[tokio::test]
    async fn cancellation_safety_bulk_response_excludes_unconfirmed_builds() {
        let history = Arc::new(BuildHistory::new(100));
        let completed = history.start_active_build(
            "no-remote-work".to_owned(),
            "missing".to_owned(),
            "cargo check".to_owned(),
            0,
            0,
            rch_common::BuildLocation::Remote,
        );
        let retained = history.start_active_build(
            "remote-work".to_owned(),
            "missing".to_owned(),
            "cargo test".to_owned(),
            0,
            1,
            rch_common::BuildLocation::Remote,
        );
        let ctx = make_test_context(WorkerPool::new(), history.clone());
        let orch = CancellationOrchestrator::new(test_config(), test_events());
        let response = orch.cancel_all_builds(&ctx, false).await;
        assert_eq!(response.status, "partial");
        assert_eq!(response.cancelled_count, 1);
        assert_eq!(response.cancelled.len(), 1);
        assert_eq!(response.cancelled[0].build_id, completed.id);
        assert!(history.active_build(retained.id).is_some());
        let retry = orch.cancel_all_builds(&ctx, true).await;
        assert_eq!(retry.status, "failed");
        assert_eq!(retry.cancelled_count, 0);
        assert!(retry.cancelled.is_empty());
        assert!(history.active_build(retained.id).is_some());
    }

    #[tokio::test]
    async fn cancellation_safety_escalation_limit_does_not_fake_success() {
        let orch = CancellationOrchestrator::new(
            CancellationConfig {
                max_escalations: 0,
                ..test_config()
            },
            test_events(),
        );
        let mut record = test_record(CancellationState::Requested, 0, false);
        record.hook_pid = 0;
        record.slots_released = 0;
        orch.execute_cancellation(&mut record, false).await;
        assert_eq!(record.state, CancellationState::Failed);
        assert_eq!(record.escalation_count, 0);
        assert!(!record.remote_kill_attempted);
    }

    #[tokio::test]
    async fn cancellation_safety_local_signal_and_exit_probe_use_only_an_owned_child() {
        struct ChildGuard(std::process::Child);
        impl Drop for ChildGuard {
            fn drop(&mut self) {
                let _ = self.0.kill();
                let _ = self.0.wait();
            }
        }
        assert!(!send_signal_to_process(0, false));
        assert!(!send_signal_to_process(1, true));
        assert!(!send_signal_to_process(u32::MAX, true));
        assert!(!is_process_alive(0));
        assert!(!is_process_alive(u32::MAX));
        assert!(is_process_alive(std::process::id()));
        let mut child = ChildGuard(
            std::process::Command::new("/bin/sleep")
                .arg("60")
                .spawn()
                .unwrap(),
        );
        let pid = child.0.id();
        assert!(is_process_alive(pid));
        assert!(!wait_for_process_exit(pid, Duration::ZERO).await);
        assert!(send_signal_to_process(pid, false));
        child.0.wait().unwrap();
        assert!(wait_for_process_exit(pid, Duration::from_secs(1)).await);
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn wrapper_identity_distinguishes_exit_from_unknown_and_pidfd_never_adopts_reuse() {
        struct OwnedChild(std::process::Child);
        impl Drop for OwnedChild {
            fn drop(&mut self) {
                let _ = self.0.kill();
                let _ = self.0.wait();
            }
        }
        let mut child = OwnedChild(
            std::process::Command::new("/bin/sleep")
                .arg("60")
                .spawn()
                .unwrap(),
        );
        let pid = child.0.id();
        let identity = crate::history::process_identity(pid).unwrap();
        assert_eq!(
            wrapper_process_state(pid, Some(&identity)),
            WrapperProcessState::Running
        );
        for unknown in [None, Some(""), Some("unparseable:identity")] {
            assert_eq!(
                wrapper_process_state(pid, unknown),
                WrapperProcessState::Unverified
            );
        }
        for invalid in [1, u32::MAX] {
            assert_eq!(
                wrapper_process_state(invalid, Some(&identity)),
                WrapperProcessState::Unverified
            );
        }
        let (boot, ticks) = identity.rsplit_once(':').unwrap();
        let previous = format!("{boot}:{}", ticks.parse::<u64>().unwrap() + 1);
        let mut record = test_record(CancellationState::Requested, 0, false);
        record.hook_pid = pid;
        record.hook_process_identity = Some(previous);
        let owner = CancellationOrchestrator::new(test_config(), test_events());
        assert_eq!(
            wrapper_process_state(pid, record.hook_process_identity.as_deref()),
            WrapperProcessState::Exited,
            "modelled prior incarnation is gone, but the current occupant is alive"
        );
        assert!(CancellationOrchestrator::verified_hook_handle(&record).is_none());
        assert!(owner.send_verified_signal(&record, true).await);
        assert!(wait_for_wrapper_exit(&record, Duration::ZERO).await);
        assert!(
            child.0.try_wait().unwrap().is_none(),
            "reused PID was signalled"
        );

        record.hook_process_identity = Some(identity);
        assert!(!wait_for_wrapper_exit(&record, Duration::ZERO).await);
        let handle = CancellationOrchestrator::verified_hook_handle(&record).unwrap();
        assert!(owner.send_verified_signal(&record, false).await);
        child.0.wait().unwrap();
        assert!(wait_for_wrapper_exit(&record, Duration::from_secs(1)).await);
        // The retained kernel handle refers to the exited process even if its
        // numeric PID is reused later; it cannot signal a replacement owner.
        assert_eq!(
            rustix::process::pidfd_send_signal(&handle, rustix::process::Signal::KILL),
            Err(rustix::io::Errno::SRCH)
        );
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn recovered_wrapper_exit_reconciles_real_remote_group_once() {
        use std::os::unix::fs::PermissionsExt;
        use std::os::unix::process::CommandExt;

        const CHILD_ROOT: &str = "RCH_RECOVERED_CANCEL_TEST_ROOT";
        struct OwnedChild(std::process::Child);
        impl Drop for OwnedChild {
            fn drop(&mut self) {
                let _ = self.0.kill();
                let _ = self.0.wait();
            }
        }
        struct OwnedGroup(std::process::Child);
        impl Drop for OwnedGroup {
            fn drop(&mut self) {
                // Keep the leader unreaped until cleanup, preventing reuse of
                // this test-owned group ID before the final group signal.
                let _ = nix::sys::signal::killpg(
                    nix::unistd::Pid::from_raw(i32::try_from(self.0.id()).unwrap()),
                    nix::sys::signal::Signal::SIGKILL,
                );
                let _ = self.0.wait();
            }
        }
        if let Some(root) = std::env::var_os(CHILD_ROOT).map(std::path::PathBuf::from) {
            for (scenario, force, model_reuse) in [("exited", false, false), ("reused", true, true)]
            {
                let case = root.join(scenario);
                std::fs::create_dir(&case).unwrap();
                let history_path = case.join("history.jsonl");
                let identity_path = case.join("remote.pgid");
                let counter_path = case.join("executions");
                let child_path = case.join("remote-child.pid");
                let mut wrapper = OwnedChild(
                    std::process::Command::new("/bin/sleep")
                        .arg("60")
                        .spawn()
                        .unwrap(),
                );
                let wrapper_pid = wrapper.0.id();
                let unrelated = OwnedGroup(
                    std::process::Command::new("/bin/sleep")
                        .arg("60")
                        .process_group(0)
                        .spawn()
                        .unwrap(),
                );
                let remote = OwnedGroup(
                    std::process::Command::new("/bin/sh")
                        .args([
                            "-c",
                            "printf 'run\\n' >> \"$1\"; trap '' TERM; sleep 60 & printf '%s\\n' \"$!\" > \"$2\"; wait",
                            "recovery-workload",
                        ])
                        .arg(&counter_path)
                        .arg(&child_path)
                        .process_group(0)
                        .spawn()
                        .unwrap(),
                );
                let remote_child: u32 = tokio::time::timeout(Duration::from_secs(3), async {
                    loop {
                        if let Ok(text) = std::fs::read_to_string(&child_path)
                            && let Ok(pid) = text.trim().parse()
                        {
                            break pid;
                        }
                        tokio::time::sleep(Duration::from_millis(10)).await;
                    }
                })
                .await
                .unwrap();
                let config = rch_common::WorkerConfig::default();
                let worker_id = config.id.clone();
                let admitted_worker = crate::workers::WorkerState::new(config.clone());
                let history = BuildHistory::new(100).with_persistence(history_path.clone());
                let wrapper_id = format!("recovery-{scenario}");
                let target = history
                    .try_start_active_build_with_waiter(
                        "recover-target".into(),
                        worker_id.to_string(),
                        "cargo test".into(),
                        wrapper_pid,
                        Some(wrapper_id.clone()),
                        1,
                        rch_common::BuildLocation::Remote,
                        None,
                        crate::disk_pressure::DiskHeadroomAdmission::default(),
                        Some(admitted_worker.endpoint_snapshot().await),
                    )
                    .unwrap()
                    .unwrap();
                let other = history.start_active_build_with_wrapper(
                    "unrelated-target".into(),
                    worker_id.to_string(),
                    "cargo check".into(),
                    unrelated.0.id(),
                    Some("unrelated-wrapper".into()),
                    2,
                    rch_common::BuildLocation::Remote,
                );
                history
                    .record_build_heartbeat(rch_common::BuildHeartbeatRequest {
                        build_id: target.id,
                        worker_id: worker_id.clone(),
                        hook_pid: Some(wrapper_pid),
                        local_wrapper_id: Some(wrapper_id.clone()),
                        remote_pgid_file: Some(identity_path.to_string_lossy().into_owned()),
                        phase: rch_common::BuildHeartbeatPhase::Execute,
                        detail: None,
                        progress_counter: None,
                        progress_percent: None,
                    })
                    .unwrap();
                wrapper.0.kill().unwrap();
                wrapper.0.wait().unwrap();
                drop(history);
                if model_reuse {
                    // Model PID reuse in the durable prior-daemon record while
                    // keeping a real unrelated process alive as the sentinel.
                    let path = history_path.with_extension("ownership.json");
                    let mut journal: serde_json::Value =
                        serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
                    let state = journal["active"]
                        .as_array_mut()
                        .unwrap()
                        .iter_mut()
                        .find(|state| state["id"].as_u64() == Some(target.id))
                        .unwrap();
                    let actual = crate::history::process_identity(unrelated.0.id()).unwrap();
                    let (boot, ticks) = actual.rsplit_once(':').unwrap();
                    state["hook_pid"] = serde_json::json!(unrelated.0.id());
                    state["hook_process_identity"] =
                        serde_json::json!(format!("{boot}:{}", ticks.parse::<u64>().unwrap() + 1));
                    std::fs::write(&path, serde_json::to_vec(&journal).unwrap()).unwrap();
                }
                let history = Arc::new(BuildHistory::load_from_file(&history_path, 100).unwrap());
                assert!(history.active_build(target.id).unwrap().recovered);
                let pool = WorkerPool::new();
                pool.add_worker(config).await;
                let worker = pool.get(&worker_id).await.unwrap();
                for active in history.active_builds() {
                    worker.restore_slots(active.slots).unwrap();
                }
                assert_eq!(worker.used_slots(), 3);
                let ctx = make_test_context(pool, history.clone());
                let owner = CancellationOrchestrator::new(
                    CancellationConfig {
                        remote_kill_timeout: Duration::from_secs(5),
                        cleanup_timeout: Duration::from_secs(10),
                        ..test_config()
                    },
                    test_events(),
                );
                let observed = crate::history::process_identity(remote.0.id()).unwrap();
                let (boot, ticks) = observed.rsplit_once(':').unwrap();
                let valid = format!(
                    "RCH_REMOTE_PROCESS_V1:{}:{boot}:{}:{ticks}\n",
                    target.id,
                    remote.0.id()
                );
                std::fs::write(&identity_path, &valid).unwrap();
                for failure in ["unavailable", "wrong-build"] {
                    std::fs::write(root.join("transport"), failure).unwrap();
                    if failure == "wrong-build" {
                        std::fs::write(
                            &identity_path,
                            format!(
                                "RCH_REMOTE_PROCESS_V1:{}:{boot}:{}:{ticks}\n",
                                target.id + 1,
                                remote.0.id()
                            ),
                        )
                        .unwrap();
                    }
                    let response = owner
                        .cancel_build(&ctx, target.id, CancelReason::User, force)
                        .await;
                    assert_eq!(response.status, "failed", "{scenario}/{failure}");
                    assert_eq!(response.slots_released, 0);
                    assert_eq!(worker.used_slots(), 3);
                    assert!(history.active_build(target.id).is_some());
                    assert!(is_process_alive(remote.0.id()));
                    assert!(is_process_alive(unrelated.0.id()));
                }
                std::fs::write(root.join("transport"), "connected").unwrap();
                std::fs::write(&identity_path, valid).unwrap();
                let config_lock = worker.config.write().await;
                let task_owner = owner.clone();
                let task_context = ctx.clone();
                let completion = tokio::spawn(async move {
                    task_owner
                        .cancel_build(&task_context, target.id, CancelReason::User, force)
                        .await
                });
                wait_for_cancellation_attempts(&owner, 1).await;
                let duplicate = owner
                    .cancel_build(&ctx, target.id, CancelReason::User, force)
                    .await;
                assert_eq!(duplicate.status, "cancelling");
                assert_eq!(duplicate.slots_released, 0);
                drop(config_lock);
                let completed = completion.await.unwrap();
                assert_eq!(completed.status, "cancelled", "{completed:?}");
                assert_eq!(completed.slots_released, 1);
                assert_eq!(worker.used_slots(), 2);
                assert!(!is_process_alive(remote.0.id()));
                assert!(!is_process_alive(remote_child));
                assert!(is_process_alive(unrelated.0.id()));
                assert!(history.active_build(other.id).is_some());
                assert!(history.active_build(target.id).is_none());
                assert_eq!(
                    history
                        .terminal_build(target.id, &wrapper_id)
                        .unwrap()
                        .exit_code,
                    130
                );
                assert_eq!(std::fs::read_to_string(&counter_path).unwrap(), "run\n");
                let duplicate = owner
                    .cancel_build(&ctx, target.id, CancelReason::User, force)
                    .await;
                assert_eq!(duplicate.slots_released, 0);
                assert_eq!(worker.used_slots(), 2);
                let reopened = BuildHistory::load_from_file(&history_path, 100).unwrap();
                assert!(reopened.active_build(target.id).is_none());
                assert!(reopened.active_build(other.id).is_some());
                assert!(reopened.terminal_build(target.id, &wrapper_id).is_some());
            }
            std::fs::write(root.join("completed"), "ok").unwrap();
            return;
        }
        // Only SSH transport is replaced. Its payload is the exact production
        // identity/termination protocol operating on real controlled groups.
        let root = tempfile::tempdir().unwrap().keep();
        let ssh = root.join("ssh");
        std::fs::write(
            &ssh,
            r#"#!/bin/sh
mode=$(/bin/cat "$RCH_RECOVERED_CANCEL_TEST_ROOT/transport") || exit 99
[ "$mode" != unavailable ] || exit 255
for argument do payload=$argument; done
exec /bin/sh -c "$payload"
"#,
        )
        .unwrap();
        std::fs::set_permissions(&ssh, std::fs::Permissions::from_mode(0o700)).unwrap();
        let output = tokio::time::timeout(
            Duration::from_secs(40),
            tokio::process::Command::new(std::env::current_exe().unwrap())
                .args([
                    "--exact",
                    "cancellation::tests::recovered_wrapper_exit_reconciles_real_remote_group_once",
                    "--nocapture",
                ])
                .env(CHILD_ROOT, &root)
                .env("PATH", format!("{}:/usr/bin:/bin", root.display()))
                .kill_on_drop(true)
                .output(),
        )
        .await
        .unwrap()
        .unwrap();
        assert!(
            output.status.success(),
            "isolated recovery failed: {output:?}"
        );
        assert!(root.join("completed").is_file());
    }

    #[test]
    fn remote_cancel_receipt_requires_success_and_one_complete_matching_record() {
        use std::os::unix::process::ExitStatusExt;

        for (code, stdout, expected) in [
            (0, "RCH_REMOTE_CANCELLED_V1:42\n", true),
            (0, "", false),
            (0, "RCH_REMOTE_CANCELLED_V1:42", false),
            (0, "RCH_REMOTE_CANCELLED_V1:41\n", false),
            (
                0,
                "RCH_REMOTE_CANCELLED_V1:42\nRCH_REMOTE_CANCELLED_V1:42\n",
                false,
            ),
            (255, "RCH_REMOTE_CANCELLED_V1:42\n", false),
        ] {
            let output = std::process::Output {
                status: std::process::ExitStatus::from_raw(code << 8),
                stdout: stdout.as_bytes().to_vec(),
                stderr: Vec::new(),
            };
            assert_eq!(
                remote_kill_confirmed(&output, 42),
                expected,
                "{code}: {stdout:?}"
            );
        }
    }

    async fn remote_cancel_probe_fixture(
        path: &std::path::Path,
        prefix: &str,
    ) -> std::process::Output {
        tokio::time::timeout(
            Duration::from_secs(5),
            tokio::process::Command::new("/bin/sh")
                .arg("-c")
                .arg(format!(
                    "{}\n{prefix}\n{REMOTE_CANCELLATION_SCRIPT}",
                    rch_common::REMOTE_PROCESS_IDENTITY_SCRIPT
                ))
                .arg("rch-cancel-fixture")
                .arg(path)
                .arg("42")
                .stdin(std::process::Stdio::null())
                .kill_on_drop(true)
                .output(),
        )
        .await
        .expect("remote cancellation probe exceeded its test deadline")
        .unwrap()
    }

    #[cfg(target_os = "linux")]
    struct RemoteCancelOwnedGroup(std::process::Child);

    #[cfg(target_os = "linux")]
    impl Drop for RemoteCancelOwnedGroup {
        fn drop(&mut self) {
            // Keep the leader unreaped until here, so its PID cannot be
            // reused by another group before this owned cleanup.
            let pid = i32::try_from(self.0.id()).unwrap();
            let _ = nix::sys::signal::killpg(
                nix::unistd::Pid::from_raw(pid),
                nix::sys::signal::Signal::SIGKILL,
            );
            let _ = self.0.wait();
        }
    }

    #[cfg(target_os = "linux")]
    fn remote_cancel_record(build_id: u64, pgid: u32) -> String {
        let identity = crate::history::process_identity(pgid).unwrap();
        let (boot, start) = identity.rsplit_once(':').unwrap();
        format!("RCH_REMOTE_PROCESS_V1:{build_id}:{boot}:{pgid}:{start}\n")
    }

    #[tokio::test]
    async fn remote_cancel_rejects_unsafe_and_missing_identity_without_signalling() {
        let root = tempfile::tempdir().unwrap().keep();
        let path = root.join("group with 'quote $dollar;`backtick`.pgid");
        // The spy cannot send a real signal even if validation regresses.
        let spy = "kill() { printf SIGNALLED; return 0; }; sleep() { :; };";
        for value in [
            "",
            "0",
            "1",
            "-1",
            "-42",
            "00",
            "012",
            "2 3",
            "2\n3",
            "2147483648",
            "999999999999999999999",
            "$(false)",
        ] {
            std::fs::write(&path, value).unwrap();
            let output = remote_cancel_probe_fixture(&path, spy).await;
            assert_eq!(output.status.code(), Some(42), "{value:?}");
            assert!(
                output.stdout.is_empty(),
                "invalid ID reached a signal: {value:?}"
            );
        }
        let link = root.join("link.pgid");
        std::os::unix::fs::symlink(&path, &link).unwrap();
        for invalid in [root.join("missing.pgid"), root.clone(), link] {
            let output = remote_cancel_probe_fixture(&invalid, spy).await;
            assert_eq!(output.status.code(), Some(42));
            assert!(!remote_kill_confirmed(&output, 42));
            // Only a truly missing record is reported as absent; a directory
            // or a symlink in its place is an unreadable record, not absence.
            let missing = invalid.ends_with("missing.pgid");
            assert_eq!(remote_record_absent(&output, 42), missing, "{invalid:?}");
            if !missing {
                assert!(output.stdout.is_empty());
            }
        }
        let script = build_remote_kill_script(None, 42);
        assert!(
            !script.contains("pkill"),
            "missing identity must not select processes by text"
        );
        let output = tokio::process::Command::new("/bin/sh")
            .args(["-c", &script])
            .kill_on_drop(true)
            .output()
            .await
            .unwrap();
        assert!(!remote_kill_confirmed(&output, 42));
    }

    #[tokio::test]
    async fn remote_cancel_observation_errors_and_survivors_never_confirm_success() {
        let root = tempfile::tempdir().unwrap().keep();
        let path = root.join("group.pgid");
        std::fs::write(&path, "123\n").unwrap();
        // Execute the production shell source with only OS failures injected.
        // All kill calls are inert; sleep is shortened for deterministic faults.
        for (probe, expected) in [
            ("ps() { return 1; };", 44),
            ("ps() { :; };", 44),
            ("ps() { printf 'not a process table\\n'; };", 44),
            ("ps() { printf '10 10 S\\n'; };", 44),
            ("ps() { printf '%s 123 S\\n' \"$$\"; };", 44),
            ("ps() { printf '%s 99 S\\n100 123 R\\n' \"$$\"; };", 43),
            (
                "ps() { printf '%s 99 S\\n100 123 Z\\n101 123 S\\n' \"$$\"; };",
                43,
            ),
            ("ps() { printf '%s 99 S\\n100 123 Z\\n' \"$$\"; };", 0),
            ("ps() { printf '%s 99 S\\n' \"$$\"; };", 0),
        ] {
            let prefix = format!("kill() {{ return 0; }}; sleep() {{ :; }}; {probe}");
            let output = remote_cancel_probe_fixture(&path, &prefix).await;
            assert_eq!(output.status.code(), Some(expected), "{probe}: {output:?}");
            assert_eq!(remote_kill_confirmed(&output, 42), expected == 0);
        }

        #[cfg(target_os = "linux")]
        {
            use std::os::unix::process::CommandExt;

            let group = RemoteCancelOwnedGroup(
                std::process::Command::new("/bin/sleep")
                    .arg("60")
                    .process_group(0)
                    .spawn()
                    .unwrap(),
            );
            let pgid = group.0.id();
            std::fs::write(&path, remote_cancel_record(42, pgid)).unwrap();
            // The leader has real matching boot/start evidence. Only OS
            // signals and observations are injected: the reported survivor
            // must keep cancellation unconfirmed after bounded escalation.
            let prefix = format!(
                "kill() {{ return 0; }}; sleep() {{ :; }}; \
                 ps() {{ printf '%s 1 S\\n{pgid} {pgid} S\\n' \"$$\"; }};"
            );
            let output = remote_cancel_probe_fixture(&path, &prefix).await;
            assert_eq!(output.status.code(), Some(45), "{output:?}");
            assert!(!remote_kill_confirmed(&output, 42));
            assert!(is_process_alive(pgid));
        }
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn remote_cancel_rejects_stale_identity_without_signalling_owned_sentinel() {
        use std::os::unix::process::CommandExt;

        let sentinel = RemoteCancelOwnedGroup(
            std::process::Command::new("/bin/sleep")
                .arg("60")
                .process_group(0)
                .spawn()
                .unwrap(),
        );
        let pgid = sentinel.0.id();
        let identity = crate::history::process_identity(pgid).unwrap();
        let (boot, start) = identity.rsplit_once(':').unwrap();
        let other_boot = if boot == "00000000-0000-0000-0000-000000000000" {
            "11111111-1111-1111-1111-111111111111"
        } else {
            "00000000-0000-0000-0000-000000000000"
        };
        let prior_start = start.parse::<u64>().unwrap().checked_add(1).unwrap();
        let root = tempfile::tempdir().unwrap().keep();
        let path = root.join("identity.pgid");
        for (case, record) in [
            ("wrong-build", remote_cancel_record(41, pgid)),
            (
                "wrong-boot",
                format!("RCH_REMOTE_PROCESS_V1:42:{other_boot}:{pgid}:{start}\n"),
            ),
            (
                // Model a reused leader PID with a different recorded start;
                // the current PID occupant is a real controlled sentinel.
                "reused-leader-start",
                format!("RCH_REMOTE_PROCESS_V1:42:{boot}:{pgid}:{prior_start}\n"),
            ),
            ("legacy-live-group", format!("{pgid}\n")),
        ] {
            std::fs::write(&path, &record).unwrap();
            let output = remote_cancel_probe_fixture(&path, "").await;
            assert_eq!(output.status.code(), Some(43), "{case}: {output:?}");
            assert!(
                output.stdout.is_empty(),
                "{case}: false cancellation receipt"
            );
            assert!(
                is_process_alive(pgid),
                "{case}: signalled unrelated occupant"
            );
            assert_eq!(
                crate::history::process_identity(pgid).as_deref(),
                Some(identity.as_str())
            );
            assert_eq!(std::fs::read_to_string(&path).unwrap(), record);
        }
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn remote_cancel_retains_live_orphan_when_recorded_leader_cannot_be_verified() {
        use std::os::unix::process::CommandExt;

        let root = tempfile::tempdir().unwrap().keep();
        let path = root.join("orphan.pgid");
        let child_pid_path = root.join("orphan-child.pid");
        let script = format!(
            "{}\nrch_remote_record \"$1\" 42 || exit 99\n\
             sleep 60 &\n\
             printf '%s\\n' \"$!\" > \"$2\"\n\
             exit 0\n",
            rch_common::REMOTE_PROCESS_IDENTITY_SCRIPT
        );
        let group = RemoteCancelOwnedGroup(
            std::process::Command::new("/bin/sh")
                .args(["-c", &script, "rch-orphan-fixture"])
                .arg(&path)
                .arg(&child_pid_path)
                .process_group(0)
                .spawn()
                .unwrap(),
        );
        let child_pid: u32 = tokio::time::timeout(Duration::from_secs(3), async {
            loop {
                if let Ok(text) = std::fs::read_to_string(&child_pid_path)
                    && let Ok(pid) = text.trim().parse::<u32>()
                    && !is_process_alive(group.0.id())
                {
                    break pid;
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .unwrap();
        let record = std::fs::read_to_string(&path).unwrap();
        assert!(record.starts_with("RCH_REMOTE_PROCESS_V1:42:"));
        assert!(is_process_alive(child_pid));
        for _ in 0..2 {
            let output = remote_cancel_probe_fixture(&path, "").await;
            assert_eq!(output.status.code(), Some(43), "{output:?}");
            assert!(output.stdout.is_empty());
            assert!(
                is_process_alive(child_pid),
                "unverified orphan was signalled"
            );
            assert_eq!(std::fs::read_to_string(&path).unwrap(), record);
        }
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn remote_cancel_real_process_group_is_verified_and_retryable() {
        use std::os::unix::process::CommandExt;

        let root = tempfile::tempdir().unwrap().keep();
        let child_pid_path = root.join("child.pid");
        let group = RemoteCancelOwnedGroup(
            std::process::Command::new("/bin/sh")
                .args([
                    "-c",
                    "trap '' TERM; sleep 60 & printf '%s\\n' \"$!\" > \"$1\"; wait",
                    "rch-group",
                ])
                .arg(&child_pid_path)
                .process_group(0)
                .spawn()
                .unwrap(),
        );
        let unrelated = RemoteCancelOwnedGroup(
            std::process::Command::new("/bin/sleep")
                .arg("60")
                .process_group(0)
                .spawn()
                .unwrap(),
        );
        let child_pid: u32 = tokio::time::timeout(Duration::from_secs(3), async {
            loop {
                if let Ok(text) = std::fs::read_to_string(&child_pid_path)
                    && let Ok(pid) = text.trim().parse::<u32>()
                {
                    break pid;
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .unwrap();
        let path = root.join("group with 'quote $dollar;`backtick`.pgid");
        std::fs::write(&path, remote_cancel_record(42, group.0.id())).unwrap();
        let script = build_remote_kill_script(Some(path.to_str().unwrap()), 42);
        for _ in 0..2 {
            let output = tokio::time::timeout(
                Duration::from_secs(8),
                tokio::process::Command::new("/bin/sh")
                    .args(["-c", &script])
                    .kill_on_drop(true)
                    .output(),
            )
            .await
            .unwrap()
            .unwrap();
            assert!(remote_kill_confirmed(&output, 42), "{output:?}");
            assert!(!is_process_alive(group.0.id()));
            assert!(!is_process_alive(child_pid));
            assert!(is_process_alive(unrelated.0.id()));
        }
        // Legacy records may confirm an already empty group, but the live
        // legacy sentinel case above proves they never grant signalling rights.
        std::fs::write(&path, format!("{}\n", group.0.id())).unwrap();
        let output = remote_cancel_probe_fixture(&path, "").await;
        assert!(remote_kill_confirmed(&output, 42), "{output:?}");
        assert!(is_process_alive(unrelated.0.id()));
    }

    #[tokio::test]
    async fn remote_cancel_transport_requires_receipt_before_releasing_reservations() {
        const CHILD_ROOT: &str = "RCH_REMOTE_CANCEL_TEST_ROOT";
        if let Some(root) = std::env::var_os(CHILD_ROOT).map(std::path::PathBuf::from) {
            let pool = WorkerPool::new();
            let config = rch_common::WorkerConfig {
                host: "admitted.example".into(),
                user: "admitted-user".into(),
                identity_file: "/admitted/key with spaces".into(),
                ..rch_common::WorkerConfig::default()
            };
            let id = config.id.clone();
            pool.add_worker(config.clone()).await;
            let worker = pool.get(&id).await.unwrap();
            assert!(worker.reserve_slots(3).await);
            let history = Arc::new(BuildHistory::new(100));
            let active = history
                .try_start_active_build_with_waiter(
                    "receipt".to_owned(),
                    id.to_string(),
                    "cargo test".to_owned(),
                    0,
                    None,
                    1,
                    rch_common::BuildLocation::Remote,
                    None,
                    crate::disk_pressure::DiskHeadroomAdmission::default(),
                    Some(worker.endpoint_snapshot().await),
                )
                .unwrap()
                .unwrap();
            // Running ownership remains on the original host after this ID
            // points elsewhere; the replacement keeps its other reservations.
            pool.add_worker(rch_common::WorkerConfig {
                host: "replacement.example".into(),
                user: "replacement-user".into(),
                identity_file: "/replacement/key".into(),
                ..config.clone()
            })
            .await;
            std::fs::write(root.join("build-id"), active.id.to_string()).unwrap();
            let ctx = make_test_context(pool, history.clone());
            let orch = CancellationOrchestrator::new(test_config(), test_events());
            for mode in [
                "empty",
                "truncated",
                "wrong-id",
                "duplicate",
                "failed",
                "hang",
                "confirmed",
            ] {
                std::fs::write(root.join("mode"), mode).unwrap();
                let mut record = test_record(CancellationState::Requested, 0, false);
                record.build_id = active.id;
                record.worker_id = id.to_string();
                record.worker_endpoint = active.worker_endpoint.clone();
                record.hook_pid = 0;
                record.slots_released = 0;
                record.remote_pgid_file = Some("/test/owned-group.pgid".to_owned());
                if mode == "hang" {
                    let bounded = CancellationOrchestrator::new(
                        CancellationConfig {
                            cleanup_timeout: Duration::from_millis(200),
                            remote_kill_timeout: Duration::from_secs(60),
                            ..test_config()
                        },
                        test_events(),
                    );
                    tokio::time::timeout(
                        Duration::from_secs(2),
                        bounded.execute_cancellation(&mut record, true),
                    )
                    .await
                    .expect("overall deadline must bound a hanging admitted transport");
                } else {
                    orch.execute_cancellation(&mut record, true).await;
                }
                let argv = std::fs::read_to_string(root.join("ssh-argv")).unwrap();
                let argv: Vec<_> = argv.lines().collect();
                assert!(
                    argv.windows(2)
                        .any(|args| args == ["-i", "/admitted/key with spaces"])
                );
                assert!(argv.contains(&"admitted-user@admitted.example"));
                assert!(!argv.contains(&"replacement-user@replacement.example"));
                if mode == "hang" {
                    let pid = std::fs::read_to_string(root.join("hanging-ssh-pid"))
                        .unwrap()
                        .trim()
                        .parse()
                        .unwrap();
                    assert!(
                        wait_for_process_exit(pid, Duration::from_secs(2)).await,
                        "deadline abandoned the owned SSH process"
                    );
                }
                orch.run_cleanup(&ctx, &mut record).await;
                if mode == "confirmed" {
                    assert_eq!(record.state, CancellationState::Completed);
                    assert_eq!(record.slots_released, 1);
                    assert_eq!(worker.used_slots(), 2);
                    assert!(history.active_build(active.id).is_none());
                    orch.run_cleanup(&ctx, &mut record).await;
                    assert_eq!(worker.used_slots(), 2);
                } else {
                    assert_eq!(record.state, CancellationState::Failed, "{mode}");
                    assert_eq!(record.slots_released, 0);
                    assert_eq!(worker.used_slots(), 3);
                    assert!(history.active_build(active.id).is_some());
                    assert!(history.recent(10).is_empty());
                }
            }
            std::fs::write(root.join("child-completed"), "ok").unwrap();
            return;
        }

        // Isolate PATH in a child test process; never change global test env or
        // contact a worker. The fake SSH controls only exit/status receipts.
        use std::os::unix::fs::PermissionsExt;
        let root = tempfile::tempdir().unwrap().keep();
        let ssh = root.join("ssh");
        std::fs::write(
            &ssh,
            r#"#!/bin/sh
id=$(/bin/cat "$RCH_REMOTE_CANCEL_TEST_ROOT/build-id") || exit 99
mode=$(/bin/cat "$RCH_REMOTE_CANCEL_TEST_ROOT/mode") || exit 99
printf '%s\n' "$@" > "$RCH_REMOTE_CANCEL_TEST_ROOT/ssh-argv"
case "$mode" in
  empty) exit 0;;
  truncated) printf 'RCH_REMOTE_CANCELLED_V1:%s' "$id";;
  wrong-id) printf 'RCH_REMOTE_CANCELLED_V1:0\n';;
  duplicate) printf 'RCH_REMOTE_CANCELLED_V1:%s\n' "$id" "$id";;
  failed) printf 'RCH_REMOTE_CANCELLED_V1:%s\n' "$id"; exit 255;;
  hang) printf '%s\n' "$$" > "$RCH_REMOTE_CANCEL_TEST_ROOT/hanging-ssh-pid"; exec /bin/sleep 60;;
  confirmed) printf 'RCH_REMOTE_CANCELLED_V1:%s\n' "$id";;
  *) exit 99;;
esac
"#,
        )
        .unwrap();
        std::fs::set_permissions(&ssh, std::fs::Permissions::from_mode(0o700)).unwrap();
        let output = tokio::time::timeout(
            Duration::from_secs(15),
            tokio::process::Command::new(std::env::current_exe().unwrap())
                .args(["--exact", "cancellation::tests::remote_cancel_transport_requires_receipt_before_releasing_reservations", "--nocapture"])
                .env(CHILD_ROOT, &root)
                .env("PATH", format!("{}:/usr/bin:/bin", root.display()))
                .kill_on_drop(true)
                .output(),
        )
        .await
        .unwrap()
        .unwrap();
        assert!(
            output.status.success(),
            "isolated SSH fixture failed: {output:?}"
        );
        assert!(
            root.join("child-completed").is_file(),
            "child regression did not execute"
        );
    }

    async fn wait_for_cancellation_attempts(orch: &CancellationOrchestrator, expected: usize) {
        tokio::time::timeout(Duration::from_secs(3), async {
            while orch.active_cancellations().await.len() != expected {
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("cancellation attempt ownership did not settle");
    }

    #[tokio::test]
    async fn cancellation_ownership_dropped_waiter_preserves_reservations_and_retryability() {
        for force in [false, true] {
            for budget in [Duration::ZERO, Duration::from_millis(40)] {
                let pool = WorkerPool::new();
                let config = rch_common::WorkerConfig::default();
                let id = config.id.clone();
                pool.add_worker(config).await;
                let worker = pool.get(&id).await.unwrap();
                assert!(worker.reserve_slots(3).await);
                let history = Arc::new(BuildHistory::new(100));
                let build = history.start_active_build(
                    "abandoned-waiter".to_owned(),
                    id.to_string(),
                    "cargo test".to_owned(),
                    0,
                    1,
                    rch_common::BuildLocation::Remote,
                );
                let ctx = make_test_context(pool, history.clone());
                let mut orch = CancellationOrchestrator::new(
                    CancellationConfig {
                        cleanup_timeout: budget,
                        ..test_config()
                    },
                    test_events(),
                );
                // Legacy ownership has no admitted remote address. It fails
                // closed, then blocks only while capturing diagnostic health.
                let lock = worker.config.write().await;
                let caller_owner = orch.clone();
                let caller_context = ctx.clone();
                let caller = tokio::spawn(async move {
                    caller_owner
                        .cancel_build(&caller_context, build.id, CancelReason::User, force)
                        .await
                });
                wait_for_cancellation_attempts(&orch, 1).await;
                caller.abort();
                assert!(caller.await.unwrap_err().is_cancelled());
                let duplicate = orch
                    .cancel_build(&ctx, build.id, CancelReason::User, force)
                    .await;
                assert_eq!(duplicate.status, "cancelling");
                assert_eq!(duplicate.slots_released, 0);
                // Keep cleanup pending past the termination budget so a
                // caller's cancellation cannot abandon the owned attempt.
                tokio::time::sleep(budget + Duration::from_millis(30)).await;
                drop(lock);
                wait_for_cancellation_attempts(&orch, 0).await;
                assert_eq!(worker.used_slots(), 3);
                assert!(history.active_build(build.id).is_some());
                assert!(history.recent(10).is_empty());
                // Retry uses the same shared admission map, without SSH.
                orch.config.cleanup_timeout = Duration::ZERO;
                let retry = orch
                    .cancel_build(&ctx, build.id, CancelReason::User, force)
                    .await;
                assert_eq!(retry.status, "failed", "abandoned attempt blocked a retry");
                assert_eq!(worker.used_slots(), 3);
                assert!(orch.active_cancellations().await.is_empty());
                assert_eq!(
                    orch.worker_stats.read().await[id.as_str()]
                        .recent_cancellations
                        .len(),
                    2,
                    "duplicate waits must not create extra cancellation attempts"
                );
            }
        }
    }

    #[tokio::test]
    async fn cancellation_ownership_guard_releases_only_its_claim_on_abort_and_unwind() {
        let orch = CancellationOrchestrator::new(test_config(), test_events());
        for unwind in [false, true] {
            {
                let mut active = orch.active.lock().unwrap();
                active.insert(42, test_record(CancellationState::Requested, 0, false));
                let mut other = test_record(CancellationState::Requested, 0, false);
                other.build_id = 43;
                active.insert(43, other);
            }
            let attempt = CancellationAttempt {
                build_id: 42,
                active: Arc::clone(&orch.active),
            };
            let task = tokio::spawn(async move {
                let _attempt = attempt;
                assert!(!unwind, "injected cancellation task unwind");
                std::future::pending::<()>().await;
            });
            if !unwind {
                task.abort();
            }
            let error = task.await.unwrap_err();
            assert_eq!(error.is_panic(), unwind);
            let active = orch.active_cancellations().await;
            assert_eq!(active.len(), 1);
            assert_eq!(active[0].build_id, 43);
        }
    }

    struct CancellationOwnedChild(std::process::Child);

    impl Drop for CancellationOwnedChild {
        fn drop(&mut self) {
            let _ = self.0.kill();
            let _ = self.0.wait();
        }
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn cancellation_ownership_termination_and_history_finish_after_waiter_abort() {
        let root = tempfile::tempdir().unwrap().keep();
        let ready = root.join("hook-ready");
        let child = CancellationOwnedChild(
            std::process::Command::new("/bin/sh")
                .args([
                    "-c",
                    "trap '' TERM; printf ready > \"$1\"; exec /bin/sleep 60",
                    "owned-cancel-hook",
                ])
                .arg(&ready)
                .spawn()
                .unwrap(),
        );
        tokio::time::timeout(Duration::from_secs(3), async {
            while std::fs::read(&ready).ok().as_deref() != Some(b"ready") {
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        })
        .await
        .unwrap();
        let history = Arc::new(BuildHistory::new(100));
        let build = history.start_active_build(
            "owned-hook".to_owned(),
            "no-remote-work".to_owned(),
            "fixture".to_owned(),
            child.0.id(),
            0,
            rch_common::BuildLocation::Remote,
        );
        let ctx = make_test_context(WorkerPool::new(), history.clone());
        let orch = CancellationOrchestrator::new(
            CancellationConfig {
                kill_timeout: Duration::from_secs(1),
                ..test_config()
            },
            test_events(),
        );
        let caller_owner = orch.clone();
        let caller = tokio::spawn(async move {
            caller_owner
                .cancel_build(&ctx, build.id, CancelReason::User, false)
                .await
        });
        wait_for_cancellation_attempts(&orch, 1).await;
        caller.abort();
        assert!(caller.await.unwrap_err().is_cancelled());
        wait_for_cancellation_attempts(&orch, 0).await;
        assert!(
            !is_process_alive(child.0.id()),
            "caller abort abandoned the owned hook"
        );
        assert!(history.active_build(build.id).is_none());
        let recent = history.recent(10);
        assert_eq!(recent.len(), 1);
        assert_eq!(recent[0].id, build.id);
        assert_eq!(
            recent[0].cancellation.as_ref().unwrap().final_state,
            "completed"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn cancellation_ownership_completed_before_dispatch_does_not_signal_stale_hook() {
        use std::future::{Future, poll_fn};
        use std::task::Poll;

        let child = CancellationOwnedChild(
            std::process::Command::new("/bin/sleep")
                .arg("60")
                .spawn()
                .unwrap(),
        );
        let history = Arc::new(BuildHistory::new(100));
        let build = history.start_active_build(
            "completion-race".to_owned(),
            "no-remote-work".to_owned(),
            "fixture".to_owned(),
            child.0.id(),
            0,
            rch_common::BuildLocation::Remote,
        );
        let ctx = make_test_context(WorkerPool::new(), history.clone());
        let orch = CancellationOrchestrator::new(test_config(), test_events());
        let mut request = Box::pin(orch.cancel_build(&ctx, build.id, CancelReason::User, true));
        // Poll admission without yielding to the newly spawned owner task.
        poll_fn(|cx| {
            assert!(request.as_mut().poll(cx).is_pending());
            Poll::Ready(())
        })
        .await;
        assert!(
            history
                .finish_active_build(build.id, 0, None, None, None)
                .is_some()
        );
        let response = request.await;
        assert_eq!(response.status, "error");
        assert_eq!(response.slots_released, 0);
        assert!(
            is_process_alive(child.0.id()),
            "signalled a completed build's stale hook"
        );
        assert!(orch.active_cancellations().await.is_empty());
        assert!(
            history
                .recent(10)
                .iter()
                .all(|record| record.cancellation.is_none())
        );
    }
}