rch 2.1.25

//! Same-identity job observation and recovery. No command execution/replay path.
use crate::ui::context::OutputContext;
use anyhow::{Context, Result};
use clap::Subcommand;
use rch_common::job_identity::{DurableJobLease, default_job_lease_directory};
use rch_common::process_identity::OwnerPresence;
use serde_json::{Value, json};
#[cfg(unix)]
use std::time::{Duration, Instant};

#[derive(Debug, Subcommand)]
pub enum JobsAction {
    /// Follow the original wrapper/result, without starting another build
    Attach {
        wrapper_id: String,
        #[arg(long, default_value_t = 300)]
        timeout_secs: u64,
    },
    /// Cancel the exact daemon build and cooperatively stop its wrapper
    Cancel { wrapper_id: String },
    /// Reconcile completion and retrieve outstanding outputs; never replay
    Recover {
        wrapper_id: String,
        #[arg(long, default_value_t = 300)]
        timeout_secs: u64,
        /// Emit the original opt-in invocation's retained Cargo JSON on stdout
        #[arg(long)]
        cargo_json: bool,
    },
}

pub async fn run(action: Option<JobsAction>, ctx: &OutputContext) -> Result<()> {
    #[cfg(unix)]
    {
        run_unix(action, ctx).await
    }
    #[cfg(not(unix))]
    {
        let _ = (action, ctx);
        anyhow::bail!("durable jobs require Unix daemon IPC")
    }
}

fn owner_presence(lease: &DurableJobLease) -> OwnerPresence {
    rch_common::process_identity::owner_presence(lease.wrapper_pid, lease.owner_identity().as_ref())
}

pub(crate) fn process_matches(lease: &DurableJobLease) -> bool {
    owner_presence(lease) == OwnerPresence::Live
}

#[cfg(unix)]
pub(crate) async fn query(lease: &DurableJobLease) -> Result<Value> {
    let id = lease
        .identity
        .remote_build_id
        .context("job was never admitted; no remote execution to attach")?;
    let response = super::send_daemon_command(&format!(
        "GET /builds/{id}?local_wrapper_id={}\n",
        lease.identity.local_wrapper_id
    ))
    .await?;
    parse_job_status(lease, &response)
}

#[cfg(unix)]
fn parse_job_status(lease: &DurableJobLease, response: &str) -> Result<Value> {
    let (header, body) = response
        .split_once("\r\n\r\n")
        .or_else(|| response.split_once("\n\n"))
        .context("daemon response has no body")?;
    // An error envelope is not authoritative absence or completion. In
    // particular, never use a 5xx `not_found` body to settle ownership.
    let mut status = header.lines().next().unwrap_or_default().split_whitespace();
    anyhow::ensure!(
        matches!(status.next(), Some("HTTP/1.0" | "HTTP/1.1")) && status.next() == Some("200"),
        "daemon refused job status; ownership remains unresolved"
    );
    let payload: Value = serde_json::from_str(body)?;
    match payload.get("status").and_then(Value::as_str) {
        Some("active") => validate_identity(lease, &payload["active"])?,
        Some("completed") => {
            validate_completion(lease, &payload)?;
        }
        Some("not_found") => {}
        Some("identity_mismatch") => {
            anyhow::bail!("daemon build identity mismatch; no action taken")
        }
        _ => anyhow::bail!("daemon did not provide authoritative job status: {payload}"),
    }
    Ok(payload)
}

fn validate_identity(lease: &DurableJobLease, record: &Value) -> Result<()> {
    anyhow::ensure!(
        record.get("local_wrapper_id").and_then(Value::as_str)
            == Some(lease.identity.local_wrapper_id.as_str()),
        "daemon wrapper identity mismatch"
    );
    anyhow::ensure!(
        record.get("worker_id").and_then(Value::as_str) == lease.worker_id.as_deref(),
        "daemon worker identity mismatch"
    );
    anyhow::ensure!(
        record["id"].as_u64() == lease.identity.remote_build_id,
        "daemon build identity mismatch"
    );
    Ok(())
}

/// Completion may win before cancellation or while its request is in flight.
/// Accept only a complete receipt for the original wrapper/build/worker tuple;
/// a status string alone is not authority to stop a wrapper or retire its work.
pub(crate) fn validate_completion(lease: &DurableJobLease, reply: &Value) -> Result<i32> {
    anyhow::ensure!(
        reply["status"] == "completed"
            && reply["local_wrapper_id"].as_str() == Some(lease.identity.local_wrapper_id.as_str()),
        "daemon completion identity mismatch"
    );
    let id = lease
        .identity
        .remote_build_id
        .filter(|id| *id > 0)
        .context("completion requires an admitted build identity")?;
    let worker = lease
        .worker_id
        .as_deref()
        .filter(|worker| !worker.is_empty())
        .context("completion requires an admitted worker identity")?;
    anyhow::ensure!(
        reply["record"]["id"].as_u64() == Some(id)
            && reply["record"]["worker_id"].as_str() == Some(worker),
        "daemon completion build/worker mismatch"
    );
    reply["record"]["exit_code"]
        .as_i64()
        .and_then(|code| i32::try_from(code).ok())
        .context("completion has no valid exit code")
}

#[derive(Debug, PartialEq, Eq)]
enum AdmittedCancellation {
    StopWrapper,
    AlreadyCompleted(i32),
}

fn validate_admitted_cancellation(
    lease: &DurableJobLease,
    reply: &Value,
) -> Result<AdmittedCancellation> {
    if reply["status"] == "completed" {
        return validate_completion(lease, reply).map(AdmittedCancellation::AlreadyCompleted);
    }
    anyhow::ensure!(
        reply["status"] == "cancelled",
        "cancellation was not acknowledged: {reply}"
    );
    let id = lease
        .identity
        .remote_build_id
        .filter(|id| *id > 0)
        .context("cancellation requires an admitted build identity")?;
    let worker = lease
        .worker_id
        .as_deref()
        .filter(|worker| !worker.is_empty())
        .context("cancellation requires an admitted worker identity")?;
    anyhow::ensure!(
        reply["build_id"].as_u64() == Some(id) && reply["worker_id"].as_str() == Some(worker),
        "daemon cancellation build/worker mismatch; no wrapper stop requested"
    );
    // The build endpoint fences the wrapper in its request and does not
    // currently echo it. If a peer does echo it, disagreement is fatal.
    if let Some(wrapper) = reply.get("local_wrapper_id") {
        anyhow::ensure!(
            wrapper.as_str() == Some(lease.identity.local_wrapper_id.as_str()),
            "daemon cancellation wrapper identity mismatch"
        );
    }
    Ok(AdmittedCancellation::StopWrapper)
}

/// Once the original owner is positively absent, reload before reconciling:
/// it may have persisted a recovery intent or final acknowledgement while the
/// daemon status request was in flight. Never erase that newer evidence.
fn same_unfinished_owner_without_recovery(
    observed: &DurableJobLease,
    latest: &DurableJobLease,
) -> bool {
    observed.identity == latest.identity
        && observed.worker_id == latest.worker_id
        && observed.wrapper_pid == latest.wrapper_pid
        && observed.process_start_ticks == latest.process_start_ticks
        && observed.boot_id == latest.boot_id
        && observed.process_birth == latest.process_birth
        && latest.recovery.is_none()
        && !latest.terminal_acknowledged
}

fn completed_cancellation_report(lease: &DurableJobLease, code: i32) -> Value {
    // Remote completion is not evidence that local artifact retrieval or
    // terminal acknowledgement finished. Leave the owner's journal intact.
    json!({
        "status": "completed",
        "identity": lease.identity,
        "exit_code": code,
        "terminal_acknowledged": lease.terminal_acknowledged,
        "has_recovery_journal": lease.recovery.is_some(),
        "wrapper_stop_requested": false,
    })
}

fn validate_queued_cancellation(wrapper: &str, reply: &Value) -> Result<()> {
    anyhow::ensure!(
        reply["local_wrapper_id"].as_str() == Some(wrapper),
        "daemon cancellation wrapper identity mismatch"
    );
    match reply["status"].as_str() {
        Some("cancelled_before_start") => anyhow::ensure!(
            reply["exit_code"].as_i64() == Some(130) && reply["build_id"].is_null(),
            "invalid cancellation-before-start receipt"
        ),
        Some("cancelled") => anyhow::ensure!(
            reply["build_id"].as_u64().is_some_and(|id| id > 0),
            "admitted cancellation has no build identity"
        ),
        Some("completed") => anyhow::ensure!(
            reply["record"]["id"].as_u64().is_some_and(|id| id > 0),
            "completion has no build identity"
        ),
        _ => anyhow::bail!("cancellation was not acknowledged: {reply}"),
    }
    Ok(())
}

/// Output failures must not produce a success status or a second stdout document.
#[derive(Debug, thiserror::Error)]
#[error("job output could not be delivered")]
pub(crate) struct JobOutputFailure(#[source] crate::ui::context::OutputError);

impl From<std::io::Error> for JobOutputFailure {
    fn from(error: std::io::Error) -> Self {
        Self(crate::ui::context::OutputError::Delivery(error))
    }
}

fn emit(ctx: &OutputContext, payload: &Value) -> Result<()> {
    let result = if ctx.is_json() {
        ctx.json(payload)
    } else {
        ctx.json_force(payload)
    };
    result.map_err(JobOutputFailure)?;
    Ok(())
}

/// A durable identity exists before the daemon admits the build. A live
/// owner in that state is still waiting, not a failed or missing remote job.
/// In particular, attach/recover must not query a nonexistent build id or
/// submit another command while admission is in progress.
fn waiting_for_admission(lease: &DurableJobLease, owner_alive: bool) -> Result<bool> {
    if lease.identity.remote_build_id.is_some() || lease.terminal_acknowledged {
        return Ok(false);
    }
    anyhow::ensure!(
        owner_alive,
        "original wrapper is absent or unverified before daemon admission; outcome remains uncertain; no command replayed"
    );
    Ok(true)
}

/// Bound the whole observation/recovery operation, not each IPC request
/// independently. An expired deadline must not even poll a mutating future.
#[cfg(unix)]
async fn within_job_deadline<T>(
    deadline: Instant,
    operation: &str,
    future: impl std::future::Future<Output = Result<T>>,
) -> Result<T> {
    anyhow::ensure!(
        Instant::now() < deadline,
        "same-id job deadline elapsed before {operation}; journal retained; no command replayed"
    );
    tokio::time::timeout_at(tokio::time::Instant::from_std(deadline), future)
        .await
        .with_context(|| {
            format!(
                "same-id job deadline elapsed during {operation}; journal retained; no command replayed"
            )
        })?
}

#[cfg(unix)]
async fn wait_for_job_poll(deadline: Instant) -> Result<()> {
    anyhow::ensure!(
        Instant::now() < deadline,
        "same-id job is still pending; no command replayed (use jobs cancel or recover)"
    );
    // Never oversleep a short remaining budget. The next iteration can still
    // observe an already-persisted terminal acknowledgement at the boundary.
    let wake = (Instant::now() + Duration::from_secs(1)).min(deadline);
    tokio::time::sleep_until(tokio::time::Instant::from_std(wake)).await;
    Ok(())
}

#[cfg(unix)]
fn read_job_journal_at(
    directory: std::os::fd::BorrowedFd<'_>,
    name: &std::ffi::CStr,
) -> Result<DurableJobLease> {
    let bytes = crate::state::primitives::read_regular_file_at(directory, name)?;
    Ok(serde_json::from_slice(&bytes)?)
}

#[cfg(unix)]
fn job_journal_error(path: &std::path::Path, error: String) -> Value {
    use std::os::unix::ffi::OsStrExt;

    // Path's Serialize implementation rejects non-UTF-8, and json! unwraps
    // that error. Keep a printable diagnostic and the exact Unix path bytes.
    let mut report = json!({"path": path.to_string_lossy(), "error": error});
    if path.to_str().is_none() {
        report["path_bytes"] = json!(path.as_os_str().as_bytes());
    }
    report
}

#[cfg(unix)]
fn read_job_listing(directory: &std::path::Path) -> Result<Value> {
    use rustix::fs::{Dir, Mode, OFlags};
    use std::os::unix::ffi::OsStrExt;

    let mut jobs = Vec::new();
    let mut journal_errors = Vec::new();
    let descriptor = match rustix::fs::open(
        directory,
        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW | OFlags::NONBLOCK,
        Mode::empty(),
    ) {
        Ok(descriptor) => Some(descriptor),
        Err(rustix::io::Errno::NOENT) => None,
        Err(error) => return Err(error.into()),
    };
    if let Some(descriptor) = descriptor {
        let mut entries = Dir::new(descriptor)?;
        // Use Dir's read API so the loop can borrow its descriptor between
        // entries without a for-loop iterator retaining a mutable borrow.
        while let Some(entry) = entries.read() {
            let entry = match entry {
                Ok(entry) => entry,
                Err(error) => {
                    journal_errors.push(job_journal_error(directory, error.to_string()));
                    continue;
                }
            };
            let name = std::ffi::OsStr::from_bytes(entry.file_name().to_bytes());
            let path = directory.join(name);
            if path.extension().and_then(|value| value.to_str()) != Some("json") {
                continue;
            }
            let observed = (|| -> Result<DurableJobLease> {
                read_job_journal_at(entries.fd()?, entry.file_name())
            })();
            match observed {
                Ok(lease) => {
                    let presence = owner_presence(&lease);
                    let status = match presence {
                        OwnerPresence::Live => "live",
                        OwnerPresence::Absent => "absent",
                        OwnerPresence::Unknown => "unknown",
                    };
                    jobs.push(json!({
                        "lease": lease,
                        "wrapper_alive": presence == OwnerPresence::Live,
                        "owner_presence": status,
                    }));
                }
                Err(error) => {
                    journal_errors.push(job_journal_error(&path, format!("{error:#}")));
                }
            }
        }
    }
    Ok(json!({
        "jobs": jobs,
        "complete": journal_errors.is_empty(),
        "journal_errors": journal_errors,
    }))
}

#[cfg(unix)]
async fn run_unix(action: Option<JobsAction>, ctx: &OutputContext) -> Result<()> {
    use crate::hook::DurableLeaseWriter;
    let Some(action) = action else {
        let listing = read_job_listing(&default_job_lease_directory())?;
        emit(ctx, &listing)?;
        if listing["complete"] != true {
            use std::io::Write;
            // The error document is already delivered and this branch always
            // exits 1. A closed diagnostic stream must not turn it into a panic.
            let _ = writeln!(
                std::io::stderr().lock(),
                "[RCH] job listing is incomplete; inspect journal_errors before reconciling ownership"
            );
            // The structured listing was already emitted. Returning Err
            // would append a second JSON error envelope in async_main.
            std::process::exit(1);
        }
        return Ok(());
    };
    let (wrapper_id, recover, cancel, timeout_secs, cargo_json) = match action {
        JobsAction::Attach {
            wrapper_id,
            timeout_secs,
        } => (wrapper_id, false, false, timeout_secs, false),
        JobsAction::Recover {
            wrapper_id,
            timeout_secs,
            cargo_json,
        } => (wrapper_id, true, false, timeout_secs, cargo_json),
        JobsAction::Cancel { wrapper_id } => (wrapper_id, false, true, 30, false),
    };
    let deadline = Instant::now() + Duration::from_secs(timeout_secs.min(86400));
    if cargo_json {
        anyhow::ensure!(
            !ctx.is_json(),
            "Cargo JSON recovery conflicts with machine envelopes"
        );
        let writer = DurableLeaseWriter::load(&wrapper_id)?;
        let (code, stdout, stderr) = within_job_deadline(
            deadline,
            "Cargo JSON recovery (use the retained journal for same-id retry)",
            crate::hook::recover_job_cargo_json(&writer),
        )
        .await?;
        {
            use std::io::Write;
            let mut diagnostics = std::io::stderr().lock();
            diagnostics
                .write_all(&stderr)
                .map_err(JobOutputFailure::from)?;
            diagnostics.flush().map_err(JobOutputFailure::from)?;
        }
        if let Some(bytes) = stdout {
            use std::io::Write;
            let mut output = std::io::stdout().lock();
            output.write_all(&bytes).map_err(JobOutputFailure::from)?;
            output.flush().map_err(JobOutputFailure::from)?;
        } else {
            use std::io::Write;
            let mut diagnostics = std::io::stderr().lock();
            writeln!(
                diagnostics,
                "[RCH] recovered invocation has no deliverable Cargo stdout (exit {code})"
            )
            .map_err(JobOutputFailure::from)?;
            diagnostics.flush().map_err(JobOutputFailure::from)?;
        }
        std::process::exit(code);
    }
    loop {
        let writer = DurableLeaseWriter::load(&wrapper_id)?;
        let lease = writer.snapshot();
        if lease.terminal_acknowledged {
            // The daemon refused before admission: no remote command ran, so
            // there is no remote exit to report or recover.
            let report = if lease.state
                == rch_common::job_identity::JobLifecycleState::FailedBeforeAdmission
            {
                json!({
                    "status": "failed_before_admission",
                    "identity": lease.identity,
                    "exit_code": null,
                    "terminal_acknowledged": true,
                    "admission": null,
                })
            } else {
                writer.terminal_job_report(&wrapper_id, "completed", None)?
            };
            emit(ctx, &report)?;
            return Ok(());
        }
        if !cancel && waiting_for_admission(&lease, process_matches(&lease))? {
            wait_for_job_poll(deadline).await?;
            continue;
        }
        if cancel && lease.identity.remote_build_id.is_none() {
            // Admission may win after this snapshot. The daemon serializes the
            // decision with registration and cancels that exact active build.
            let response = within_job_deadline(
                deadline,
                "queued cancellation",
                super::send_daemon_command(&format!("POST /jobs/{wrapper_id}/cancel\n")),
            )
            .await?;
            let body = response
                .split_once("\r\n\r\n")
                .or_else(|| response.split_once("\n\n"))
                .map(|(_, body)| body)
                .context("missing queued cancellation response")?;
            let mut reply: Value = serde_json::from_str(body)?;
            validate_queued_cancellation(&wrapper_id, &reply)?;
            if reply["status"] == "cancelled_before_start" {
                // Daemon cancellation already succeeded. A live wrapper owns
                // its journal and acknowledges the selection response itself.
                // Report deferred acknowledgment without overwriting that
                // owner or misreporting its cancellation as a daemon failure.
                reply["journal_acknowledged"] = json!(writer.confirm_selection_cancelled()?);
            } else if reply["status"] == "cancelled" {
                let mut identity = lease.identity.clone();
                identity.admit(
                    reply["build_id"]
                        .as_u64()
                        .context("missing cancelled build id")?,
                );
                let path = default_job_lease_directory().join(format!("{wrapper_id}.cancel"));
                crate::state::primitives::atomic_write(&path, &serde_json::to_vec(&identity)?)?;
            }
            emit(ctx, &reply)?;
            return Ok(());
        }
        let status = within_job_deadline(deadline, "job status", query(&lease)).await?;
        if cancel {
            if status["status"] == "completed" {
                let code = validate_completion(&lease, &status)?;
                emit(ctx, &completed_cancellation_report(&lease, code))?;
                return Ok(());
            }
            let id = lease
                .identity
                .remote_build_id
                .context("job has no admitted build")?;
            anyhow::ensure!(
                status["status"] == "active",
                "daemon has no identity evidence for cancellation (status: {}); no process was signalled",
                status["status"]
            );
            let response = within_job_deadline(
                deadline,
                "admitted cancellation",
                super::send_daemon_command(&format!(
                    "POST /builds/{id}/cancel?local_wrapper_id={}\n",
                    wrapper_id
                )),
            )
            .await?;
            let body = response
                .split_once("\r\n\r\n")
                .or_else(|| response.split_once("\n\n"))
                .map(|(_, b)| b)
                .context("missing cancellation response")?;
            let reply: Value = serde_json::from_str(body)?;
            if let AdmittedCancellation::AlreadyCompleted(code) =
                validate_admitted_cancellation(&lease, &reply)?
            {
                emit(ctx, &completed_cancellation_report(&lease, code))?;
                return Ok(());
            }
            // The original wrapper consumes this exact-identity receipt itself.
            // Never signal a PID obtained from a persisted lease.
            let path = default_job_lease_directory().join(format!("{wrapper_id}.cancel"));
            crate::state::primitives::atomic_write(&path, &serde_json::to_vec(&lease.identity)?)?;
            emit(
                ctx,
                &json!({"status":"cancelled", "identity":lease.identity, "wrapper_stop_requested":true}),
            )?;
            return Ok(());
        }
        if recover && process_matches(&lease) && lease.recovery.is_some() {
            // The live wrapper's retrieval select consumes this exact-identity
            // receipt and re-drives its interrupted retrieval itself.
            let receipt = default_job_lease_directory().join(format!("{wrapper_id}.recover"));
            if !receipt.exists() {
                crate::state::primitives::atomic_write(
                    &receipt,
                    &serde_json::to_vec(&lease.identity)?,
                )?;
                emit(
                    ctx,
                    &json!({"status":"recover_requested", "identity":lease.identity}),
                )?;
                return Ok(());
            }
        }
        if !process_matches(&lease) && recover && lease.recovery.is_some() {
            let (code, recovered_writer) = within_job_deadline(
                deadline,
                "artifact recovery (use the retained journal for same-id retry)",
                crate::hook::recover_job(&writer),
            )
            .await?;
            emit(
                ctx,
                &recovered_writer.terminal_job_report(&wrapper_id, "recovered", Some(code))?,
            )?;
            return Ok(());
        }
        if status["status"] == "completed"
            && lease.recovery.is_none()
            && owner_presence(&lease) == OwnerPresence::Absent
        {
            let latest_writer = within_job_deadline(
                deadline,
                "terminal journal ownership",
                writer.claim_recovery(),
            )
            .await?;
            let latest = latest_writer.snapshot();
            if !same_unfinished_owner_without_recovery(&lease, &latest) {
                wait_for_job_poll(deadline).await?;
                continue;
            }
            let code = validate_completion(&latest, &status)?;
            anyhow::ensure!(
                latest.exit_code.is_none_or(|observed| observed == code),
                "local and daemon completion disagree; no journal rewritten; no command replayed"
            );
            latest_writer.record_exit(code)?;
            latest_writer.acknowledge_terminal()?;
            emit(
                ctx,
                &latest_writer.terminal_job_report(&wrapper_id, "completed", Some(code))?,
            )?;
            return Ok(());
        }
        if !process_matches(&lease) {
            anyhow::bail!(
                "original wrapper is absent or its identity cannot be verified; use jobs recover with retained retrieval evidence (daemon status: {}); command will not be replayed",
                status["status"]
            );
        }
        wait_for_job_poll(deadline).await?;
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    #[cfg(unix)]
    use std::os::fd::AsFd;

    #[cfg(unix)]
    #[test]
    fn job_status_accepts_successful_http_envelopes() {
        let lease = queued_lease();
        for protocol in ["HTTP/1.0", "HTTP/1.1"] {
            for separator in ["\r\n", "\n"] {
                let response = format!(
                    "{protocol} 200 OK{separator}Content-Type: application/json{separator}{separator}{{\"status\":\"not_found\"}}"
                );
                assert_eq!(
                    parse_job_status(&lease, &response).unwrap()["status"],
                    "not_found"
                );
            }
        }
    }

    #[cfg(unix)]
    #[test]
    fn job_status_refuses_error_envelopes_before_trusting_json() {
        let lease = queued_lease();
        let original = lease.clone();
        for header in [
            "HTTP/1.1 500 Internal Server Error",
            "HTTP/1.1 503 Service Unavailable",
            "HTTP/1.1 404 Not Found",
            "HTTP/1.1 302 Found",
            "HTTP/1.1 2000 Invalid Status",
            "HTTP/2 200 OK",
            "200 OK",
            "",
        ] {
            for body in ["{\"status\":\"not_found\"}", "{\"status\":\"completed\"}"] {
                let response = format!("{header}\r\n\r\n{body}");
                let error = parse_job_status(&lease, &response).unwrap_err();
                assert!(format!("{error:#}").contains("ownership remains unresolved"));
                assert_eq!(lease, original);
            }
        }
        assert!(parse_job_status(&lease, "HTTP/1.1 200 OK").is_err());
        assert!(parse_job_status(&lease, "HTTP/1.1 200 OK\r\n\r\ninvalid JSON").is_err());
    }

    #[cfg(unix)]
    #[test]
    fn job_listing_preserves_valid_jobs_and_reports_corrupt_journals() {
        let retained = tempfile::tempdir().unwrap().keep();
        let lease = DurableJobLease::new(
            rch_common::job_identity::JobIdentity::new_local(),
            1,
            None,
            None,
            0,
            true,
            false,
            "hash".into(),
        );
        let valid = retained.join("valid.json");
        let valid_bytes = serde_json::to_vec(&lease).unwrap();
        std::fs::write(&valid, &valid_bytes).unwrap();
        let corrupt = retained.join("corrupt.json");
        let corrupt_bytes = b"{\"identity\":";
        std::fs::write(&corrupt, corrupt_bytes).unwrap();
        let pending = retained.join("not-a-journal.pending");
        std::fs::write(&pending, b"pending evidence").unwrap();

        let listing = read_job_listing(&retained).unwrap();
        assert_eq!(listing["complete"], false);
        assert_eq!(listing["jobs"].as_array().unwrap().len(), 1);
        assert_eq!(
            listing["jobs"][0]["lease"],
            serde_json::to_value(&lease).unwrap()
        );
        assert_eq!(listing["jobs"][0]["owner_presence"], "unknown");
        assert_eq!(listing["jobs"][0]["wrapper_alive"], false);
        assert_eq!(listing["journal_errors"].as_array().unwrap().len(), 1);
        assert_eq!(listing["journal_errors"][0]["path"], json!(corrupt));
        assert!(
            listing["journal_errors"][0]["error"]
                .as_str()
                .unwrap()
                .contains("EOF")
        );
        assert_eq!(std::fs::read(valid).unwrap(), valid_bytes);
        assert_eq!(std::fs::read(corrupt).unwrap(), corrupt_bytes);
        assert_eq!(std::fs::read(pending).unwrap(), b"pending evidence");
    }

    #[cfg(unix)]
    #[test]
    fn job_listing_preserves_jobs_and_exact_non_utf8_error_paths() {
        use std::os::unix::ffi::OsStrExt;

        let retained = tempfile::tempdir().unwrap().keep();
        let directory = retained.join(std::ffi::OsStr::from_bytes(b"journals-\xfe"));
        match std::fs::create_dir(&directory) {
            // APFS and other UTF-8-only filesystems cannot hold such a path.
            Err(error) if error.raw_os_error() == Some(rustix::io::Errno::ILSEQ.raw_os_error()) => {
                return;
            }
            result => result.unwrap(),
        }
        let lease = queued_lease();
        let valid = directory.join(std::ffi::OsStr::from_bytes(b"valid-\xfe.json"));
        let valid_bytes = serde_json::to_vec(&lease).unwrap();
        std::fs::write(&valid, &valid_bytes).unwrap();
        let corrupt = directory.join(std::ffi::OsStr::from_bytes(b"corrupt-\xff.json"));
        let corrupt_bytes = b"{\"identity\":";
        std::fs::write(&corrupt, corrupt_bytes).unwrap();

        let listing = read_job_listing(&directory).unwrap();
        assert_eq!(listing["complete"], false);
        assert_eq!(listing["jobs"].as_array().unwrap().len(), 1);
        assert_eq!(
            listing["jobs"][0]["lease"],
            serde_json::to_value(lease).unwrap()
        );
        assert_eq!(listing["jobs"][0]["owner_presence"], "unknown");
        assert_eq!(listing["jobs"][0]["wrapper_alive"], false);
        assert_eq!(listing["journal_errors"].as_array().unwrap().len(), 1);
        let error = &listing["journal_errors"][0];
        assert_eq!(error["path"], json!(corrupt.to_string_lossy()));
        assert_eq!(error["path_bytes"], json!(corrupt.as_os_str().as_bytes()));
        assert!(error["error"].as_str().unwrap().contains("EOF"));
        let encoded = serde_json::to_vec(&listing).unwrap();
        assert_eq!(serde_json::from_slice::<Value>(&encoded).unwrap(), listing);
        assert_eq!(std::fs::read(valid).unwrap(), valid_bytes);
        assert_eq!(std::fs::read(corrupt).unwrap(), corrupt_bytes);
    }

    #[test]
    fn job_output_preserves_json_toon_and_plain_formats_in_real_files() {
        use crate::ui::context::{ColorChoice, OutputConfig, OutputFormat, OutputMode};
        use crate::ui::writer::OutputWriter;

        let retained = tempfile::tempdir().unwrap().keep();
        let payload = json!({"status": "observed", "owner_presence": "unknown"});
        for mode in [OutputMode::Json, OutputMode::Plain] {
            for format in [OutputFormat::Json, OutputFormat::Toon] {
                let path = retained.join(format!("{mode:?}-{format:?}.output"));
                let stdout = OutputWriter::new(std::fs::File::create_new(&path).unwrap(), false);
                let ctx = OutputContext::with_writers(
                    OutputConfig {
                        force_mode: Some(mode),
                        color: ColorChoice::Never,
                        format,
                        ..Default::default()
                    },
                    stdout,
                    OutputWriter::stderr(),
                );
                emit(&ctx, &payload).unwrap();
                let bytes = std::fs::read(path).unwrap();
                assert!(bytes.ends_with(b"\n"));
                let observed = if mode == OutputMode::Json && format == OutputFormat::Toon {
                    let text = std::str::from_utf8(&bytes).unwrap();
                    assert!(!text.trim_start().starts_with('{'));
                    serde_json::from_str::<Value>(&toon_rust::toon_to_json(text).unwrap()).unwrap()
                } else {
                    serde_json::from_slice::<Value>(&bytes).unwrap()
                };
                assert_eq!(observed, payload);
            }
        }
    }

    #[cfg(unix)]
    #[test]
    fn job_output_reports_real_closed_socket_write_and_flush_failures() {
        use crate::ui::context::{ColorChoice, OutputConfig, OutputFormat, OutputMode};
        use crate::ui::writer::OutputWriter;
        use std::os::unix::net::UnixStream;

        for mode in [OutputMode::Json, OutputMode::Plain] {
            for format in [OutputFormat::Json, OutputFormat::Toon] {
                for buffered in [false, true] {
                    let (socket, peer) = UnixStream::pair().unwrap();
                    socket
                        .set_write_timeout(Some(Duration::from_secs(2)))
                        .unwrap();
                    drop(peer);
                    let stdout = if buffered {
                        // The small payload fits in the buffer; the real socket
                        // failure must be observed when the writer flushes.
                        OutputWriter::new(std::io::BufWriter::with_capacity(8192, socket), false)
                    } else {
                        OutputWriter::new(socket, false)
                    };
                    let ctx = OutputContext::with_writers(
                        OutputConfig {
                            force_mode: Some(mode),
                            color: ColorChoice::Never,
                            format,
                            ..Default::default()
                        },
                        stdout,
                        OutputWriter::stderr(),
                    );
                    let error = emit(&ctx, &json!({"status": "observed"})).unwrap_err();
                    let failure = error.downcast_ref::<JobOutputFailure>().unwrap();
                    let crate::ui::context::OutputError::Delivery(cause) = &failure.0 else {
                        panic!("closed socket must report a delivery error: {error:#}");
                    };
                    assert_eq!(cause.kind(), std::io::ErrorKind::BrokenPipe);
                    assert!(format!("{error:#}").contains("job output could not be delivered"));
                }
            }
        }
    }

    #[cfg(unix)]
    #[test]
    fn job_listing_rejects_nonregular_journals_and_preserves_them() {
        let retained = tempfile::tempdir().unwrap().keep();
        let target = retained.join("retained-evidence");
        let target_bytes = serde_json::to_vec(&queued_lease()).unwrap();
        std::fs::write(&target, &target_bytes).unwrap();
        let link = retained.join("symlink.json");
        std::os::unix::fs::symlink(&target, &link).unwrap();
        let directory = retained.join("directory.json");
        std::fs::create_dir(&directory).unwrap();

        let listing = read_job_listing(&retained).unwrap();
        assert_eq!(listing["complete"], false);
        assert!(listing["jobs"].as_array().unwrap().is_empty());
        let errors = listing["journal_errors"].as_array().unwrap();
        assert_eq!(errors.len(), 2);
        for (path, reason) in [
            (&link, "journal open without following symlinks failed"),
            (&directory, "journal is not a regular file"),
        ] {
            assert!(errors.iter().any(|error| {
                error["path"] == json!(path) && error["error"].as_str().unwrap().contains(reason)
            }));
        }
        assert!(
            std::fs::symlink_metadata(link)
                .unwrap()
                .file_type()
                .is_symlink()
        );
        assert!(directory.is_dir());
        assert_eq!(std::fs::read(target).unwrap(), target_bytes);
    }

    #[cfg(unix)]
    #[test]
    fn job_listing_reads_all_regular_journals_with_spaces_and_unicode() {
        let retained = tempfile::tempdir().unwrap().keep();
        let leases = [queued_lease(), queued_lease()];
        let names = ["first journal.json", "snowman-\u{2603}.json"];
        for (name, lease) in names.iter().zip(&leases) {
            std::fs::write(retained.join(name), serde_json::to_vec(lease).unwrap()).unwrap();
        }

        let listing = read_job_listing(&retained).unwrap();
        assert_eq!(listing["complete"], true);
        assert_eq!(listing["journal_errors"], json!([]));
        let jobs = listing["jobs"].as_array().unwrap();
        assert_eq!(jobs.len(), leases.len());
        for job in jobs {
            assert_eq!(job["owner_presence"], "unknown");
            assert_eq!(job["wrapper_alive"], false);
        }
        let mut observed: Vec<_> = jobs.iter().map(|job| job["lease"].clone()).collect();
        let mut expected: Vec<_> = leases
            .iter()
            .map(|lease| serde_json::to_value(lease).unwrap())
            .collect();
        for records in [&mut observed, &mut expected] {
            records.sort_by_key(|record| {
                record["identity"]["local_wrapper_id"]
                    .as_str()
                    .unwrap()
                    .to_owned()
            });
        }
        assert_eq!(observed, expected);
        for (name, lease) in names.iter().zip(&leases) {
            assert_eq!(
                std::fs::read(retained.join(name)).unwrap(),
                serde_json::to_vec(lease).unwrap()
            );
        }
    }

    #[cfg(unix)]
    #[test]
    fn job_journal_read_rejects_symlink_replacement_after_regular_observation() {
        use rustix::fs::{Dir, Mode, OFlags};

        let retained = tempfile::tempdir().unwrap().keep();
        let journal = retained.join("observed.json");
        let original_bytes = serde_json::to_vec(&queued_lease()).unwrap();
        std::fs::write(&journal, &original_bytes).unwrap();
        let target = retained.join("other-lease");
        let target_bytes = serde_json::to_vec(&queued_lease()).unwrap();
        std::fs::write(&target, &target_bytes).unwrap();
        let entries = Dir::new(
            rustix::fs::open(
                &retained,
                OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
                Mode::empty(),
            )
            .unwrap(),
        )
        .unwrap();
        assert!(std::fs::symlink_metadata(&journal).unwrap().is_file());

        // Move the original into a fresh retained name; replace only the test's
        // now-vacant journal name. No input or symlink target is discarded.
        let original = retained.join("observed-original.retained");
        assert!(!original.exists());
        std::fs::rename(&journal, &original).unwrap();
        std::os::unix::fs::symlink(&target, &journal).unwrap();
        let error = read_job_journal_at(entries.fd().unwrap(), c"observed.json").unwrap_err();
        assert!(format!("{error:#}").contains("journal open without following symlinks failed"));
        assert_eq!(std::fs::read(original).unwrap(), original_bytes);
        assert_eq!(std::fs::read(target).unwrap(), target_bytes);
        assert!(
            std::fs::symlink_metadata(journal)
                .unwrap()
                .file_type()
                .is_symlink()
        );
    }

    #[cfg(unix)]
    #[test]
    fn job_journal_read_rejects_fifo_replacement_without_waiting_for_a_writer() {
        use rustix::fs::{Mode, OFlags};
        use std::os::unix::fs::FileTypeExt;

        let retained = tempfile::tempdir().unwrap().keep();
        let journal = retained.join("observed.json");
        let original_bytes = serde_json::to_vec(&queued_lease()).unwrap();
        std::fs::write(&journal, &original_bytes).unwrap();
        let descriptor = rustix::fs::open(
            &retained,
            OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::empty(),
        )
        .unwrap();
        assert!(std::fs::symlink_metadata(&journal).unwrap().is_file());
        let original = retained.join("observed-original.retained");
        assert!(!original.exists());
        std::fs::rename(&journal, &original).unwrap();
        #[cfg(not(target_vendor = "apple"))]
        rustix::fs::mkfifoat(
            descriptor.as_fd(),
            c"observed.json",
            Mode::RUSR | Mode::WUSR,
        )
        .unwrap();
        // rustix has no mkfifoat on Apple targets and this crate forbids
        // unsafe code, so make the FIFO by path; `descriptor` is open on that
        // same directory.
        #[cfg(target_vendor = "apple")]
        assert!(
            std::process::Command::new("/usr/bin/mkfifo")
                .args(["-m", "600"])
                .arg(&journal)
                .status()
                .unwrap()
                .success()
        );

        // No writer is started. Bound observation so a missing NONBLOCK flag
        // fails this case instead of hanging the complete test runner.
        let (sender, receiver) = std::sync::mpsc::channel();
        let reader = std::thread::spawn(move || {
            sender
                .send(read_job_journal_at(descriptor.as_fd(), c"observed.json"))
                .unwrap();
        });
        let result = receiver
            .recv_timeout(Duration::from_secs(2))
            .expect("journal read waited on a FIFO without a writer");
        reader.join().unwrap();
        assert!(format!("{:#}", result.unwrap_err()).contains("journal is not a regular file"));
        assert!(
            std::fs::symlink_metadata(journal)
                .unwrap()
                .file_type()
                .is_fifo()
        );
        assert_eq!(std::fs::read(original).unwrap(), original_bytes);
    }

    #[cfg(unix)]
    #[test]
    fn job_journal_read_keeps_the_open_directory_identity_after_path_replacement() {
        use rustix::fs::{Dir, Mode, OFlags};

        let retained = tempfile::tempdir().unwrap().keep();
        let directory = retained.join("journals");
        std::fs::create_dir(&directory).unwrap();
        let lease = queued_lease();
        let original_bytes = serde_json::to_vec(&lease).unwrap();
        std::fs::write(directory.join("observed.json"), &original_bytes).unwrap();
        let entries = Dir::new(
            rustix::fs::open(
                &directory,
                OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
                Mode::empty(),
            )
            .unwrap(),
        )
        .unwrap();
        let original = retained.join("journals-original.retained");
        assert!(!original.exists());
        std::fs::rename(&directory, &original).unwrap();
        std::fs::create_dir(&directory).unwrap();
        let replacement = queued_lease();
        assert_ne!(replacement.identity, lease.identity);
        let replacement_bytes = serde_json::to_vec(&replacement).unwrap();
        std::fs::write(directory.join("observed.json"), &replacement_bytes).unwrap();

        let observed = read_job_journal_at(entries.fd().unwrap(), c"observed.json").unwrap();
        assert_eq!(
            serde_json::to_value(observed).unwrap(),
            serde_json::to_value(lease).unwrap()
        );
        assert_eq!(
            std::fs::read(original.join("observed.json")).unwrap(),
            original_bytes
        );
        assert_eq!(
            std::fs::read(directory.join("observed.json")).unwrap(),
            replacement_bytes
        );
    }

    #[cfg(unix)]
    #[test]
    fn job_listing_refuses_a_symlink_root_and_preserves_its_journals() {
        let retained = tempfile::tempdir().unwrap().keep();
        let directory = retained.join("journals");
        std::fs::create_dir(&directory).unwrap();
        let journal = directory.join("valid.json");
        let bytes = serde_json::to_vec(&queued_lease()).unwrap();
        std::fs::write(&journal, &bytes).unwrap();
        let link = retained.join("journal-link");
        std::os::unix::fs::symlink(&directory, &link).unwrap();

        assert!(read_job_listing(&link).is_err());
        assert_eq!(std::fs::read(journal).unwrap(), bytes);
        assert!(
            std::fs::symlink_metadata(link)
                .unwrap()
                .file_type()
                .is_symlink()
        );
    }

    #[cfg(unix)]
    #[test]
    fn job_listing_of_absent_directory_is_complete_and_does_not_create_it() {
        let retained = tempfile::tempdir().unwrap().keep();
        let absent = retained.join("no-journals-yet");
        let listing = read_job_listing(&absent).unwrap();
        assert_eq!(
            listing,
            json!({"jobs": [], "complete": true, "journal_errors": []})
        );
        assert!(!absent.exists());
    }

    #[test]
    fn queued_cancellation_receipt_requires_exact_identity_and_terminal_evidence() {
        let valid = json!({"status":"cancelled_before_start", "local_wrapper_id":"wrapper", "exit_code":130});
        validate_queued_cancellation("wrapper", &valid).unwrap();
        assert!(validate_queued_cancellation("other", &valid).is_err());
        for reply in [
            json!({"status":"not_queued", "local_wrapper_id":"wrapper"}),
            json!({"status":"cancelled_before_start", "local_wrapper_id":"wrapper", "exit_code":0}),
            json!({"status":"cancelled_before_start", "local_wrapper_id":"wrapper", "exit_code":130, "build_id":42}),
            json!({"status":"cancelled", "local_wrapper_id":"wrapper"}),
        ] {
            assert!(validate_queued_cancellation("wrapper", &reply).is_err());
        }
        validate_queued_cancellation(
            "wrapper",
            &json!({"status":"cancelled", "local_wrapper_id":"wrapper", "build_id":42}),
        )
        .unwrap();
    }

    #[test]
    fn mismatched_identity_cannot_authorize_action() {
        let mut lease = DurableJobLease::new(
            rch_common::job_identity::JobIdentity::new_local(),
            1,
            None,
            None,
            0,
            true,
            false,
            "hash".into(),
        );
        lease.admit(1, "worker-a".into(), 0);
        assert!(
            validate_identity(
                &lease,
                &json!({"local_wrapper_id":"other", "worker_id":"worker-a"})
            )
            .is_err()
        );
        assert!(
            validate_identity(
                &lease,
                &json!({"local_wrapper_id":lease.identity.local_wrapper_id, "worker_id":"worker-b"})
            )
            .is_err()
        );
        assert!(!process_matches(&lease));
    }

    fn queued_lease() -> DurableJobLease {
        DurableJobLease::new(
            rch_common::job_identity::JobIdentity::new_local(),
            1,
            None,
            None,
            0,
            true,
            false,
            "hash".into(),
        )
    }

    #[test]
    fn live_pre_admission_job_is_followed_until_admitted() {
        let mut lease = queued_lease();
        let wrapper = lease.identity.local_wrapper_id.clone();
        assert!(waiting_for_admission(&lease, true).unwrap());
        lease.admit(42, "worker-a".into(), 1);
        assert!(!waiting_for_admission(&lease, true).unwrap());
        assert_eq!(lease.identity.local_wrapper_id, wrapper);
        assert_eq!(lease.identity.remote_build_id, Some(42));
    }

    #[test]
    fn absent_pre_admission_owner_is_uncertain_not_replayed() {
        let error = waiting_for_admission(&queued_lease(), false).unwrap_err();
        assert!(error.to_string().contains("outcome remains uncertain"));
    }

    #[test]
    fn terminal_pre_admission_job_does_not_need_a_live_owner() {
        let mut lease = queued_lease();
        lease.acknowledge_terminal(1);
        assert!(!waiting_for_admission(&lease, false).unwrap());
    }

    #[test]
    fn admitted_job_is_reconciled_even_after_owner_disappears() {
        let mut lease = queued_lease();
        lease.admit(42, "worker-a".into(), 1);
        assert!(!waiting_for_admission(&lease, false).unwrap());
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn expired_job_deadline_does_not_poll_mutating_operation() {
        let polled = std::cell::Cell::new(false);
        let result = within_job_deadline(Instant::now(), "cancel", async {
            polled.set(true);
            Ok(())
        })
        .await;
        assert!(result.is_err());
        assert!(!polled.get());
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn job_deadline_bounds_a_stalled_daemon_call() {
        let result = within_job_deadline(
            Instant::now() + Duration::from_millis(10),
            "job status",
            std::future::pending::<Result<()>>(),
        )
        .await;
        let error = result.unwrap_err().to_string();
        assert!(error.contains("deadline elapsed during job status"));
        assert!(error.contains("journal retained"));
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn job_deadline_preserves_completed_results_and_inner_errors() {
        let deadline = Instant::now() + Duration::from_secs(60);
        assert_eq!(
            within_job_deadline(deadline, "status", async { Ok(42) })
                .await
                .unwrap(),
            42
        );
        let error: Result<()> = within_job_deadline(deadline, "status", async {
            anyhow::bail!("identity mismatch")
        })
        .await;
        assert_eq!(error.unwrap_err().to_string(), "identity mismatch");
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn expired_poll_budget_is_an_error() {
        assert!(wait_for_job_poll(Instant::now()).await.is_err());
    }

    fn admitted_lease() -> DurableJobLease {
        let mut lease = queued_lease();
        lease.admit(42, "worker-a".into(), 1);
        lease
    }

    fn completion(lease: &DurableJobLease, code: i32) -> Value {
        json!({
            "status": "completed",
            "local_wrapper_id": lease.identity.local_wrapper_id,
            "record": {"id": 42, "worker_id": "worker-a", "exit_code": code},
        })
    }

    #[test]
    fn cancellation_accepts_same_identity_completion_without_a_stop_receipt() {
        let lease = admitted_lease();
        for code in [0, 1, 101, 102, 130, 137] {
            assert_eq!(
                validate_admitted_cancellation(&lease, &completion(&lease, code)).unwrap(),
                AdmittedCancellation::AlreadyCompleted(code)
            );
        }
    }

    #[test]
    fn cancellation_completion_does_not_claim_artifacts_or_retire_the_journal() {
        let mut lease = admitted_lease();
        lease.recovery = Some(json!({"retired": false}));
        let before = lease.clone();
        let report = completed_cancellation_report(&lease, 0);
        assert_eq!(report["exit_code"], 0);
        assert_eq!(report["terminal_acknowledged"], false);
        assert_eq!(report["has_recovery_journal"], true);
        assert_eq!(report["wrapper_stop_requested"], false);
        assert_eq!(lease, before);
    }

    #[test]
    fn cancellation_completion_rejects_missing_or_foreign_identity_components() {
        let lease = admitted_lease();
        for pointer in ["/local_wrapper_id", "/record/id", "/record/worker_id"] {
            let mut reply = completion(&lease, 0);
            *reply.pointer_mut(pointer).unwrap() = Value::Null;
            assert!(validate_admitted_cancellation(&lease, &reply).is_err());
        }
        for (pointer, wrong) in [
            ("/local_wrapper_id", json!("different-wrapper")),
            ("/record/id", json!(43)),
            ("/record/worker_id", json!("worker-b")),
        ] {
            let mut reply = completion(&lease, 0);
            *reply.pointer_mut(pointer).unwrap() = wrong;
            assert!(validate_admitted_cancellation(&lease, &reply).is_err());
        }
    }

    #[test]
    fn completion_requires_a_representable_exit_code() {
        let lease = admitted_lease();
        for code in [Value::Null, json!("0"), json!(i64::MAX), json!(0.5)] {
            let mut reply = completion(&lease, 0);
            reply["record"]["exit_code"] = code;
            assert!(validate_completion(&lease, &reply).is_err());
        }
    }

    #[test]
    fn stopped_wrapper_requires_exact_admitted_build_and_worker_receipt() {
        let lease = admitted_lease();
        let valid = json!({"status": "cancelled", "build_id": 42, "worker_id": "worker-a"});
        assert_eq!(
            validate_admitted_cancellation(&lease, &valid).unwrap(),
            AdmittedCancellation::StopWrapper
        );
        for reply in [
            json!({"status": "cancelled"}),
            json!({"status": "cancelled", "build_id": 43, "worker_id": "worker-a"}),
            json!({"status": "cancelled", "build_id": 42, "worker_id": "worker-b"}),
            json!({"status": "cancelled", "build_id": 42}),
            json!({"status": "not_found", "build_id": 42, "worker_id": "worker-a"}),
            json!({"status": "error", "build_id": 42, "worker_id": "worker-a"}),
            json!({"status": "cancelled_before_start", "local_wrapper_id": lease.identity.local_wrapper_id, "exit_code": 130}),
            json!({"status": "cancelled", "build_id": 42, "worker_id": "worker-a", "local_wrapper_id": "different-wrapper"}),
        ] {
            assert!(validate_admitted_cancellation(&lease, &reply).is_err());
        }
    }

    #[test]
    fn cancellation_cannot_invent_an_admitted_identity() {
        let lease = queued_lease();
        let reply = json!({"status": "cancelled", "build_id": 42, "worker_id": "worker-a"});
        assert!(validate_admitted_cancellation(&lease, &reply).is_err());
        let mut lease = admitted_lease();
        lease.worker_id = None;
        assert!(validate_admitted_cancellation(&lease, &reply).is_err());
    }

    #[test]
    fn unknown_owner_cannot_be_attached_as_a_live_queued_wrapper() {
        let mut lease = queued_lease();
        lease.wrapper_pid = std::process::id();
        assert_eq!(owner_presence(&lease), OwnerPresence::Unknown);
        assert!(waiting_for_admission(&lease, process_matches(&lease)).is_err());
    }

    #[cfg(target_os = "macos")]
    #[test]
    fn native_mac_lease_reload_can_attach_to_its_original_queued_wrapper() {
        let mut lease = queued_lease();
        lease.wrapper_pid = std::process::id();
        lease.process_birth = Some(
            rch_common::process_identity::current_process_identity().expect("native Darwin birth"),
        );
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("lease.json");
        std::fs::write(&path, serde_json::to_vec(&lease).unwrap()).unwrap();
        let restored: DurableJobLease =
            serde_json::from_slice(&std::fs::read(path).unwrap()).unwrap();
        assert!(process_matches(&restored));
        assert!(waiting_for_admission(&restored, process_matches(&restored)).unwrap());
    }

    #[test]
    fn reconciliation_reloads_and_preserves_newer_owner_evidence() {
        let observed = admitted_lease();
        assert!(same_unfinished_owner_without_recovery(&observed, &observed));
        let mut latest = observed.clone();
        latest.recovery = Some(json!({"retired": false}));
        assert!(!same_unfinished_owner_without_recovery(&observed, &latest));
        latest = observed.clone();
        latest.acknowledge_terminal(2);
        assert!(!same_unfinished_owner_without_recovery(&observed, &latest));
        latest = observed.clone();
        latest.identity.admit(43);
        assert!(!same_unfinished_owner_without_recovery(&observed, &latest));
        latest = observed.clone();
        latest.wrapper_pid += 1;
        assert!(!same_unfinished_owner_without_recovery(&observed, &latest));
        latest = observed.clone();
        latest.process_birth = rch_common::process_identity::ProcessIdentity::from_record(
            "3f1c2a9e-5b7d-4e2a-9c1f-0a1b2c3d4e5f:darwin:1791280000:123456",
        );
        assert!(!same_unfinished_owner_without_recovery(&observed, &latest));
    }
}