mj_core/

targets.rs

//! Declarative execution plans for Hel session targets.
//!
//! Plans deliberately contain argv vectors instead of local shell strings.  A
//! shell is used only at the SSH boundary, where OpenSSH necessarily sends a
//! command string; every remotely supplied argument is POSIX-quoted there.

use std::collections::{BTreeMap, BTreeSet};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};

use anyhow::{Context, Result, bail, ensure};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};

use crate::config::{HarnessKind, ImagePullPolicy};

mod diagnostics;
pub use diagnostics::{BlockingOperation, BlockingOperationSnapshot, active_blocking_operations};

pub const SESSION_LABEL: &str = "dev.mj.session";
pub const MANAGED_LABEL: &str = "dev.mj.managed";
pub const SESSION_TAG: &str = "dev.mj.session";
pub const MANAGED_TAG: &str = "dev.mj.managed";
/// Which Mjolnir instance (named `--instance` or data-directory fingerprint)
/// created a worker; see `config::instance_identity`.
pub const INSTANCE_LABEL: &str = "dev.mj.instance";
pub const INSTANCE_TAG: &str = "dev.mj.instance";
/// The shared in-container workspace every container session used before
/// per-session workspaces existed. New sessions record a path under it; see
/// `container_workspace_root`.
pub const CONTAINER_WORKSPACE: &str = "/workspace";

/// The launch phase a command belongs to, reported as launch progress.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub enum ProvisionStage {
    /// Waiting for a background download of this target's image to finish,
    /// rather than starting a second download of the same image.
    PullingImage,
    Provisioning,
    Booting,
    Cloning,
    Syncing,
    Restoring,
    Starting,
    Installing(HarnessKind),
    Compacting,
    RecoveryCopy,
    Verifying,
    Closing,
    StoppingTarget,
    RemovingContainer,
    RemovingStorage,
    CleaningCache,
}

impl ProvisionStage {
    pub fn label(self) -> String {
        match self {
            Self::PullingImage => "Pull image".into(),
            Self::Provisioning => "Provision".into(),
            Self::Booting => "Boot".into(),
            Self::Cloning => "Clone".into(),
            Self::Syncing => "Sync".into(),
            Self::Restoring => "Restore".into(),
            Self::Starting => "Start".into(),
            Self::Installing(harness) => format!("Installing {}", harness.display_name()),
            Self::Compacting => "Compact".into(),
            Self::RecoveryCopy => "Recovery copy".into(),
            Self::Verifying => "Verify".into(),
            Self::Closing => "Shut down agent".into(),
            Self::StoppingTarget => "Stop target".into(),
            Self::RemovingContainer => "Remove container".into(),
            Self::RemovingStorage => "Remove container storage".into(),
            Self::CleaningCache => "Clean cache".into(),
        }
    }
}

#[derive(Clone, PartialEq, Eq)]
struct SensitiveCommandInput(Vec<u8>);

impl std::fmt::Debug for SensitiveCommandInput {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str("<redacted>")
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CommandSpec {
    pub program: String,
    pub args: Vec<String>,
    #[serde(default)]
    pub env: BTreeMap<String, String>,
    /// Replace ambient process variables with the explicitly supplied environment.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub clear_env: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cwd: Option<std::path::PathBuf>,
    pub purpose: String,
    #[serde(default)]
    pub stage: Option<ProvisionStage>,
    /// Commands that share this marker and appear consecutively in a plan's
    /// command list may run concurrently under
    /// [`CommandPlan::execute_concurrent`]. Commands without a marker, or
    /// whose neighbors do not share it, keep running strictly in plan order.
    #[serde(default)]
    pub parallel_group: Option<u32>,
    /// Whether this command brings the session's target into existence. Every
    /// command after it in a provisioning plan runs against a target that
    /// already exists, so a later failure owes that target's teardown.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub creates_target: bool,
    /// This command deliberately leaves a process running after it exits: it
    /// starts a detached worker, whose lifetime belongs to the session rather
    /// than to the launch. The executor still gives the command its own
    /// process group, and cancellation still signals that group, but the
    /// successful completion of a launch must not kill what it launched.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub detaches: bool,
    /// The SSH destination this command opens a connection to, when it does.
    /// Tagged commands pass through [`SshAdmission`] so the daemon never
    /// exceeds the remote `sshd`'s `MaxStartups` budget, and a transport
    /// rejection is retried rather than reported as a command failure.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ssh_destination: Option<String>,
    /// The shared SSH connection this command runs on as one session, when it
    /// does. Just before spawning, the executor leases a session on one of the
    /// connection's masters with [`SshSessions::lease`] and puts the options
    /// that bind the command to that master in front of its arguments (see
    /// [`CommandSpec::open_ssh_session`]). The arguments stored here never
    /// contain them.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ssh_session: Option<SshTarget>,
    /// The session is a fail-fast probe: it is counted on a shard but never
    /// opens a master (see [`CommandSpec::ssh_probe_session`]).
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub ssh_session_probe: bool,
    /// Input that must reach the child without becoming part of its arguments,
    /// environment, serialized plan, or debug representation.
    #[serde(skip)]
    sensitive_stdin: Option<SensitiveCommandInput>,
}

impl CommandSpec {
    pub fn new(
        program: impl Into<String>,
        args: impl IntoIterator<Item = impl Into<String>>,
    ) -> Self {
        Self {
            program: program.into(),
            args: args.into_iter().map(Into::into).collect(),
            env: BTreeMap::new(),
            clear_env: false,
            cwd: None,
            purpose: String::new(),
            stage: None,
            parallel_group: None,
            creates_target: false,
            detaches: false,
            ssh_destination: None,
            ssh_session: None,
            ssh_session_probe: false,
            sensitive_stdin: None,
        }
    }

    pub fn purpose(mut self, purpose: impl Into<String>) -> Self {
        self.purpose = purpose.into();
        self
    }

    pub fn stage(mut self, stage: ProvisionStage) -> Self {
        self.stage = Some(stage);
        self
    }

    /// Mark this command as eligible to run concurrently with its
    /// plan-adjacent siblings that share the same group.
    pub fn parallel_group(mut self, group: u32) -> Self {
        self.parallel_group = Some(group);
        self
    }

    /// Record that this command opens an `ssh` connection to `destination`,
    /// which is the host (or `user@host`) argument, never an option value.
    pub fn ssh_destination(mut self, destination: impl Into<String>) -> Self {
        self.ssh_destination = Some(destination.into());
        self
    }

    /// Record that this command is an `ssh` or `scp` invocation that runs as
    /// one session on `ssh`'s shared connection. It also opens a connection
    /// to that destination, for admission, when sharing is off.
    pub fn ssh_session(mut self, ssh: &SshTarget) -> Self {
        self.ssh_destination = Some(ssh.destination.clone());
        self.ssh_session = Some(ssh.clone());
        self
    }

    /// Record that this command is a fail-fast `ssh` probe, such as a target
    /// validation, that joins `ssh`'s shared connection without ever opening
    /// a master. Its short timeouts and keepalives belong to the probe alone:
    /// a master opened by it would impose them on every later session. The
    /// probe still uses one of the master's sessions while it runs, so the
    /// executor counts it in the session ledger like any other session.
    pub fn ssh_probe_session(mut self, ssh: &SshTarget) -> Self {
        self = self.ssh_session(ssh);
        self.ssh_session_probe = true;
        self
    }

    /// Lease the SSH session this command asks for and return the command as
    /// it must be spawned, with the lease that must outlive the child.
    ///
    /// A command without a session request is returned unchanged. `executor`
    /// runs the master check and, when needed, the opener.
    pub fn open_ssh_session(&self, executor: &dyn CommandExecutor) -> Result<SessionCommand<'_>> {
        let Some(ssh) = &self.ssh_session else {
            return Ok(SessionCommand {
                command: std::borrow::Cow::Borrowed(self),
                lease: None,
            });
        };
        let _waiting = BlockingOperation::start(
            &format!(
                "lease SSH connection to {} for {}",
                ssh.destination, self.purpose
            ),
            "ssh",
        );
        let lease = if self.ssh_session_probe {
            SshSessions::lease_probe(ssh)
        } else {
            SshSessions::lease(ssh, executor)?
        };
        let mut command = self.clone();
        command.ssh_session = None;
        command.ssh_session_probe = false;
        command.args = session_command_args(&self.program, &self.args, ssh, &lease);
        Ok(SessionCommand {
            command: std::borrow::Cow::Owned(command),
            lease: Some(lease),
        })
    }

    /// Mark this command as the one that creates the session's target.
    pub fn creates_target(mut self) -> Self {
        self.creates_target = true;
        self
    }

    /// Feed private file content through the shared concurrent pipe handler.
    /// The bytes stay out of argv, environments, serialization, and Debug.
    pub fn with_sensitive_stdin(mut self, input: Vec<u8>) -> Self {
        self.sensitive_stdin = Some(SensitiveCommandInput(input));
        self
    }
}

/// A command ready to spawn, together with the SSH session lease it runs on.
/// Keep this value alive until the child has exited: dropping it frees the
/// session slot.
#[derive(Debug)]
pub struct SessionCommand<'a> {
    command: std::borrow::Cow<'a, CommandSpec>,
    lease: Option<SshSessionLease>,
}

impl SessionCommand<'_> {
    pub fn command(&self) -> &CommandSpec {
        &self.command
    }

    pub fn lease(&self) -> Option<&SshSessionLease> {
        self.lease.as_ref()
    }

    pub fn into_parts(self) -> (CommandSpec, Option<SshSessionLease>) {
        (self.command.into_owned(), self.lease)
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CommandOutput {
    pub status: i32,
    pub stdout: Vec<u8>,
    pub stderr: Vec<u8>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeploymentCapacityKind {
    Host,
    AwsFleet,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DeploymentCapacityTarget {
    pub id: String,
    pub host: String,
    pub target_ids: Vec<String>,
    pub kind: DeploymentCapacityKind,
    pub local: bool,
    /// Alternative commands for a host, or one command per live AWS instance.
    pub probes: Vec<CommandSpec>,
    /// Prevents a partial AWS fleet sample when one live instance cannot be probed yet.
    pub probe_error: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DeploymentCapacityUsage {
    pub cpu_percent: Option<u8>,
    pub memory_used_bytes: u64,
    pub memory_total_bytes: u64,
    pub logical_cores: u64,
    pub disk_total_bytes: Option<u64>,
}

/// An additional directory made available to one session.
///
/// Containers use isolated mounts. Remote targets may instead receive a
/// controller-packed snapshot at the destination while retaining this shared
/// persisted shape.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(from = "AdditionalMountRepr", into = "AdditionalMountRepr")]
pub struct AdditionalMount {
    pub source: PathBuf,
    pub destination: PathBuf,
    pub access: MountAccess,
}

/// How a container sees an attached directory.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MountAccess {
    /// The container can read the directory but not change it.
    Ro,
    /// The container writes to a private copy-on-write overlay; the host
    /// directory never changes.
    Cow,
    /// The container writes straight through to the host directory.
    Rw,
}

impl MountAccess {
    pub const ALL: [Self; 3] = [Self::Ro, Self::Cow, Self::Rw];

    pub fn label(self) -> &'static str {
        match self {
            Self::Ro => "ro",
            Self::Cow => "cow",
            Self::Rw => "rw",
        }
    }

    /// The mode this attachment takes where the host filesystem cannot carry
    /// a copy-on-write overlay. Read-only is the substitute, because the
    /// alternative would let the container write through to the host
    /// directory the user asked to keep unchanged.
    ///
    /// This is the one rule for an unavailable overlay: the wizard offers the
    /// same modes [`Self::offered`] lists, and the runtime downgrades a stored
    /// mount the same way.
    pub fn without_overlay(self) -> Self {
        match self {
            Self::Cow => Self::Ro,
            kept => kept,
        }
    }

    /// The modes an attachment may be given, given whether the host filesystem
    /// can carry the copy-on-write overlay.
    pub fn offered(overlay_available: bool) -> Vec<Self> {
        Self::ALL
            .into_iter()
            .filter(|access| overlay_available || access.without_overlay() == *access)
            .collect()
    }
}

/// The numeric identity of a container image's configured user, read from the
/// image on the host that runs it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct ImageUser {
    pub uid: u32,
    pub gid: u32,
}

/// Podman's user-namespace option for a session container, or `None` when the
/// container keeps Podman's default mapping.
///
/// A rootless Podman container maps the image's user onto the host user
/// running the daemon, so a file the container writes into an attached
/// directory is owned by that host user instead of a subordinate id. The
/// mapping has to name the image's own ids: plain `keep-id` maps the host user
/// onto uid 1000 inside the container and demotes an image that runs as root,
/// which is a change in how the container runs. Without the ids there is no
/// safe option to pass, so the container runs the way it did before.
pub fn podman_userns_option(image_user: Option<ImageUser>) -> Option<String> {
    image_user.map(|ImageUser { uid, gid }| format!("--userns=keep-id:uid={uid},gid={gid}"))
}

/// The persisted shape. Read-only and copy-on-write mounts keep the original
/// `read_only` boolean alone, so archives and API payloads that older builds
/// read stay unchanged; only read-write adds `access`, which older builds
/// cannot represent.
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct AdditionalMountRepr {
    source: PathBuf,
    destination: PathBuf,
    #[serde(default)]
    read_only: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    access: Option<MountAccess>,
}

impl From<AdditionalMountRepr> for AdditionalMount {
    fn from(repr: AdditionalMountRepr) -> Self {
        let access = repr.access.unwrap_or(if repr.read_only {
            MountAccess::Ro
        } else {
            MountAccess::Cow
        });
        Self {
            source: repr.source,
            destination: repr.destination,
            access,
        }
    }
}

impl From<AdditionalMount> for AdditionalMountRepr {
    fn from(mount: AdditionalMount) -> Self {
        Self {
            source: mount.source,
            destination: mount.destination,
            read_only: mount.access == MountAccess::Ro,
            access: (mount.access == MountAccess::Rw).then_some(MountAccess::Rw),
        }
    }
}

/// Why a filesystem cannot host a container target's copy-on-write overlay,
/// or `None` when it can. Unknown types are allowed: the overlay is the better
/// mount and only a filesystem known to break it is downgraded.
///
/// The names are those `stat -f -c %T` reports, matched case-insensitively.
pub fn overlay_unsupported_filesystem(filesystem: &str) -> Option<&'static str> {
    let name = filesystem.trim().to_ascii_lowercase();
    // FUSE reports the backing driver as `fuse.sshfs`, `fuse.s3fs`, and so on.
    if name == "fuse" || name == "fuseblk" || name.starts_with("fuse.") {
        return Some("FUSE filesystem");
    }
    match name.as_str() {
        "nfs" | "nfs4" | "cifs" | "smb2" | "smb3" | "9p" | "v9fs" | "virtiofs" | "ceph"
        | "lustre" | "afs" | "glusterfs" | "ocfs2" | "gfs" | "gfs2" => Some("network filesystem"),
        "msdos" | "vfat" | "fat" | "exfat" | "ntfs" | "ntfs3" => Some("no POSIX metadata"),
        "overlayfs" => Some("overlay stacking limit"),
        _ => None,
    }
}

/// Container destinations cannot use the controller or login user's home.
pub fn validate_mount_destination(path: &Path) -> Result<()> {
    ensure!(
        path.is_absolute()
            && !path
                .components()
                .any(|part| part == std::path::Component::ParentDir),
        "additional mount destination must be a safe absolute container path; ~ is not supported"
    );
    Ok(())
}

pub fn validate_additional_mounts(mounts: &[AdditionalMount]) -> Result<()> {
    let mut destinations = BTreeSet::new();
    for mount in mounts {
        if !mount.source.is_absolute() || mount.source.as_os_str().is_empty() {
            bail!("additional mount source must be an absolute directory path");
        }
        validate_mount_destination(&mount.destination)?;
        if !destinations.insert(mount.destination.clone()) {
            bail!(
                "additional mount destination {:?} is configured more than once",
                mount.destination
            );
        }
    }
    Ok(())
}

/// Choose the editable default destination for an additional host directory.
pub fn default_mount_destination(source: &Path, existing: &[AdditionalMount]) -> PathBuf {
    let basename = source
        .file_name()
        .filter(|name| !name.is_empty())
        .unwrap_or_else(|| std::ffi::OsStr::new("mount"));
    let base = PathBuf::from("/mnt").join(basename);
    if !existing.iter().any(|mount| mount.destination == base) {
        return base;
    }
    for number in 2.. {
        let candidate =
            PathBuf::from("/mnt").join(format!("{}-{number}", basename.to_string_lossy()));
        if !existing.iter().any(|mount| mount.destination == candidate) {
            return candidate;
        }
    }
    unreachable!("a finite mount list always has an unused numbered destination")
}

pub trait CommandExecutor {
    fn execute(&self, command: &CommandSpec) -> Result<CommandOutput>;

    /// Whether the operation supervising this executor has requested
    /// cancellation. Test executors and ordinary process execution are not
    /// cancellable unless they opt in.
    fn cancellation_requested(&self) -> bool {
        false
    }

    /// Report entry into a lifecycle stage. Callers that cover more than one
    /// command should hold a [`ProvisionStageGuard`] for the whole operation
    /// so concurrent stages remain visible between subprocesses.
    fn stage_started(&self, _stage: ProvisionStage) {}

    /// Report exit from a lifecycle stage previously passed to
    /// [`Self::stage_started`].
    fn stage_finished(&self, _stage: ProvisionStage) {}

    /// Report a decision an operation made on the user's behalf. This is not a
    /// failure: the work continues, and the user is told what changed.
    fn notify_notice(&self, _notice: &str) {}

    /// The Move source has been sealed; its destination belongs exclusively
    /// to the lifecycle until queue admission completes.
    fn reserve_move_destination(&self) {}

    /// The durable Move owns this interruptible copy; daemon handoff may resume it.
    fn begin_resumable_move_work(&self) -> Result<()> {
        Ok(())
    }
    fn end_resumable_move_work(&self) -> Result<()> {
        Ok(())
    }

    fn execute_with_stdin(
        &self,
        _command: &CommandSpec,
        _input: &mut (dyn Read + Send),
    ) -> Result<CommandOutput> {
        bail!("this command executor does not support streamed stdin")
    }
}

/// A scoped lifecycle-stage report for controller-side work or a sequence of
/// commands. Dropping the guard reports completion even when the work returns
/// early with an error.
pub struct ProvisionStageGuard<'a, E: CommandExecutor + ?Sized> {
    executor: &'a E,
    stage: ProvisionStage,
}

impl<'a, E: CommandExecutor + ?Sized> ProvisionStageGuard<'a, E> {
    pub fn new(executor: &'a E, stage: ProvisionStage) -> Self {
        executor.stage_started(stage);
        Self { executor, stage }
    }
}

impl<E: CommandExecutor + ?Sized> Drop for ProvisionStageGuard<'_, E> {
    fn drop(&mut self) {
        self.executor.stage_finished(self.stage);
    }
}

pub struct ProcessExecutor;

/// Run one `ssh` invocation under process-wide admission control, retrying it
/// when the server turned the connection away before authentication.
///
/// A permit is held only while a child is actually running and is released
/// between attempts, because a waiting retry occupies no connection slot. A
/// transport rejection means the remote command never started, so re-running
/// the whole invocation cannot repeat a side effect.
///
/// A command that asks for a shared-connection session leases it on each
/// attempt, before taking its permit: opening a master takes a permit of its
/// own, and waiting for that while holding one could exhaust the gate. A
/// session turned away by the transport usually means its master died, so
/// the lease is invalidated and the retry checks and reopens the master.
///
/// Commands that are not tagged with a destination run untouched.
fn with_ssh_admission(
    command: &CommandSpec,
    executor: &dyn CommandExecutor,
    is_cancelled: &dyn Fn() -> bool,
    mut run: impl FnMut(&CommandSpec) -> Result<CommandOutput>,
) -> Result<CommandOutput> {
    let Some(destination) = command.ssh_destination.as_deref() else {
        return run(command);
    };
    for attempt in 1..=SSH_RETRY_ATTEMPTS {
        let session = command.open_ssh_session(executor)?;
        let output = {
            let _permit = SshAdmission::acquire_unless(destination, is_cancelled)?;
            run(session.command())?
        };
        let refusal = ssh_refusal(output.status, &String::from_utf8_lossy(&output.stderr));
        // A refused session found its master alive; any other refusal may
        // mean the master is gone.
        if refusal == Some(SshRefusal::BeforeAuthentication)
            && let Some(lease) = session.lease()
        {
            lease.invalidate();
        }
        drop(session);
        let Some(refusal) = refusal else {
            return Ok(output);
        };
        let stderr = String::from_utf8_lossy(&output.stderr);
        if attempt == SSH_RETRY_ATTEMPTS {
            refusal.log_exhausted(destination, &command.purpose, stderr.trim());
            return Ok(output);
        }
        let delay = ssh_retry_delay(attempt);
        refusal.log_retry(destination, &command.purpose, attempt, delay, stderr.trim());
        if !sleep_unless_cancelled(delay, is_cancelled) {
            bail!("operation cancelled while {}", command.purpose);
        }
    }
    unreachable!("the final attempt always returns");
}

/// Wait out `delay`, giving up early if the supervising operation is
/// cancelled. Returns whether the wait completed.
fn sleep_unless_cancelled(delay: Duration, is_cancelled: &dyn Fn() -> bool) -> bool {
    let deadline = Instant::now() + delay;
    loop {
        if is_cancelled() {
            return false;
        }
        let remaining = deadline.saturating_duration_since(Instant::now());
        if remaining.is_zero() {
            return true;
        }
        std::thread::sleep(remaining.min(Duration::from_millis(50)));
    }
}

/// One debug line per finished target command, so a slow launch or resume
/// phase can be attributed from logs instead of re-profiled by hand.
pub fn trace_command_duration(command: &CommandSpec, started: Instant, status: i32) {
    tracing::debug!(
        purpose = command.purpose.as_str(),
        program = command.program.as_str(),
        status,
        elapsed_ms = started.elapsed().as_millis() as u64,
        "target command finished"
    );
}

impl ProcessExecutor {
    /// One attempt, with no admission or retry of its own.
    fn run_once(&self, command: &CommandSpec) -> Result<CommandOutput> {
        if let Some(input) = &command.sensitive_stdin {
            let mut input = std::io::Cursor::new(input.0.as_slice());
            // Owned bytes, so each attempt gets its own reader.
            return stream_command_with_stdin(
                cancellable_command(command),
                command,
                &mut input,
                &|| false,
            );
        }
        let started = Instant::now();
        let output = configured_command(command)
            .stdin(Stdio::null())
            .output()
            .with_context(|| format!("run {} for {}", command.program, command.purpose))?;
        let status = output.status.code().unwrap_or(-1);
        trace_command_duration(command, started, status);
        Ok(CommandOutput {
            status,
            stdout: output.stdout,
            stderr: output.stderr,
        })
    }
}

impl CommandExecutor for ProcessExecutor {
    fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
        let _running = BlockingOperation::command(command);
        with_ssh_admission(command, self, &|| false, |command| self.run_once(command))
    }

    fn execute_with_stdin(
        &self,
        command: &CommandSpec,
        input: &mut (dyn Read + Send),
    ) -> Result<CommandOutput> {
        let _running = BlockingOperation::command(command);
        // A caller's stream cannot be replayed, so this path takes a session
        // and a permit but never retries.
        let session = command.open_ssh_session(self)?;
        let _permit = command
            .ssh_destination
            .as_deref()
            .map(SshAdmission::acquire);
        let command = session.command();
        let process = cancellable_command(command);
        // Plain process execution is not cancellable, so the transfer only
        // ends when the child does.
        stream_command_with_stdin(process, command, input, &|| false)
    }
}

/// Streams `input` into a freshly spawned child and collects its output.
///
/// Both executors share this one implementation because the pipe edge cases
/// below are easy to get subtly wrong in a second copy.
///
/// `is_cancelled` reports whether the supervising operation wants the transfer
/// abandoned; [`ProcessExecutor`] passes a check that is never true, which also
/// makes the kill path below unreachable for it.
fn stream_command_with_stdin(
    mut process: Command,
    command: &CommandSpec,
    input: &mut (dyn Read + Send),
    is_cancelled: &(dyn Fn() -> bool + Sync),
) -> Result<CommandOutput> {
    let started = Instant::now();
    if is_cancelled() {
        bail!("operation cancelled");
    }
    let mut child = process
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .with_context(|| format!("run {} for {}", command.program, command.purpose))?;
    let group =
        (!command.detaches).then(|| crate::subprocess::ProcessGroupGuard::new(Some(child.id())));
    let stdin = child
        .stdin
        .take()
        .context("streamed command stdin missing")?;
    let stdout = child
        .stdout
        .take()
        .context("streamed command stdout missing")?;
    let stderr = child
        .stderr
        .take()
        .context("streamed command stderr missing")?;
    // Reader threads keep the child's output pipes drained; a child that fills
    // one while nobody reads would block instead of exiting.
    let stdout_reader = PipeCollector::spawn(stdout);
    let stderr_reader = PipeCollector::spawn(stderr);
    let process_result = std::thread::scope(|scope| -> Result<_> {
        // Pipe writes can block forever when a remote helper stops reading.
        // Keep the writer off the supervising thread so cancellation can kill
        // the process group and thereby close the blocked pipe.
        let input_writer = scope.spawn(move || -> Result<()> {
            // Owning `stdin` here is what closes the pipe's write end once the
            // transfer finishes. A child that reads to EOF, such as
            // `mj worker export-checkpoint --spec -`, never exits while any
            // copy of the write end is still open.
            let mut stdin = stdin;
            let mut buffer = [0_u8; 64 * 1024];
            loop {
                // Checking before each chunk makes large checkpoint copies
                // cooperatively cancellable without changing the executor
                // interface.
                if is_cancelled() {
                    bail!("operation cancelled");
                }
                let count = input.read(&mut buffer).context("read command input")?;
                if count == 0 {
                    break;
                }
                stdin
                    .write_all(&buffer[..count])
                    .context("stream command input")?;
            }
            stdin.flush().context("flush command input")
        });
        let mut status = None;
        let mut exited_at = None;
        let mut group_killed = false;
        let status = loop {
            if is_cancelled() {
                terminate_cancellable_child(&mut child);
                if let Err(error) = input_writer.join() {
                    tracing::warn!(
                        purpose = command.purpose.as_str(),
                        "streamed command input writer panicked while cancelling: {error:?}"
                    );
                }
                bail!("operation cancelled while {}", command.purpose);
            }
            match if status.is_some() {
                Ok(status)
            } else {
                child.try_wait()
            } {
                Ok(observed) => status = observed,
                Err(error) => {
                    terminate_cancellable_child(&mut child);
                    if let Err(join_error) = input_writer.join() {
                        tracing::warn!(
                            purpose = command.purpose.as_str(),
                            "streamed command input writer panicked while waiting: {join_error:?}"
                        );
                    }
                    return Err(error).with_context(|| format!("wait for {}", command.purpose));
                }
            }
            // The command completes when its leader exits. Descendants can
            // retain the pipes, so output is collected for a bounded time;
            // then the owned group is terminated, which also releases a
            // writer blocked on a pipe a descendant holds.
            if let Some(status) = status {
                let exited_at = *exited_at.get_or_insert_with(Instant::now);
                let drained = (stdout_reader.is_finished() && stderr_reader.is_finished())
                    || exited_at.elapsed() >= IO_DRAIN_TIMEOUT;
                if drained && !input_writer.is_finished() && !group_killed {
                    group_killed = true;
                    if let Some(group) = &group {
                        group.kill();
                    }
                }
                if drained && input_writer.is_finished() {
                    break status;
                }
            }
            std::thread::sleep(Duration::from_millis(25));
        };
        let input_result = input_writer
            .join()
            .map_err(|_| anyhow::anyhow!("streamed command input writer panicked"))?;
        Ok((status, input_result))
    });
    let deadline = Instant::now();
    let stdout = stdout_reader.finish("stdout", deadline)?;
    let stderr = stderr_reader.finish("stderr", deadline)?;
    let (status, input_result) = process_result?;
    drop(group);
    if status.success() {
        // A child that exited first explains the failure through its own
        // status and stderr; the broken pipe that exit caused would only hide
        // it. A successful child must not hide an input error.
        input_result?;
    }
    let status = status.code().unwrap_or(-1);
    trace_command_duration(command, started, status);
    Ok(CommandOutput {
        status,
        stdout,
        stderr,
    })
}

#[derive(Clone)]
pub struct CancellableProcessExecutor {
    cancelled: Arc<AtomicBool>,
    deadline: Option<Instant>,
}

/// Cancels synchronous executor work when its supervising async wait is dropped.
pub struct ProcessCancellationGuard(Arc<AtomicBool>);

impl Drop for ProcessCancellationGuard {
    fn drop(&mut self) {
        self.0.store(true, Ordering::Release);
    }
}

impl CancellableProcessExecutor {
    pub fn cancel_on_drop(&self) -> ProcessCancellationGuard {
        ProcessCancellationGuard(self.cancelled.clone())
    }

    pub fn new(cancelled: Arc<AtomicBool>) -> Self {
        Self {
            cancelled,
            deadline: None,
        }
    }

    pub fn is_cancelled(&self) -> bool {
        self.cancelled.load(Ordering::Acquire)
            || self
                .deadline
                .is_some_and(|deadline| Instant::now() >= deadline)
    }

    pub fn with_timeout(timeout: Duration) -> Self {
        Self {
            cancelled: Arc::new(AtomicBool::new(false)),
            deadline: Some(Instant::now() + timeout),
        }
    }

    /// Bounds an existing flag-based executor with a deadline, so a wedged
    /// child becomes a reported failure instead of running forever.
    pub fn with_deadline(mut self, timeout: Duration) -> Self {
        self.deadline = Some(Instant::now() + timeout);
        self
    }

    fn check_cancelled(&self) -> Result<()> {
        if self.is_cancelled() {
            bail!("operation cancelled");
        }
        Ok(())
    }
}

fn configured_command(command: &CommandSpec) -> Command {
    let mut process = Command::new(&command.program);
    if command.clear_env {
        process.env_clear();
    }
    if let Some(cwd) = &command.cwd {
        process.current_dir(cwd);
    }
    process.args(&command.args).envs(&command.env);
    process
}

fn cancellable_command(command: &CommandSpec) -> Command {
    #[cfg(unix)]
    let mut process = configured_command(command);
    #[cfg(not(unix))]
    let process = configured_command(command);
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt as _;
        process.process_group(0);
    }
    process
}

/// How long a command's output may keep arriving after its leader exits.
/// A backgrounded descendant inherits the output pipes and can hold them open
/// for its whole life, although the command it was started by is finished.
/// The same bound as Codex's `IO_DRAIN_TIMEOUT_MS`. The owned process group is
/// terminated once this drain ends.
const IO_DRAIN_TIMEOUT: Duration = Duration::from_secs(2);

/// A child pipe that can be waited on with a timeout, so its reader can be
/// stopped while a descendant still holds the write end.
trait PollablePipe: Read + Send + 'static {
    fn readable(&self, timeout: Duration) -> bool;
}

macro_rules! pollable_pipe {
    ($pipe:ty) => {
        impl PollablePipe for $pipe {
            #[cfg(unix)]
            fn readable(&self, timeout: Duration) -> bool {
                use std::os::fd::AsRawFd as _;
                let mut poll = libc::pollfd {
                    fd: self.as_raw_fd(),
                    events: libc::POLLIN,
                    revents: 0,
                };
                let millis = i32::try_from(timeout.as_millis()).unwrap_or(i32::MAX);
                // SAFETY: `poll` is one initialized pollfd for an open descriptor.
                unsafe { libc::poll(&raw mut poll, 1, millis) > 0 }
            }

            // Without polling, a read blocks until the pipe closes.
            #[cfg(not(unix))]
            fn readable(&self, _timeout: Duration) -> bool {
                true
            }
        }
    };
}

pollable_pipe!(std::process::ChildStdout);
pollable_pipe!(std::process::ChildStderr);

/// Collects one child pipe on its own thread. Stopping it drops the thread's
/// end of the pipe, so a descendant that keeps the pipe open cannot hold the
/// reader past the drain.
struct PipeCollector {
    bytes: Arc<std::sync::Mutex<Vec<u8>>>,
    stop: Arc<AtomicBool>,
    thread: Option<std::thread::JoinHandle<std::io::Result<()>>>,
}

impl PipeCollector {
    fn spawn(mut pipe: impl PollablePipe) -> Self {
        let bytes = Arc::new(std::sync::Mutex::new(Vec::new()));
        let stop = Arc::new(AtomicBool::new(false));
        let (collected, stopped) = (bytes.clone(), stop.clone());
        let thread = std::thread::spawn(move || {
            let mut chunk = [0_u8; 8192];
            while !stopped.load(Ordering::Acquire) {
                if !pipe.readable(Duration::from_millis(25)) {
                    continue;
                }
                match pipe.read(&mut chunk) {
                    Ok(0) => break,
                    Ok(count) => collected
                        .lock()
                        .unwrap_or_else(std::sync::PoisonError::into_inner)
                        .extend_from_slice(&chunk[..count]),
                    Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
                    Err(error) => return Err(error),
                }
            }
            Ok(())
        });
        Self {
            bytes,
            stop,
            thread: Some(thread),
        }
    }

    fn is_finished(&self) -> bool {
        self.thread
            .as_ref()
            .is_none_or(std::thread::JoinHandle::is_finished)
    }

    /// Waits for end of stream until `deadline`, then stops the reader and
    /// returns everything it collected.
    fn finish(mut self, stream: &str, deadline: Instant) -> Result<Vec<u8>> {
        while !self.is_finished() && Instant::now() < deadline {
            std::thread::sleep(Duration::from_millis(5));
        }
        self.stop.store(true, Ordering::Release);
        self.thread
            .take()
            .context("command reader already joined")?
            .join()
            .map_err(|_| anyhow::anyhow!("command {stream} reader panicked"))?
            .with_context(|| format!("read command {stream}"))?;
        let mut bytes = self
            .bytes
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        Ok(std::mem::take(&mut bytes))
    }
}

impl Drop for PipeCollector {
    fn drop(&mut self) {
        self.stop.store(true, Ordering::Release);
    }
}

fn terminate_cancellable_child(child: &mut std::process::Child) {
    #[cfg(unix)]
    // The child owns a fresh process group, so descendants such as an SSH or
    // shell helper cannot keep its output pipes open after cancellation. A
    // group that is already gone is the wanted outcome, not a failure, so the
    // shared helper decides what deserves a warning.
    if let Err(error) = crate::subprocess::signal_process_group(child.id() as i32, libc::SIGKILL) {
        tracing::warn!(pid = child.id(), %error, "could not terminate cancelled command process group");
    }
    #[cfg(not(unix))]
    if let Err(error) = child.kill() {
        tracing::warn!(pid = child.id(), %error, "could not terminate cancelled command");
    }
    if let Err(error) = child.wait() {
        tracing::warn!(pid = child.id(), %error, "could not reap cancelled command");
    }
}

impl CancellableProcessExecutor {
    /// One attempt, with no admission or retry of its own.
    fn run_once(&self, command: &CommandSpec) -> Result<CommandOutput> {
        if let Some(input) = &command.sensitive_stdin {
            let mut input = std::io::Cursor::new(input.0.as_slice());
            // Owned bytes, so each attempt gets its own reader.
            return stream_command_with_stdin(
                cancellable_command(command),
                command,
                &mut input,
                &|| self.is_cancelled(),
            );
        }
        let started = Instant::now();
        self.check_cancelled()?;
        let mut child = cancellable_command(command)
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
            .with_context(|| format!("run {} for {}", command.program, command.purpose))?;
        let group = (!command.detaches)
            .then(|| crate::subprocess::ProcessGroupGuard::new(Some(child.id())));
        let stdout = child.stdout.take().context("command stdout missing")?;
        let stderr = child.stderr.take().context("command stderr missing")?;
        let stdout_reader = PipeCollector::spawn(stdout);
        let stderr_reader = PipeCollector::spawn(stderr);
        let mut status = None;
        let mut exited_at = None;
        let status = loop {
            if self.is_cancelled() {
                terminate_cancellable_child(&mut child);
                let deadline = Instant::now() + IO_DRAIN_TIMEOUT;
                for (stream, reader) in [("stdout", stdout_reader), ("stderr", stderr_reader)] {
                    if let Err(error) = reader.finish(stream, deadline) {
                        tracing::warn!(stream, %error, "cancelled command reader failed");
                    }
                }
                bail!("operation cancelled while {}", command.purpose);
            }
            if status.is_none() {
                status = child
                    .try_wait()
                    .with_context(|| format!("wait for {}", command.purpose))?;
            }
            // The command completes when its leader exits. Descendants can
            // retain the pipes, so output is collected for a bounded time.
            if let Some(status) = status {
                let exited_at = *exited_at.get_or_insert_with(Instant::now);
                if (stdout_reader.is_finished() && stderr_reader.is_finished())
                    || exited_at.elapsed() >= IO_DRAIN_TIMEOUT
                {
                    break status;
                }
            }
            std::thread::sleep(Duration::from_millis(25));
        };
        let deadline = Instant::now();
        let stdout = stdout_reader.finish("stdout", deadline)?;
        let stderr = stderr_reader.finish("stderr", deadline)?;
        let status = status.code().unwrap_or(-1);
        drop(group);
        trace_command_duration(command, started, status);
        Ok(CommandOutput {
            status,
            stdout,
            stderr,
        })
    }
}

impl CommandExecutor for CancellableProcessExecutor {
    fn cancellation_requested(&self) -> bool {
        self.is_cancelled()
    }

    fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
        let _running = BlockingOperation::command(command);
        with_ssh_admission(command, self, &|| self.is_cancelled(), |command| {
            self.run_once(command)
        })
    }

    fn execute_with_stdin(
        &self,
        command: &CommandSpec,
        input: &mut (dyn Read + Send),
    ) -> Result<CommandOutput> {
        let _running = BlockingOperation::command(command);
        // A caller's stream cannot be replayed, so this path takes a session
        // and a permit but never retries.
        let session = command.open_ssh_session(self)?;
        let _permit = command
            .ssh_destination
            .as_deref()
            .map(|destination| SshAdmission::acquire_unless(destination, &|| self.is_cancelled()))
            .transpose()?;
        let command = session.command();
        // The child runs in its own process group so cancellation can kill the
        // whole group, which is what releases a writer blocked on a full pipe.
        stream_command_with_stdin(cancellable_command(command), command, input, &|| {
            self.is_cancelled()
        })
    }
}

/// A command supervised by [`BoundedProcessExecutor`] did not finish before
/// its deadline. Callers downcast to this to tell a hung probe apart from a
/// command that could not be started at all.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CommandTimedOut {
    pub program: String,
    pub purpose: String,
    pub timeout: Duration,
}

impl std::fmt::Display for CommandTimedOut {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            formatter,
            "`{}` did not answer within {} seconds while trying to {}",
            self.program,
            self.timeout.as_secs(),
            self.purpose
        )
    }
}

impl std::error::Error for CommandTimedOut {}

/// Runs every command with its own deadline.
///
/// [`CancellableProcessExecutor::with_timeout`] bounds a whole operation from a
/// single shared deadline, which suits one provisioning run. Prerequisite
/// probes are different: each one is expected to answer quickly, and a wedged
/// socket or blackholed network must not stall the probes that follow it. A
/// timeout here names the probe that hung, so the caller can report it the same
/// way it reports any other probe failure.
#[derive(Debug, Clone, Copy)]
pub struct BoundedProcessExecutor {
    timeout: Duration,
}

impl BoundedProcessExecutor {
    pub const fn new(timeout: Duration) -> Self {
        Self { timeout }
    }
}

impl CommandExecutor for BoundedProcessExecutor {
    fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
        let executor = CancellableProcessExecutor::with_timeout(self.timeout);
        executor.execute(command).map_err(|error| {
            if executor.is_cancelled() {
                anyhow::Error::new(CommandTimedOut {
                    program: command.program.clone(),
                    purpose: command.purpose.clone(),
                    timeout: self.timeout,
                })
            } else {
                error
            }
        })
    }

    fn execute_with_stdin(
        &self,
        command: &CommandSpec,
        input: &mut (dyn Read + Send),
    ) -> Result<CommandOutput> {
        CancellableProcessExecutor::with_timeout(self.timeout).execute_with_stdin(command, input)
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CommandPlan {
    pub description: String,
    pub commands: Vec<CommandSpec>,
}

impl CommandPlan {
    /// Supply one container environment value without placing it in the
    /// Podman/SSH argument vector. The target launcher reads the value from
    /// stdin, exports it, and asks the container engine to inherit it by name.
    pub fn provide_target_environment_secret(
        &mut self,
        target: &TargetTemplate,
        name: &str,
        value: &str,
    ) -> Result<()> {
        ensure!(
            !name.is_empty()
                && name.bytes().enumerate().all(|(index, byte)| byte == b'_'
                    || byte.is_ascii_alphabetic()
                    || (index > 0 && byte.is_ascii_digit())),
            "invalid secret environment variable name"
        );
        ensure!(
            !value.as_bytes().contains(&b'\n') && !value.as_bytes().contains(&b'\r'),
            "secret environment value cannot contain a newline"
        );
        let command = self
            .commands
            .iter_mut()
            .find(|command| command.creates_target)
            .context("provisioning plan has no target creation command")?;
        let read_and_export = format!("IFS= read -r {name} || exit 1; export {name};");
        match target {
            TargetTemplate::LocalPodman(_)
            | TargetTemplate::LocalDocker(_)
            | TargetTemplate::AppleContainer(_) => {
                let program = std::mem::replace(&mut command.program, "sh".to_owned());
                let args = std::mem::take(&mut command.args);
                command.args = vec![
                    "-c".to_owned(),
                    format!("{read_and_export} exec \"$@\""),
                    "mj-secret-env".to_owned(),
                    program,
                ];
                command.args.extend(args);
            }
            TargetTemplate::SshPodman { .. } | TargetTemplate::SshDocker { .. } => {
                let remote = command
                    .args
                    .last_mut()
                    .context("remote container command has no SSH command argument")?;
                *remote = format!("{read_and_export} exec {remote}");
            }
            TargetTemplate::LocalBare
            | TargetTemplate::AwsEc2(_)
            | TargetTemplate::SshBare { .. } => {
                bail!("target does not support inherited container environment")
            }
        }
        let mut input = value.as_bytes().to_vec();
        input.push(b'\n');
        command.sensitive_stdin = Some(SensitiveCommandInput(input));
        Ok(())
    }

    pub fn execute(&self, executor: &impl CommandExecutor) -> Result<Vec<CommandOutput>> {
        let mut outputs = Vec::with_capacity(self.commands.len());
        for command in &self.commands {
            let output = executor.execute(command)?;
            if output.status != 0 {
                bail!(
                    "{} failed with status {}: {}",
                    command.purpose,
                    output.status,
                    String::from_utf8_lossy(&output.stderr)
                );
            }
            outputs.push(output);
        }
        Ok(outputs)
    }

    /// Execute the plan the same way [`Self::execute`] does, except that
    /// commands sharing a [`CommandSpec::parallel_group`] marker and
    /// appearing consecutively in `commands` run concurrently as one batch.
    ///
    /// A batch starts only once every earlier command has succeeded, and a
    /// batch that fails reports the first failure in plan order regardless
    /// of which command finished first — the same fail-fast contract
    /// [`Self::execute`] provides between individual commands. This method
    /// requires a `Sync` executor because a batch shares it across threads;
    /// [`Self::execute`] keeps working with non-`Sync` executors such as
    /// test fakes built on `RefCell`.
    pub fn execute_concurrent(
        &self,
        executor: &(impl CommandExecutor + Sync),
    ) -> Result<Vec<CommandOutput>> {
        let mut outputs = Vec::with_capacity(self.commands.len());
        let mut index = 0;
        while index < self.commands.len() {
            let group = self.commands[index].parallel_group;
            let mut end = index + 1;
            if group.is_some() {
                while end < self.commands.len() && self.commands[end].parallel_group == group {
                    end += 1;
                }
            }
            let batch = &self.commands[index..end];
            if let [command] = batch {
                outputs.push(checked_command_output(command, executor.execute(command)?)?);
            } else {
                let results: Vec<Result<CommandOutput>> = std::thread::scope(|scope| {
                    let handles: Vec<_> = batch
                        .iter()
                        .map(|command| scope.spawn(|| executor.execute(command)))
                        .collect();
                    handles
                        .into_iter()
                        .map(|handle| match handle.join() {
                            Ok(result) => result,
                            Err(panic) => Err(anyhow::anyhow!(
                                "concurrent command thread panicked: {}",
                                command_thread_panic_message(panic.as_ref())
                            )),
                        })
                        .collect()
                });
                for (command, result) in batch.iter().zip(results) {
                    outputs.push(checked_command_output(command, result?)?);
                }
            }
            index = end;
        }
        Ok(outputs)
    }

    /// Split the plan around the command that creates the session's target:
    /// the commands through that one, then the commands that run against a
    /// target which already exists.
    ///
    /// A plan that creates nothing — an existing project directory, say —
    /// splits into nothing, so a caller never arms a teardown for a target it
    /// did not bring into existence.
    pub fn split_at_target_creation(&self) -> Option<(Self, Self)> {
        let created = self
            .commands
            .iter()
            .position(|command| command.creates_target)?;
        let (creation, remainder) = self.commands.split_at(created + 1);
        Some((
            Self {
                description: self.description.clone(),
                commands: creation.to_vec(),
            },
            Self {
                description: self.description.clone(),
                commands: remainder.to_vec(),
            },
        ))
    }
}

/// Fail the same way [`CommandPlan::execute`] does for a non-zero exit
/// status; kept as a shared helper so [`CommandPlan::execute_concurrent`]
/// reports identical error text.
pub fn checked_command_output(
    command: &CommandSpec,
    output: CommandOutput,
) -> Result<CommandOutput> {
    if output.status != 0 {
        bail!(
            "{} failed with status {}: {}",
            command.purpose,
            output.status,
            String::from_utf8_lossy(&output.stderr)
        );
    }
    Ok(output)
}

/// Describe a spawned command thread's panic payload for error context.
pub fn command_thread_panic_message(payload: &(dyn std::any::Any + Send)) -> String {
    if let Some(message) = payload.downcast_ref::<&str>() {
        (*message).to_owned()
    } else if let Some(message) = payload.downcast_ref::<String>() {
        message.clone()
    } else {
        "non-string panic payload".to_owned()
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RepositorySpec {
    /// Network clone URL. Managed workspaces require a configured remote.
    pub url: Option<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub push_urls: Vec<String>,
    pub destination: String,
    pub git_ref: Option<String>,
    /// Read-only bare repository mounted into the target for Git object reuse.
    /// A missing or unusable reference is only an optimization miss.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub reference: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProjectBundleSpec {
    pub primary: String,
    pub repositories: Vec<RepositorySpec>,
}

impl ProjectBundleSpec {
    pub fn validate(&self) -> Result<()> {
        validate_relative_path(&self.primary)?;
        if self.repositories.is_empty() {
            bail!("a project bundle must contain at least one repository");
        }
        let mut destinations = std::collections::BTreeSet::new();
        for repository in &self.repositories {
            validate_relative_path(&repository.destination)?;
            ensure!(
                repository
                    .url
                    .as_deref()
                    .is_some_and(|url| !url.trim().is_empty() && !url.starts_with('-')),
                "isolated repositories require a network Git remote; configure a remote or use a raw local session"
            );
            crate::remote_git::validate_network_url(
                repository.url.as_deref().expect("checked above"),
            )?;
            for push_url in &repository.push_urls {
                crate::remote_git::validate_network_url(push_url)?;
            }
            ensure!(
                repository.git_ref.is_none(),
                "git_ref is no longer supported; remove it to start from the remote's default branch"
            );
            if !destinations.insert(&repository.destination) {
                bail!(
                    "duplicate repository destination {}",
                    repository.destination
                );
            }
        }
        if !destinations.contains(&self.primary) {
            bail!("primary repository is not present in the bundle");
        }
        Ok(())
    }
}

#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum PodmanWorkspaceStorage {
    PodmanVolume,
    HostHelper {
        root: String,
        helper: Vec<String>,
    },
    #[default]
    ContainerLayer,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ContainerTemplate {
    pub image: String,
    #[serde(default)]
    pub pull_policy: ImagePullPolicy,
    #[serde(default)]
    pub extra_run_args: Vec<String>,
    #[serde(default)]
    pub workspace_storage: PodmanWorkspaceStorage,
    /// Per-target mbx build cache overrides, carried from the user
    /// configuration so cache resolution can read them off a runtime target.
    #[serde(default)]
    pub build_cache: Option<crate::config::TargetBuildCache>,
}

impl ImagePullPolicy {
    /// How fresh this target wants its image, with `Auto` read from the image
    /// reference. This is the freshness the background refresher acts on.
    pub fn resolve(self, image: &str) -> Self {
        if self != Self::Auto {
            return self;
        }
        if image_is_digest_pinned(image) {
            Self::Missing
        } else if image_is_remote(image) && image_uses_latest_tag(image) {
            Self::Newer
        } else {
            Self::Missing
        }
    }

    /// How fresh a launch insists on being. `Auto` never pulls here: the daemon
    /// refreshes remote `:latest` images on its own schedule, so a session
    /// starts from the cached image instead of blocking a launch on a
    /// multi-gigabyte download. An explicit policy still means what it says.
    pub fn at_launch(self, image: &str) -> Self {
        if self == Self::Auto {
            Self::Missing
        } else {
            self.resolve(image)
        }
    }

    /// What this policy does to `image`, in the words a settings screen can
    /// show. `Auto` is derived rather than an alias, so it names the launch
    /// behavior and the background refresh it implies, from the same image
    /// reading `resolve` uses.
    pub fn describe(self, image: &str) -> &'static str {
        match self {
            Self::Always => "Pull every launch",
            Self::Newer => "Pull when the registry is newer",
            Self::Missing => "Pull only if missing",
            Self::Never => "Never pull",
            Self::Auto => match self.resolve(image) {
                Self::Newer => "Pull if missing at launch; refresh :latest in background",
                _ => "Pull if missing",
            },
        }
    }

    /// Podman's spelling of an already-resolved policy.
    pub fn podman_value(self) -> &'static str {
        match self {
            Self::Always => "always",
            Self::Newer => "newer",
            Self::Missing => "missing",
            Self::Never => "never",
            Self::Auto => unreachable!("auto pull policy must resolve"),
        }
    }
}

/// Where a background image refresh runs.
///
/// The SSH form wraps commands the way provisioning does rather than the way
/// the preflight probes do: a pull runs for minutes, and the probes' two-second
/// keepalive would drop the connection underneath it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ImageHost {
    LocalPodman,
    LocalDocker,
    AppleContainer,
    SshPodman(SshTarget),
    SshDocker(SshTarget),
}

impl ImageHost {
    pub const fn engine(&self) -> &'static str {
        match self {
            Self::LocalPodman | Self::SshPodman(_) => "podman",
            Self::LocalDocker | Self::SshDocker(_) => "docker",
            Self::AppleContainer => "container",
        }
    }

    /// How this host is named in a log line.
    pub fn label(&self) -> String {
        match self {
            Self::LocalPodman => "local podman".to_owned(),
            Self::LocalDocker => "local docker".to_owned(),
            Self::AppleContainer => "apple container".to_owned(),
            Self::SshPodman(ssh) => format!("podman on {}", ssh.destination),
            Self::SshDocker(ssh) => format!("docker on {}", ssh.destination),
        }
    }

    fn command(&self, args: Vec<String>, purpose: String) -> CommandSpec {
        match self {
            Self::LocalPodman | Self::LocalDocker | Self::AppleContainer => {
                CommandSpec::new(args[0].clone(), args[1..].iter().cloned())
            }
            Self::SshPodman(ssh) | Self::SshDocker(ssh) => ssh_command_owned(ssh, args),
        }
        .purpose(purpose)
    }
}

/// When a background refresh downloads an image.
///
/// A host that already has the image is the common case, and most targets
/// only need the copy to exist. Ordering matters: merging two targets that
/// share an image takes the larger value, so one demanding target upgrades
/// the pair.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum RefreshWhen {
    /// Pull only when the host has no copy (missing, or auto on a fixed tag).
    WhenAbsent,
    /// Pull every tick (always, or newer / auto on a moving tag).
    Always,
}

/// The commands that keep one host's copy of one image current, away from any
/// session launch. They run in this order, and only for that host.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ImageRefresh {
    pub host: ImageHost,
    pub image: String,
    pub platform: Option<String>,
    /// Whether this image is downloaded only when the host lacks it, or on
    /// every refresh.
    pub when: RefreshWhen,
    /// Reads the cached image id, so a pull that changed nothing stays quiet.
    /// Run before and after the pull.
    pub image_id: CommandSpec,
    pub pull: CommandSpec,
    /// Dangling images only. Both engines keep an image a container still uses.
    /// `None` for Apple's `container` engine, which has no verified prune form.
    pub prune: Option<CommandSpec>,
}

/// The background refresh for one configured container target, or `None` when
/// the target asked never to download the image in the background.
///
/// `never` is the only opt-out. Every other policy at least wants the host to
/// have a copy, so the daemon downloads a missing image once; `always` and
/// `newer` keep the hourly pull they always had.
pub fn image_refresh(
    host: ImageHost,
    image: &str,
    platform: Option<&str>,
    pull_policy: ImagePullPolicy,
) -> Option<ImageRefresh> {
    let when = match pull_policy.resolve(image) {
        ImagePullPolicy::Always | ImagePullPolicy::Newer => RefreshWhen::Always,
        ImagePullPolicy::Missing => RefreshWhen::WhenAbsent,
        ImagePullPolicy::Never => return None,
        ImagePullPolicy::Auto => unreachable!("auto pull policy must resolve"),
    };
    let engine = host.engine();
    // Apple's `container` CLI has no `--format` on `image inspect`: the exit
    // status is what says the image is present, and the printed description
    // stands in for the id when comparing before and after a pull.
    let apple = matches!(host, ImageHost::AppleContainer);
    let mut image_id_args = vec![engine.to_owned(), "image".to_owned(), "inspect".to_owned()];
    if !apple {
        image_id_args.push("--format".to_owned());
        image_id_args.push("{{.Id}}".to_owned());
    }
    image_id_args.push(image.to_owned());
    let image_id = host.command(
        image_id_args,
        format!("read the cached id of container image {image}"),
    );
    let mut pull_args = vec![engine.to_owned()];
    if apple {
        pull_args.push("image".to_owned());
    }
    pull_args.push("pull".to_owned());
    // Apple's engine runs only native images, so it takes no platform.
    if let Some(platform) = platform.filter(|_| !apple) {
        pull_args.push(format!("--platform={platform}"));
    }
    pull_args.push(image.to_owned());
    let pull = host.command(pull_args, format!("refresh container image {image}"));
    let prune = (!apple).then(|| {
        host.command(
            vec![
                engine.to_owned(),
                "image".to_owned(),
                "prune".to_owned(),
                "-f".to_owned(),
            ],
            "remove dangling container images".to_owned(),
        )
    });
    Some(ImageRefresh {
        host,
        image: image.to_owned(),
        platform: platform.map(str::to_owned),
        when,
        image_id,
        pull,
        prune,
    })
}

fn image_is_digest_pinned(image: &str) -> bool {
    image
        .rsplit_once('@')
        .is_some_and(|(_, digest)| !digest.is_empty())
}

fn image_is_remote(image: &str) -> bool {
    !image.starts_with("localhost/") && !image.starts_with("local/")
}

fn image_uses_latest_tag(image: &str) -> bool {
    let name = image.split_once('@').map_or(image, |(name, _)| name);
    let final_component = name.rsplit('/').next().unwrap_or(name);
    !final_component.contains(':') || final_component.ends_with(":latest")
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SshTarget {
    pub destination: String,
    #[serde(default)]
    pub ssh_args: Vec<String>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AwsTemplate {
    pub profile: String,
    pub region: String,
    pub launch_template: String,
    pub launch_template_version: Option<String>,
    pub instance_type: Option<String>,
    pub ssh: SshTarget,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum TargetTemplate {
    LocalBare,
    LocalPodman(ContainerTemplate),
    LocalDocker(ContainerTemplate),
    AppleContainer(ContainerTemplate),
    AwsEc2(AwsTemplate),
    SshBare {
        ssh: SshTarget,
        #[serde(default = "default_ssh_prefix")]
        workspace_prefix: String,
    },
    SshPodman {
        ssh: SshTarget,
        container: ContainerTemplate,
    },
    SshDocker {
        ssh: SshTarget,
        container: ContainerTemplate,
    },
}

fn default_ssh_prefix() -> String {
    ".local/share/hel/workspaces".to_owned()
}

#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum PodmanWorkspaceLocator {
    #[default]
    ContainerLayer,
    Volume {
        name: String,
    },
    HostPath {
        path: String,
        helper: Vec<String>,
        resource: String,
    },
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum TargetLocator {
    LocalBare {
        worker_root: String,
    },
    LocalPodman {
        container_id: String,
        #[serde(default)]
        workspace_storage: PodmanWorkspaceLocator,
        /// The session that owns the container when this locator is a
        /// sub-agent child borrowing its parent's container; `None` when the
        /// session owns the container itself.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        borrowed_from: Option<String>,
    },
    LocalDocker {
        container_id: String,
        /// The session that owns the container when this locator is a
        /// sub-agent child borrowing its parent's container; `None` when the
        /// session owns the container itself.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        borrowed_from: Option<String>,
    },
    AppleContainer {
        container_id: String,
        /// The session that owns the container when this locator is a
        /// sub-agent child borrowing its parent's container; `None` when the
        /// session owns the container itself.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        borrowed_from: Option<String>,
    },
    AwsEc2 {
        profile: String,
        region: String,
        instance_id: String,
        ssh: SshTarget,
        workspace: String,
    },
    SshBare {
        ssh: SshTarget,
        workspace: String,
        /// A borrowed-target child has its own worker identity in the parent's workspace.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        worker_id: Option<String>,
    },
    SshPodman {
        ssh: SshTarget,
        container_id: String,
        #[serde(default)]
        workspace_storage: PodmanWorkspaceLocator,
        /// The session that owns the container when this locator is a
        /// sub-agent child borrowing its parent's container; `None` when the
        /// session owns the container itself.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        borrowed_from: Option<String>,
    },
    SshDocker {
        ssh: SshTarget,
        container_id: String,
        /// The session that owns the container when this locator is a
        /// sub-agent child borrowing its parent's container; `None` when the
        /// session owns the container itself.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        borrowed_from: Option<String>,
    },
}

impl TargetTemplate {
    pub const fn container_engine(&self) -> Option<&'static str> {
        match self {
            Self::LocalPodman(_) | Self::SshPodman { .. } => Some("podman"),
            Self::LocalDocker(_) | Self::SshDocker { .. } => Some("docker"),
            Self::AppleContainer(_) => Some("container"),
            _ => None,
        }
    }

    /// The host that downloads this target's image, with the container
    /// settings that name the image. `None` for targets that run no image.
    pub fn image_host(&self) -> Option<(ImageHost, &ContainerTemplate)> {
        match self {
            Self::LocalPodman(container) => Some((ImageHost::LocalPodman, container)),
            Self::LocalDocker(container) => Some((ImageHost::LocalDocker, container)),
            Self::AppleContainer(container) => Some((ImageHost::AppleContainer, container)),
            Self::SshPodman { ssh, container } => {
                Some((ImageHost::SshPodman(ssh.clone()), container))
            }
            Self::SshDocker { ssh, container } => {
                Some((ImageHost::SshDocker(ssh.clone()), container))
            }
            Self::LocalBare | Self::AwsEc2(_) | Self::SshBare { .. } => None,
        }
    }
}

impl TargetLocator {
    /// The target kind spelling shared with [`crate::config::TargetTemplate`].
    pub const fn kind_name(&self) -> &'static str {
        match self {
            Self::LocalBare { .. } => "local-bare",
            Self::LocalPodman { .. } => "local-podman",
            Self::LocalDocker { .. } => "local-docker",
            Self::AppleContainer { .. } => "apple-container",
            Self::AwsEc2 { .. } => "aws-ec2",
            Self::SshBare { .. } => "ssh-bare",
            Self::SshPodman { .. } => "ssh-podman",
            Self::SshDocker { .. } => "ssh-docker",
        }
    }

    pub const fn container_engine(&self) -> Option<&'static str> {
        match self {
            Self::LocalPodman { .. } | Self::SshPodman { .. } => Some("podman"),
            Self::LocalDocker { .. } | Self::SshDocker { .. } => Some("docker"),
            Self::AppleContainer { .. } => Some("container"),
            _ => None,
        }
    }
}

/// Commands and identity needed to bring a stopped managed target back online.
/// Only runtimes whose stopped resources retain their durable files provide
/// one; callers leave every other target kind alone.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TargetRecoveryPlan {
    pub exists: CommandSpec,
    pub inspect: CommandSpec,
    pub start: CommandSpec,
    pub session_id: String,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TargetRecoveryOutcome {
    NotRequired,
    Missing,
    AlreadyRunning,
    Started,
}

pub fn resource_name(session_id: &str) -> Result<String> {
    validate_session_id(session_id)?;
    let readable: String = session_id
        .chars()
        .filter(|character| character.is_ascii_alphanumeric())
        .take(12)
        .map(|character| character.to_ascii_lowercase())
        .collect();
    let digest = Sha256::digest(session_id.as_bytes());
    Ok(format!(
        "mj-{readable}-{:02x}{:02x}{:02x}",
        digest[0], digest[1], digest[2]
    ))
}

/// Resource generations let a Move retain its source while provisioning the destination.
pub fn move_resource_name(session_id: &str, operation_id: &str) -> Result<String> {
    let digest = Sha256::digest(operation_id.as_bytes());
    Ok(format!(
        "{}-move-{}",
        resource_name(session_id)?,
        crate::hex::lower_hex(&digest[..8])
    ))
}

pub fn resource_name_belongs_to(name: &str, session_id: &str) -> Result<bool> {
    let base = resource_name(session_id)?;
    Ok(name == base
        || name
            .strip_prefix(&format!("{base}-move-"))
            .is_some_and(|suffix| {
                suffix.len() == 16 && suffix.bytes().all(|b| b.is_ascii_hexdigit())
            }))
}

pub fn podman_workspace_locator(
    template: &ContainerTemplate,
    session_id: &str,
) -> Result<PodmanWorkspaceLocator> {
    podman_workspace_locator_named(template, &resource_name(session_id)?)
}

pub fn podman_workspace_locator_named(
    template: &ContainerTemplate,
    name: &str,
) -> Result<PodmanWorkspaceLocator> {
    let resource = format!("{name}-workspace");
    match &template.workspace_storage {
        PodmanWorkspaceStorage::PodmanVolume => {
            Ok(PodmanWorkspaceLocator::Volume { name: resource })
        }
        PodmanWorkspaceStorage::HostHelper { root, helper } => {
            let root = Path::new(root);
            ensure!(
                root.is_absolute(),
                "Podman workspace storage root must be absolute"
            );
            ensure!(
                !helper.is_empty() && helper.iter().all(|argument| !argument.is_empty()),
                "Podman workspace storage helper must contain non-empty arguments"
            );
            Ok(PodmanWorkspaceLocator::HostPath {
                path: root.join(&resource).to_string_lossy().into_owned(),
                helper: helper.clone(),
                resource,
            })
        }
        PodmanWorkspaceStorage::ContainerLayer => Ok(PodmanWorkspaceLocator::ContainerLayer),
    }
}

/// The in-container workspace root a session's repositories live under.
///
/// `recorded` is the session record's `container_workspace`. A session whose
/// container predates per-session workspaces has none and keeps the legacy
/// shared `/workspace`; every session created since records
/// `/workspace/<session id>`, so a build cache shared by every container on a
/// host never sees two checkouts of one project at the same absolute path.
pub fn container_workspace_root(recorded: Option<&Path>) -> String {
    recorded.map_or_else(
        || CONTAINER_WORKSPACE.to_owned(),
        |path| path.to_string_lossy().into_owned(),
    )
}

/// The per-session container workspace recorded for a session created now.
pub fn new_container_workspace(session_id: &str) -> Result<PathBuf> {
    validate_session_id(session_id)?;
    Ok(Path::new(CONTAINER_WORKSPACE).join(session_id))
}

/// The workspace directory an EC2 session owns in its login home.
///
/// EC2 instances are provisioned per session, so the path is decided by the
/// session id alone and is the same whether it is read from a target template
/// or rebuilt from a stored locator.
pub fn aws_workspace(session_id: &str) -> String {
    format!(".local/share/hel/workspaces/{session_id}")
}

pub fn workspace_for(template: &TargetTemplate, session_id: &str) -> Result<String> {
    validate_session_id(session_id)?;
    match template {
        TargetTemplate::LocalBare => bail!("local bare projects use their selected directory"),
        // A container workspace is per session and recorded on the session
        // record, so it is read with `container_workspace_root` instead.
        TargetTemplate::LocalPodman(_)
        | TargetTemplate::LocalDocker(_)
        | TargetTemplate::AppleContainer(_)
        | TargetTemplate::SshPodman { .. }
        | TargetTemplate::SshDocker { .. } => {
            bail!("container targets use the session's recorded container workspace")
        }
        TargetTemplate::AwsEc2(_) => Ok(aws_workspace(session_id)),
        TargetTemplate::SshBare {
            workspace_prefix, ..
        } => {
            validate_workspace_prefix(workspace_prefix)?;
            // Interpret a leading "~/" as home-relative. Remote commands are
            // single-quoted, so a literal tilde would name a directory called
            // "~"; a relative path resolves against the login home for ssh
            // and scp alike.
            let prefix = workspace_prefix
                .strip_prefix("~/")
                .unwrap_or(workspace_prefix);
            Ok(format!("{}/{session_id}", prefix.trim_end_matches('/')))
        }
    }
}

/// Wrap an argv vector for execution at a provisioned session target.
pub fn command_on_locator(
    locator: &TargetLocator,
    session_id: &str,
    args: Vec<String>,
    purpose: impl Into<String>,
) -> Result<CommandSpec> {
    verify_locator(locator, session_id)?;
    if args.is_empty() {
        bail!("target command must not be empty");
    }
    Ok(locator_command(locator, args).purpose(purpose))
}

/// Wrap a non-empty argv vector for execution at a target, without checking
/// that the locator belongs to a particular session. Every per-target command
/// is built here, so each target kind is spelled out once.
pub fn locator_command(locator: &TargetLocator, args: Vec<String>) -> CommandSpec {
    match locator {
        TargetLocator::LocalBare { .. } => {
            let mut args = args.into_iter();
            let program = args.next().expect("target command must not be empty");
            CommandSpec::new(program, args)
        }
        TargetLocator::LocalPodman { container_id, .. }
        | TargetLocator::LocalDocker { container_id, .. }
        | TargetLocator::AppleContainer { container_id, .. } => container_exec(
            locator.container_engine().expect("local container"),
            container_id,
            args,
        ),
        TargetLocator::AwsEc2 { ssh, .. } | TargetLocator::SshBare { ssh, .. } => {
            ssh_command_owned(ssh, args)
        }
        TargetLocator::SshPodman {
            ssh, container_id, ..
        }
        | TargetLocator::SshDocker {
            ssh, container_id, ..
        } => {
            let mut remote = vec![
                locator
                    .container_engine()
                    .expect("remote container")
                    .to_owned(),
                "exec".to_owned(),
                "-i".to_owned(),
                container_id.to_owned(),
            ];
            remote.extend(args);
            ssh_command_owned(ssh, remote)
        }
    }
}
pub fn worker_root(locator: &TargetLocator, session_id: &str) -> Result<String> {
    verify_locator(locator, session_id)?;
    Ok(match locator {
        TargetLocator::LocalBare { worker_root } => worker_root.clone(),
        TargetLocator::LocalPodman { .. }
        | TargetLocator::LocalDocker { .. }
        | TargetLocator::AppleContainer { .. }
        | TargetLocator::SshPodman { .. }
        | TargetLocator::SshDocker { .. } => format!("/var/lib/hel/workers/{session_id}"),
        TargetLocator::AwsEc2 { .. } => format!(".local/share/hel/workers/{session_id}"),
        TargetLocator::SshBare { worker_id, .. } => format!(
            ".local/share/hel/workers/{}",
            worker_id.as_deref().unwrap_or(session_id)
        ),
    })
}
mod convert;
pub use convert::{
    RecordedTarget, StoredTarget, TargetConversionError, locator_needs_connection,
    ssh_args_with_identity,
};

mod ssh;
pub use ssh::*;

pub fn container_exec(
    engine: &str,
    container_id: &str,
    args: impl IntoIterator<Item = impl Into<String>>,
) -> CommandSpec {
    let mut command_args = vec!["exec".to_owned(), "-i".to_owned(), container_id.to_owned()];
    command_args.extend(args.into_iter().map(Into::into));
    CommandSpec::new(engine, command_args)
}

#[cfg(all(test, unix))]
mod executor_tests {
    use std::fs;

    use super::*;

    #[cfg(target_os = "linux")]
    #[test]
    fn successful_commands_stop_descendants_that_closed_their_pipes() {
        for streamed in [false, true] {
            let temp = tempfile::tempdir().unwrap();
            let pid_file = temp.path().join("descendant");
            let command = CommandSpec::new("sh", [
                "-c".to_owned(),
                "cat >/dev/null; head -c 131072 /dev/zero; sleep 60 </dev/null >/dev/null 2>&1 & echo $! > \"$1\"".to_owned(),
                "owned-descendant".to_owned(), pid_file.display().to_string(),
            ]);
            let executor = CancellableProcessExecutor::with_timeout(Duration::from_secs(5));
            let output = if streamed {
                executor
                    .execute_with_stdin(&command, &mut std::io::Cursor::new(vec![b'x'; 256 * 1024]))
            } else {
                executor.execute(&command)
            }
            .unwrap();
            assert_eq!(output.stdout.len(), 131072);
            let pid: i32 = fs::read_to_string(pid_file)
                .unwrap()
                .trim()
                .parse()
                .unwrap();
            let deadline = Instant::now() + Duration::from_secs(2);
            loop {
                let state = fs::read_to_string(format!("/proc/{pid}/stat")).ok();
                if state.as_ref().is_none_or(|state| {
                    state
                        .rsplit_once(") ")
                        .is_some_and(|(_, fields)| fields.starts_with('Z'))
                }) {
                    break;
                }
                if Instant::now() >= deadline {
                    // Do not leave the fixture behind when the regression fails.
                    unsafe {
                        libc::kill(pid, libc::SIGKILL);
                    }
                    panic!(
                        "successful command left descendant {pid} running (streamed={streamed})"
                    );
                }
                std::thread::sleep(Duration::from_millis(10));
            }
        }
    }

    /// B-2 at the executor: a backgrounded descendant that holds the output
    /// pipes must not hold the completed command past the drain.
    #[cfg(target_os = "linux")]
    #[test]
    fn commands_complete_at_leader_exit_when_a_descendant_holds_the_pipes() {
        for streamed in [false, true] {
            let temp = tempfile::tempdir().unwrap();
            let pid_file = temp.path().join("descendant");
            let command = CommandSpec::new(
                "sh",
                [
                    "-c".to_owned(),
                    "sleep 300 & echo $! > \"$1\"; echo hi; exit 3".to_owned(),
                    "held-pipes".to_owned(),
                    pid_file.display().to_string(),
                ],
            );
            let executor = CancellableProcessExecutor::with_timeout(Duration::from_secs(30));
            let started = Instant::now();
            let output = if streamed {
                executor.execute_with_stdin(&command, &mut std::io::Cursor::new(b"input".to_vec()))
            } else {
                executor.execute(&command)
            }
            .unwrap();
            assert!(
                started.elapsed() < Duration::from_secs(5),
                "streamed={streamed} took {:?}",
                started.elapsed()
            );
            assert_eq!(output.status, 3);
            assert_eq!(output.stdout, b"hi\n");
            let pid: i32 = fs::read_to_string(pid_file)
                .unwrap()
                .trim()
                .parse()
                .unwrap();
            let deadline = Instant::now() + Duration::from_secs(2);
            loop {
                let state = fs::read_to_string(format!("/proc/{pid}/stat")).ok();
                if state.as_ref().is_none_or(|state| {
                    state
                        .rsplit_once(") ")
                        .is_some_and(|(_, fields)| fields.starts_with('Z'))
                }) {
                    break;
                }
                if Instant::now() >= deadline {
                    // SAFETY: cleans up this test's own fixture.
                    unsafe {
                        libc::kill(pid, libc::SIGKILL);
                    }
                    panic!("completed command left descendant {pid} (streamed={streamed})");
                }
                std::thread::sleep(Duration::from_millis(10));
            }
        }
    }

    #[test]
    fn streamed_deadline_survives_leader_exit_and_inherited_pipes() {
        let command = CommandSpec::new(
            "sh",
            [
                "-c",
                "head -c 131072 /dev/zero; cat >/dev/null; (trap '' TERM; sleep 60) & exit 0",
            ],
        );
        let mut input = std::io::Cursor::new(vec![b'x'; 256 * 1024]);
        let started = Instant::now();
        let error = CancellableProcessExecutor::with_timeout(Duration::from_millis(300))
            .execute_with_stdin(&command, &mut input)
            .unwrap_err();
        assert!(error.to_string().contains("cancelled"), "{error:#}");
        assert!(started.elapsed() < Duration::from_secs(5));
    }

    /// A stand-in for `ssh` that is refused by the server on its first call and
    /// connects on the next, the way a host at its `MaxStartups` ceiling
    /// behaves once the daemon's burst drains.
    fn flaky_ssh_script(directory: &Path) -> CommandSpec {
        let counter = directory.join("attempts");
        let script = format!(
            "count=$(cat {counter} 2>/dev/null || echo 0)\n\
             echo $((count + 1)) > {counter}\n\
             if [ \"$count\" -eq 0 ]; then\n\
             echo 'kex_exchange_identification: Connection closed by 10.0.0.1 port 22' >&2\n\
             exit 255\n\
             fi\n\
             echo connected\n",
            counter = counter.display()
        );
        CommandSpec::new("sh", ["-c".to_owned(), script])
            .ssh_destination("build@10.0.0.1")
            .purpose("run the flaky SSH fixture")
    }

    fn attempts(directory: &Path) -> u32 {
        fs::read_to_string(directory.join("attempts"))
            .expect("the fixture records its attempts")
            .trim()
            .parse()
            .expect("attempt count is a number")
    }

    /// An executor that models `ssh` masters and the sessions on them. The
    /// first session finds its master dead (it was killed after the ledger
    /// last checked it), which `ssh` reports the way a session guarded with
    /// `ProxyCommand=false` does.
    #[derive(Default)]
    struct MasterKilledOnce {
        running: std::cell::RefCell<BTreeSet<String>>,
        openers: std::cell::Cell<usize>,
        sessions: std::cell::RefCell<Vec<Vec<String>>>,
    }

    impl CommandExecutor for MasterKilledOnce {
        fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
            let reply = |status: i32, stderr: &str| CommandOutput {
                status,
                stdout: Vec::new(),
                stderr: stderr.as_bytes().to_vec(),
            };
            if command.ssh_session.is_some() {
                // A session. The first one arrives just after its master died.
                return with_ssh_admission(command, self, &|| false, |spawned| {
                    self.sessions.borrow_mut().push(spawned.args.clone());
                    if self.sessions.borrow().len() == 1 {
                        self.running.borrow_mut().clear();
                        return Ok(reply(255, "Connection closed by UNKNOWN port 65535"));
                    }
                    Ok(reply(0, ""))
                });
            }
            let socket = command
                .args
                .iter()
                .find_map(|arg| arg.strip_prefix("ControlPath="))
                .expect("a master command names its socket")
                .to_owned();
            if command.args.windows(2).any(|pair| pair == ["-O", "check"]) {
                return Ok(reply(
                    if self.running.borrow().contains(&socket) {
                        0
                    } else {
                        255
                    },
                    "",
                ));
            }
            assert!(command.args.contains(&"ControlMaster=yes".to_owned()));
            self.openers.set(self.openers.get() + 1);
            self.running.borrow_mut().insert(socket);
            Ok(reply(0, ""))
        }
    }

    #[test]
    fn a_session_whose_master_died_is_retried_on_a_reopened_master() {
        let _guard = ssh::SHARING_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        set_ssh_retry_backoff_for_test(Some(Duration::from_millis(5)));
        let socket_dir = tempfile::tempdir_in("/tmp").expect("short socket directory");
        set_ssh_connection_sharing_for_test(Some(SshSharingForTest::Directory(
            socket_dir.path().to_path_buf(),
        )));
        let ssh = SshTarget {
            destination: "master-killed-once-host".to_owned(),
            ssh_args: Vec::new(),
        };
        let executor = MasterKilledOnce::default();
        let output = executor.execute(&ssh_command(&ssh, ["true"]));
        set_ssh_connection_sharing_for_test(None);
        set_ssh_retry_backoff_for_test(None);

        assert_eq!(output.expect("the retry succeeds").status, 0);
        let sessions = executor.sessions.borrow();
        assert_eq!(sessions.len(), 2);
        assert_eq!(
            executor.openers.get(),
            2,
            "the retry reopens the master instead of trusting the earlier check"
        );
        for args in sessions.iter() {
            assert_eq!(args[..6][5], "ProxyCommand=false", "{args:?}");
        }
    }

    #[test]
    fn a_transport_rejected_ssh_command_is_retried_once_and_then_succeeds() {
        set_ssh_retry_backoff_for_test(Some(Duration::from_millis(5)));
        let directory = tempfile::tempdir().expect("temp dir");
        let command = flaky_ssh_script(directory.path());

        let output = ProcessExecutor
            .execute(&command)
            .expect("the retry must reach the successful attempt");

        assert_eq!(output.status, 0);
        assert_eq!(String::from_utf8_lossy(&output.stdout).trim(), "connected");
        assert_eq!(attempts(directory.path()), 2);
        set_ssh_retry_backoff_for_test(None);
    }

    #[test]
    fn an_untagged_command_is_not_retried_after_the_same_failure() {
        set_ssh_retry_backoff_for_test(Some(Duration::from_millis(5)));
        let directory = tempfile::tempdir().expect("temp dir");
        let mut command = flaky_ssh_script(directory.path());
        command.ssh_destination = None;

        let output = ProcessExecutor.execute(&command).expect("runs once");

        assert_eq!(output.status, 255);
        assert_eq!(attempts(directory.path()), 1);
        set_ssh_retry_backoff_for_test(None);
    }

    #[test]
    fn the_cancellable_executor_also_retries_a_transport_rejection() {
        set_ssh_retry_backoff_for_test(Some(Duration::from_millis(5)));
        let directory = tempfile::tempdir().expect("temp dir");
        let command = flaky_ssh_script(directory.path());

        let output = CancellableProcessExecutor::new(Arc::new(AtomicBool::new(false)))
            .execute(&command)
            .expect("the retry must reach the successful attempt");

        assert_eq!(output.status, 0);
        assert_eq!(attempts(directory.path()), 2);
        set_ssh_retry_backoff_for_test(None);
    }
}