haz_exec/

run_task.rs

//! Single-task lifecycle: cache lookup, hit-path restoration, or
//! miss-path spawn-and-capture with success-only store.
//!
//! [`run_task`] is the async entry point. It does no scheduling, no
//! cap accounting, no mutex acquisition, no cancellation, and emits
//! no presentation-layer bytes itself: captured streams flow to the
//! caller through the [`RunObserver`] trait. Concrete observer
//! implementations distinguish the two `EXEC-016` modes: `live`
//! (per-line `[project:task] `-tag-prefixed writes to a shared
//! parent sink) and `buffered` (one contiguous block per stream on
//! task completion).
//!
//! The function composes the cache-key derivation
//! ([`haz_cache_key::build_cache_key`]) with the
//! [`crate::process::ProcessSpawner`] trait
//! into the executor's smallest
//! end-to-end unit. A scheduler layer combines many [`run_task`]
//! calls with the workspace's concurrency caps and mutex hold-set
//! (`EXEC-004..007`).
//!
//! # Spec coverage
//!
//! - `EXEC-007` step 1 -- key derivation and lookup precede every
//!   other concern. Mutex acquisition is not implemented by this
//!   function; that responsibility belongs to a scheduler layer.
//! - `EXEC-007` step 2 -- cache-hit restoration runs mutex-free.
//! - `EXEC-009` -- terminal classification: zero exit is
//!   [`RunState::Succeeded`]; non-zero or signalled is
//!   [`RunState::Failed`]. A cache hit is [`RunState::Succeeded`].
//!   The cancelled state of `EXEC-009`'s third option is not
//!   reachable from this function: cancellation is not implemented
//!   here.
//! - `EXEC-016` -- the full `stdout` and `stderr` byte streams of
//!   every fresh run are captured (the cache stores them per
//!   `CACHE-012`) and surfaced through [`RunObserver`].
//! - `EXEC-017` -- on a cache hit, the recorded `stdout` and
//!   `stderr` flow through [`RunObserver`] with the same
//!   mode-agnostic signal as a fresh run.
//! - `CACHE-008` (runtime) -- the spawned process sees only the
//!   effective env: allow-listed host values plus task-level
//!   overrides, override wins on collision. The spawn-plan
//!   builder assembles that env vector; the std backend's
//!   `env_clear()` ensures the child sees this set and nothing
//!   else.
//! - `CACHE-014..016` -- [`haz_cache::CacheReader::lookup`] folds every
//!   failure into a clean miss, so no [`RunTaskError`] variant
//!   covers it.
//! - `CACHE-017..018` -- the cache store fires only when the fresh
//!   run's exit status was zero. The store call itself surfaces
//!   as [`RunTaskError::StoreFailed`].
//! - `CACHE-019` -- restoration uses [`haz_cache::CacheWriter::restore`],
//!   surfacing [`haz_cache::RestoredStreams`] through the observer.
//!
//! # Out of scope for this function
//!
//! - Concurrency caps (`EXEC-004..006`) and the canonical-order
//!   tie-breaking of `EXEC-003`: scheduler-level concerns.
//! - Mutex acquisition (`EXEC-007` step 3) and `EXEC-006`
//!   condition 3: scheduler-level concerns.
//! - Failure cascade and the `skipped` state (`EXEC-009..011`):
//!   scheduler-level concerns.
//! - Cancellation (`EXEC-012..015`): not implemented in this
//!   function.
//! - Live-mode per-line tag prefixing and atomic sink writes
//!   (`EXEC-016` presentation): implemented inside concrete
//!   [`RunObserver`] types.
//! - Runtime cycle (`EXEC-019`) and output-overlap (`EXEC-020`)
//!   detection: scheduler-level concerns.

use std::collections::BTreeMap;
use std::ffi::OsString;
use std::io;
use std::path::{Path, PathBuf};

use snafu::{ResultExt, Snafu};
use tokio::io::AsyncReadExt;
use tokio_util::sync::CancellationToken;

use haz_cache::{CacheWriter, Hasher, RestoreError, StoreError, StoreInputs, StoredOutput};
use haz_dag::graph::TaskGraph;
use haz_domain::action::{ShellType, TaskAction};
use haz_domain::env::{EnvSettings, EnvVarName};
use haz_domain::path::{CanonicalPath, OutputSpec, ParseAbsoluteError, PathPattern, ProjectRoot};
use haz_domain::project::Project;
use haz_domain::settings::cache::HashAlgo;
use haz_domain::task::Task;
use haz_domain::task_id::TaskId;
use haz_domain::workspace::Workspace;
use haz_vfs::{EntryKind, Filesystem, FsError, WritableFilesystem};

use crate::process::{
    ExitStatus, Process, ProcessError, ProcessSpawner, Signal, SpawnPlan, Spawned,
};
use haz_cache_key::pattern_walk::{
    GlobMatchAction, GlobWalk, glob_walk_origin, host_path_from_segments,
    literal_workspace_segments, workspace_absolute_string_from_segments,
};
use haz_cache_key::{BuildKeyError, PredecessorStreamHashes, build_cache_key};

/// Where a [`CompletedRecord`]'s success came from.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum RunSource {
    /// The cache held a valid entry for the derived key
    /// (`CACHE-015`); restoration materialised the outputs and the
    /// recorded streams were surfaced through the observer.
    CacheHit,
    /// The cache missed; the executor spawned the task's command,
    /// captured its stdout and stderr, and (on success) stored the
    /// result for future hits.
    FreshRun,
}

/// Terminal classification of a single-task run per `EXEC-009`.
///
/// The variants exhaustively cover the spec's three terminal
/// classifications: succeeded, failed, and cancelled-by-executor.
/// The cancelled variant is produced only by the spawn-step
/// future when the run-context's cancellation token fires while
/// the child is in flight; until that wiring lands no
/// [`run_task`] call returns [`RunState::Cancelled`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum RunState {
    /// A fresh run with exit status zero, or a cache hit (a hit
    /// asserts a prior successful run per `CACHE-018`).
    Succeeded,
    /// A fresh run whose command terminated with a non-zero exit
    /// status or whose process was killed by a signal not initiated
    /// by the executor.
    Failed,
    /// A fresh run whose child was signalled by the executor in
    /// response to the run's cancellation token firing
    /// (`EXEC-009` cancelled state, `EXEC-013` step 2). Distinct
    /// from [`Self::Failed`] so the run summary can tell
    /// executor-initiated termination apart from a task that
    /// failed under its own power.
    Cancelled,
}

/// Run-record fields produced by one [`run_task`] invocation: the
/// observation of a task that the scheduler admitted into the
/// lookup-then-spawn pipeline and that reached a run classification
/// (`EXEC-009`).
///
/// Carries enough information for a scheduler layer to feed the
/// per-`TaskId` `predecessor_streams` map that
/// [`haz_cache_key::build_cache_key`] consumes: `stdout_hash` and
/// `stderr_hash` are always present, whether the run was a cache
/// hit (sourced from the manifest's recorded hashes per
/// `CACHE-011`) or a fresh run (computed from the captured byte
/// buffers under the cache's active [`HashAlgo`]).
///
/// A scheduler-level cascade decision produces
/// [`RunOutcome::Skipped`] instead (`EXEC-010` / `EXEC-011`);
/// [`run_task`] itself never produces a Skipped because skipped
/// tasks never enter the pipeline.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CompletedRecord {
    /// The task this record describes.
    pub task: TaskId,
    /// Whether the success came from a cache hit or a fresh run.
    pub source: RunSource,
    /// Terminal classification per `EXEC-009`.
    pub state: RunState,
    /// Process exit status for a fresh run, [`None`] for a cache
    /// hit (no process ran).
    pub exit_status: Option<ExitStatus>,
    /// Hash of the run's captured stdout under the cache's
    /// [`HashAlgo`]. For [`RunSource::CacheHit`] this is read from
    /// the manifest; for [`RunSource::FreshRun`] it is computed
    /// from the captured byte buffer.
    pub stdout_hash: [u8; 32],
    /// Hash of the run's captured stderr under the cache's
    /// [`HashAlgo`].
    pub stderr_hash: [u8; 32],
    /// Workspace-absolute paths the task materialised on disk.
    ///
    /// Populated only for [`RunState::Succeeded`]; empty for
    /// [`RunState::Failed`] and [`RunState::Cancelled`]. The
    /// source is the executor's output-resolution pass for a
    /// fresh run and [`haz_cache::Manifest::outputs`] for a
    /// cache hit.
    ///
    /// Consumed by the scheduler's runtime DAG validation pass
    /// (`EXEC-019` runtime cycle detection and `EXEC-020` runtime
    /// output-overlap detection) to derive runtime producer-
    /// matching edges and to populate the output-claim tracker.
    pub materialised_outputs: Vec<CanonicalPath>,
}

/// Record of a task that the scheduler cascade-skipped because an
/// upstream task failed (`EXEC-010` / `EXEC-011`).
///
/// A skipped task is never started by the executor and never
/// produces a cache entry. The `cause` field names the root
/// failing task (NOT an intermediate cascade-skipped predecessor),
/// so the run summary can attribute every skip to a single
/// actually-failed task.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SkipRecord {
    /// The task that was skipped.
    pub task: TaskId,
    /// Reason the task was skipped, naming the originating
    /// failure.
    pub cause: SkipCause,
}

/// Reason a task was cascade-skipped per `EXEC-011`.
///
/// The `upstream` field on each variant carries the root failing
/// task identity, not the immediate hard predecessor: when `A`
/// fails and the cascade marks `B` (direct hard child) and `C`
/// (hard child of `B`), both record `upstream = A`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SkipCause {
    /// A transitively-upstream task reached
    /// [`RunState::Failed`] (per `EXEC-009`: non-zero exit,
    /// non-executor signal, or executor-imposed timeout).
    UpstreamFailed {
        /// Identity of the originally-failed upstream task.
        upstream: TaskId,
    },
    /// A transitively-upstream task surfaced a [`RunTaskError`]
    /// (executor-level failure: cache derive, spawn, wait,
    /// stream-read, output-resolution, or store). Distinct from
    /// [`Self::UpstreamFailed`] so the run summary can tell
    /// "the upstream task ran and failed" from "the executor
    /// could not even run the upstream task".
    UpstreamErrored {
        /// Identity of the upstream task whose [`run_task`] call
        /// returned an [`Err`].
        upstream: TaskId,
    },
    /// The task was identified as a member of a runtime cycle
    /// detected by `EXEC-019`. The scheduler stops admitting
    /// further tasks once the cycle is observed; cycle members
    /// still in the ready set never start. The cycle's full node
    /// set and the offending edge appear in the run's
    /// [`crate::run_graph::RunGraphOutcome::invariant_violations`]
    /// diagnostic; this per-task cause attributes the skip to
    /// that workspace-level event without rewriting the cycle
    /// node's outcome to a misleading "failed" classification.
    RuntimeCycle,
}

/// Record of a task that ended in the executor-initiated
/// cancelled state per `EXEC-009` / `EXEC-013`.
///
/// The three variants distinguish the three structural shapes a
/// cancellation can take in the run-graph view:
///
/// - [`Self::SignaledInFlight`] is the only shape a single-task
///   spawn-step future can produce: the cancellation token fired
///   while the child was running, and the executor sent SIGTERM
///   (and possibly SIGKILL after `EXEC-014`'s grace period).
/// - [`Self::UpstreamCancelled`] is the cascade counterpart to
///   [`SkipCause::UpstreamFailed`]: a hard descendant of a
///   cancelled task is itself cancelled. `upstream` carries the
///   root cancelled task, not the immediate hard predecessor
///   (same root-cause attribution as [`SkipRecord`]).
/// - [`Self::RunCancelled`] covers tasks that were still in
///   `state.ready` (or whose lookup-step was in flight) when the
///   scheduler observed the cancellation signal and drained the
///   admission front. These tasks never made it to the spawn
///   step; no process was started or signalled.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CancelledRecord {
    /// The task was admitted, spawned, and signalled by the
    /// executor in response to the cancellation token firing.
    /// `exit_status` records the child's exit (typically
    /// signal-terminated, but a polite child may exit cleanly
    /// after SIGTERM); the captured stream hashes are present
    /// for diagnostic purposes even though `EXEC-015` blocks
    /// the cache store.
    SignaledInFlight {
        /// The task that was signalled.
        task: TaskId,
        /// The child's exit status after the signal flow ran to
        /// completion.
        exit_status: ExitStatus,
        /// Hash of captured stdout under the cache's
        /// [`HashAlgo`], for diagnostic / predecessor-streams use.
        stdout_hash: [u8; 32],
        /// Hash of captured stderr under the cache's
        /// [`HashAlgo`], for diagnostic / predecessor-streams use.
        stderr_hash: [u8; 32],
    },
    /// The task was cascade-cancelled because a transitively-
    /// upstream task entered the cancelled state. Mirrors
    /// [`SkipCause::UpstreamFailed`] for the cancellation flow.
    UpstreamCancelled {
        /// The task that was cascade-cancelled.
        task: TaskId,
        /// The root cancelled task that triggered the cascade.
        upstream: TaskId,
    },
    /// The task was in `state.ready` (or its lookup-step was in
    /// flight) when the scheduler observed the cancellation
    /// signal. No process was spawned; the task simply never
    /// entered the spawn-step pipeline.
    RunCancelled {
        /// The task whose admission was drained on cancel.
        task: TaskId,
    },
}

impl CancelledRecord {
    /// The task this record describes, available on every
    /// variant.
    #[must_use]
    pub fn task(&self) -> &TaskId {
        match self {
            Self::SignaledInFlight { task, .. }
            | Self::UpstreamCancelled { task, .. }
            | Self::RunCancelled { task } => task,
        }
    }
}

/// Terminal state of one task in a [`crate::run_graph`] invocation.
///
/// `Completed` wraps a [`CompletedRecord`] for any task that the
/// scheduler admitted into the lookup-then-spawn pipeline and
/// that reached a run classification (`EXEC-009`); `Skipped`
/// wraps a [`SkipRecord`] for any task the cascade marked
/// do-not-schedule before admission (`EXEC-010` / `EXEC-011`);
/// `Cancelled` wraps a [`CancelledRecord`] for any task that
/// the cancellation flow (`EXEC-012..015`) reached, either
/// directly (signalled in flight, drained from the ready set)
/// or via cascade.
///
/// [`run_task`] itself returns the bare [`CompletedRecord`]
/// (never wrapped in this enum) because a single-task lifecycle
/// has no cascade to inspect.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RunOutcome {
    /// A task that ran to a [`CompletedRecord`] (succeeded or
    /// failed per `EXEC-009`).
    Completed(CompletedRecord),
    /// A task the scheduler cascade-skipped before admission.
    Skipped(SkipRecord),
    /// A task the cancellation flow caught at any point (in
    /// flight, on the ready set, or via cascade).
    Cancelled(CancelledRecord),
}

impl RunOutcome {
    /// The task this outcome describes, available on every
    /// variant.
    #[must_use]
    pub fn task(&self) -> &TaskId {
        match self {
            Self::Completed(record) => &record.task,
            Self::Skipped(record) => &record.task,
            Self::Cancelled(record) => record.task(),
        }
    }
}

/// Observation surface for one [`run_task`] invocation.
///
/// The four callbacks define a mode-agnostic surface: byte chunks
/// flow through [`RunObserver::on_stdout`] and
/// [`RunObserver::on_stderr`] for both fresh runs and cache hits.
/// Concrete implementations distinguish the `EXEC-016` presentation
/// modes:
///
/// - A `buffered` observer collects the chunks per task and emits
///   one contiguous block to the parent process on
///   [`RunObserver::on_task_finished`].
/// - A `live` observer prefixes each line with `[project:task] `
///   and writes it to a shared sink with per-line atomicity.
///
/// Implementations MAY use interior mutability; the trait methods
/// take `&self` so that one observer can serve many concurrent
/// [`run_task`] calls (the scheduler exercises this via concurrent
/// invocations).
pub trait RunObserver {
    /// Called once at the start of the run, before any cache work.
    fn on_task_started(&self, task: &TaskId);

    /// One chunk of captured stdout bytes. For a cache hit, called
    /// exactly once with the recorded stdout in full. For a fresh
    /// run, the call shape depends on the observer's presentation
    /// strategy: a buffered observer typically receives a single
    /// call with the whole captured stdout; a live observer may
    /// receive many sub-line chunks as bytes flow from the pipe.
    fn on_stdout(&self, task: &TaskId, bytes: &[u8]);

    /// One chunk of captured stderr bytes. Same call shape as
    /// [`Self::on_stdout`].
    fn on_stderr(&self, task: &TaskId, bytes: &[u8]);

    /// Called once at the end of the run with the run-record
    /// classification. Fires only for tasks the scheduler
    /// admitted into the lookup-then-spawn pipeline; cascade-
    /// skipped tasks surface through
    /// [`Self::on_task_skipped`] instead.
    fn on_task_finished(&self, task: &TaskId, record: &CompletedRecord);

    /// Called once at the moment the scheduler marks `task`
    /// cascade-skipped per `EXEC-010` / `EXEC-011`. Fires before
    /// the next admission round; NEVER paired with
    /// [`Self::on_task_started`] or [`Self::on_task_finished`]
    /// for the same task.
    fn on_task_skipped(&self, task: &TaskId, record: &SkipRecord);

    /// Called once at the moment the cancellation flow records a
    /// terminal state for `task` per `EXEC-012..015`. The
    /// [`CancelledRecord`] variant distinguishes the three
    /// shapes: an in-flight task that was signalled by the
    /// executor; a cascade descendant of a cancelled task; or a
    /// task drained from `state.ready` on cancel-fire. For an
    /// in-flight task, [`Self::on_task_started`] has already
    /// fired and [`Self::on_task_finished`] does NOT fire; for
    /// the other two shapes, no other lifecycle callback fires
    /// for the same task.
    fn on_task_cancelled(&self, task: &TaskId, record: &CancelledRecord);
}

/// Failure modes of [`run_task`].
///
/// The variants enumerate only the *executor-level* failures: cases
/// where the lifecycle could not be carried through to a meaningful
/// [`CompletedRecord`]. Task-level failures (non-zero exit, signalled
/// exit) surface as [`Ok(CompletedRecord { state: Failed, .. })`]
/// per `EXEC-009`, not as an [`Err`] here.
#[derive(Debug, Snafu)]
#[snafu(visibility(pub(crate)))]
pub enum RunTaskError {
    /// Cache-key derivation failed before lookup could be attempted.
    #[snafu(display("failed to derive cache key: {source}"))]
    BuildKeyFailed {
        /// Originating [`BuildKeyError`].
        source: BuildKeyError,
    },

    /// [`haz_cache::CacheWriter::restore`] failed during the cache-hit
    /// path (`CACHE-019`). Cache misses are not errors per
    /// `CACHE-016`, so this variant covers only the successful-
    /// lookup-then-restore-failed shape.
    #[snafu(display("failed to restore cache entry: {source}"))]
    RestoreFailed {
        /// Originating [`RestoreError`].
        source: RestoreError,
    },

    /// The [`ProcessSpawner`] refused to start the child (executable
    /// not found, permission denied, fork failure, etc.). Distinct
    /// from `EXEC-009`'s failed state: that rule covers commands
    /// that *did* run; this variant is for commands that never did.
    #[snafu(display("failed to spawn process: {source}"))]
    SpawnFailed {
        /// Originating [`ProcessError`].
        source: ProcessError,
    },

    /// [`crate::process::Process::wait`] surfaced an I/O failure
    /// while reaping the child. Rare on healthy hosts; usually indicates
    /// that the child was reaped out from under us by a parent
    /// signal handler or that the OS lost the descriptor.
    #[snafu(display("failed to wait for spawned process: {source}"))]
    WaitFailed {
        /// Originating [`ProcessError`].
        source: ProcessError,
    },

    /// Reading the child's stdout or stderr pipe to EOF failed
    /// during stream capture. The reader task that returned this
    /// error is the one whose stream did NOT reach EOF; the other
    /// stream's capture may or may not have completed.
    #[snafu(display("failed to read captured stream: {source}"))]
    CapturedStreamReadFailed {
        /// Which of the two streams the failure came from, for
        /// diagnostic precision (the underlying [`io::Error`] often
        /// looks the same on both pipes).
        stream: CapturedStream,
        /// Underlying I/O error.
        source: io::Error,
    },

    /// Walking the filesystem to resolve a task's `outputs`
    /// patterns failed (output-side parallel of
    /// [`BuildKeyError::InputPatternResolutionFailed`]). Distinct
    /// from [`Self::OutputNotARegularFile`] and
    /// [`Self::OutputDeclaredButNotProduced`] so the run summary
    /// can tell "walk failed" from "the task did not honour its
    /// declared outputs".
    #[snafu(display(
        "failed to resolve output patterns under: {}: {source}",
        root.display()
    ))]
    OutputPatternResolutionFailed {
        /// The absolute path being walked when the failure occurred.
        root: PathBuf,
        /// Underlying filesystem error.
        source: FsError,
    },

    /// Reading the Unix permission bits of a matched output file
    /// failed. The bytes themselves are read inside
    /// [`haz_cache::CacheWriter::store`]; this variant covers only the
    /// mode lookup the executor performs to build
    /// [`haz_cache::StoredOutput`].
    #[snafu(display("failed to read mode of output file: {}: {source}", path.display()))]
    OutputModeReadFailed {
        /// The output file whose mode could not be read.
        path: PathBuf,
        /// Underlying filesystem error.
        source: FsError,
    },

    /// A matched output path is not a regular file (a directory,
    /// symlink to a non-file, socket, FIFO, etc.). The cache only
    /// stores regular-file blobs (`CACHE-013`); non-file matches
    /// cannot be ingested.
    #[snafu(display("output path is not a regular file: {}", path.display()))]
    OutputNotARegularFile {
        /// Absolute path of the offending entry.
        path: PathBuf,
    },

    /// A literal output pattern named a path the task did not
    /// produce (the file does not exist on disk after a successful
    /// run). Distinct from the input-side
    /// [`BuildKeyError::InputPatternResolutionFailed`] because the
    /// caller's intent is the inverse: outputs are a contract the
    /// task is supposed to honour; a missing literal output is a
    /// task-level bug, not a filesystem accident.
    #[snafu(display("task declared output but did not produce it: {}", path.display()))]
    OutputDeclaredButNotProduced {
        /// Absolute path of the declared-but-missing output.
        path: PathBuf,
    },

    /// [`haz_cache::CacheWriter::store`] failed while persisting a
    /// successful run as a cache entry (`CACHE-017`).
    #[snafu(display("failed to store cache entry: {source}"))]
    StoreFailed {
        /// Originating [`StoreError`].
        source: StoreError,
    },

    /// A workspace-absolute path produced by the executor's
    /// output-resolution pass failed to parse as a
    /// [`CanonicalPath`].
    ///
    /// Indicates an internal invariant violation: the resolver
    /// builds each path string from already-validated
    /// [`haz_domain::path::segment::PathSegment`]s, so the parse
    /// is expected to succeed unconditionally. The variant exists
    /// so the failure surfaces typed rather than panicking; in
    /// practice it should never fire.
    #[snafu(display("materialised output path is not workspace-absolute: {path}: {source}"))]
    MaterialisedOutputPathInvalid {
        /// The offending path string.
        path: String,
        /// Originating parse error.
        source: ParseAbsoluteError,
    },
}

/// Which of the two captured byte streams a
/// [`RunTaskError::CapturedStreamReadFailed`] refers to.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum CapturedStream {
    /// The child's standard output pipe.
    Stdout,
    /// The child's standard error pipe.
    Stderr,
}

/// Borrowed bundle of long-lived state that every [`run_task`]
/// invocation in a given `haz` run shares.
///
/// The split between this context and the per-task arguments
/// matches a scheduler's natural shape: one `RunContext` covers
/// the entire scheduling loop, while `task`, `predecessor_streams`,
/// and `created_at_unix` change between individual [`run_task`]
/// calls.
pub struct RunContext<'a, F, S, O>
where
    F: WritableFilesystem,
    S: ProcessSpawner,
    O: RunObserver,
{
    /// Filesystem the executor reads inputs / outputs / cache state
    /// through.
    pub fs: &'a F,
    /// Cache handle (lookup, restore, store).
    pub cache: &'a CacheWriter<F>,
    /// Process spawner (a test-double mock under `cfg(test)`,
    /// [`crate::std_impl::StdProcessSpawner`] in production).
    pub spawner: &'a S,
    /// Observer that receives lifecycle events and captured streams.
    pub observer: &'a O,
    /// Validated workspace state.
    pub workspace: &'a Workspace,
    /// Validated dependency graph.
    pub graph: &'a TaskGraph,
    /// Host-environment snapshot. Names not present here are
    /// treated as absent per `CACHE-008`.
    pub host_env: &'a BTreeMap<EnvVarName, String>,
    /// Active hash algorithm for cache-key derivation, input /
    /// output content hashing, and captured-stream hashing.
    pub algo: HashAlgo,
    /// Cancellation signal for the run (`EXEC-012` triggers).
    /// The scheduler and per-task spawn-step futures observe
    /// this token: when it fires, the scheduler stops admitting
    /// new tasks and in-flight futures send SIGTERM to their
    /// children. The token is long-lived ambient state owned by
    /// the run's caller (typically `haz-cli` installing an OS
    /// signal handler); tests pass a never-cancelled token.
    pub cancel: &'a CancellationToken,
}

/// Run one task per its workspace declaration, cache-aware.
///
/// Lifecycle:
///
/// 1. [`RunObserver::on_task_started`] fires.
/// 2. [`haz_cache_key::build_cache_key`] derives the key from
///    `task`'s declared inputs, env contribution, and hard-edge
///    predecessors' captured stream hashes.
/// 3. [`haz_cache::CacheReader::lookup`] is consulted:
///    - On a hit, [`haz_cache::CacheWriter::restore`] materialises the
///      manifest's outputs at their workspace-absolute paths;
///      `stdout` and `stderr` flow through the observer; the
///      function returns [`RunSource::CacheHit`] with the
///      manifest's recorded stream hashes.
///    - On a miss, the function builds a [`crate::process::SpawnPlan`]
///      from the task's [`haz_domain::action::TaskAction`] and the
///      effective env (`CACHE-008` runtime view: allow-listed host
///      values plus task overrides, override wins), spawns through
///      `spawner`, captures both streams to memory, awaits exit,
///      emits the streams through the observer, classifies per
///      `EXEC-009`. On success the outputs are enumerated and
///      [`haz_cache::CacheWriter::store`] persists the entry. On
///      failure no store fires.
/// 4. [`RunObserver::on_task_finished`] fires with the terminal
///    [`CompletedRecord`].
///
/// The function is async because spawning, waiting, and stream
/// capture all use the tokio runtime. Filesystem and cache calls
/// are sync and run on the calling worker thread; callers that want
/// to off-load the cache-key derivation (which hashes input file
/// bytes) and the store phase (which hashes output file bytes) from
/// an async context can wrap the function in
/// [`tokio::task::spawn_blocking`] -- both phases dominate the
/// runtime cost of a single-task lifecycle.
///
/// `created_at_unix` is caller-supplied so the function stays a
/// pure function of its arguments: tests pin the value, the
/// scheduler passes the current wall-clock time.
///
/// # Errors
///
/// Returns a [`RunTaskError`] for any executor-level failure
/// (cache-key derivation, restore, spawn, wait, stream read,
/// output resolution, store). Task-level failures (non-zero exit
/// or signalled exit) are reported as
/// `Ok(CompletedRecord { state: Failed, .. })` per `EXEC-009`;
/// the cache is not consulted for store in that branch
/// (`CACHE-018`).
/// Carrier for the borrows + computed values
/// [`cache_lookup_phase`] hands back to the caller (typically the
/// scheduler in [`crate::run_graph::run_graph`]).
///
/// The struct lets `EXEC-007` step 1 (cache lookup) be performed
/// independently of `EXEC-007` step 2 (restore) and step 3 (spawn);
/// the scheduler interposes its mutex-compatibility check between
/// the lookup and the spawn branch.
///
/// Lifetime `'ws` ties the `project` and `task_def` borrows to the
/// workspace borrow held inside the supplying [`RunContext`].
#[derive(Debug)]
pub struct TaskLookup<'ws> {
    /// Bearing project of `task` (looked up once so neither
    /// downstream branch has to repeat the search).
    pub project: &'ws Project,
    /// Bearing task definition (action, mutex, env, etc.).
    pub task_def: &'ws Task,
    /// Computed cache key.
    pub key: haz_cache::CacheKey,
    /// Manifest of the matching cache entry, if any. `Some(_)`
    /// drives the restore branch; `None` drives the spawn branch.
    pub manifest: Option<haz_cache::Manifest>,
}

/// `EXEC-007` step 1 of the single-task lifecycle: resolve the
/// bearing project + task, derive the cache key, and consult
/// the cache. No mutex hold is taken; no observer event fires.
///
/// Returns a [`TaskLookup`] the caller drives forward into either
/// [`restore_from_hit`] (manifest present) or [`run_fresh`]
/// (manifest absent). The scheduler interposes its `EXEC-006`
/// condition-3 mutex check between this function and
/// [`run_fresh`].
///
/// # Errors
///
/// - [`RunTaskError::BuildKeyFailed`] when the task is missing from
///   the workspace, or when cache-key derivation fails for any
///   reason (input pattern resolution, file read, etc.).
pub fn cache_lookup_phase<'ws, F, S, O>(
    ctx: &RunContext<'ws, F, S, O>,
    task: &TaskId,
    predecessor_streams: &BTreeMap<TaskId, PredecessorStreamHashes>,
) -> Result<TaskLookup<'ws>, RunTaskError>
where
    F: WritableFilesystem,
    S: ProcessSpawner,
    O: RunObserver,
{
    let project =
        ctx.workspace
            .projects
            .get(&task.project)
            .ok_or_else(|| RunTaskError::BuildKeyFailed {
                source: BuildKeyError::TaskNotInWorkspace { task: task.clone() },
            })?;
    let task_def = project
        .tasks
        .get(&task.task)
        .ok_or_else(|| RunTaskError::BuildKeyFailed {
            source: BuildKeyError::TaskNotInWorkspace { task: task.clone() },
        })?;

    let key = build_cache_key(
        ctx.fs,
        ctx.workspace,
        ctx.graph,
        task,
        ctx.host_env,
        predecessor_streams,
        ctx.algo,
    )
    .context(BuildKeyFailedSnafu)?;

    let manifest = ctx.cache.reader().lookup(&key);
    Ok(TaskLookup {
        project,
        task_def,
        key,
        manifest,
    })
}

/// Single-task lifecycle entry point: composes
/// [`cache_lookup_phase`], [`restore_from_hit`], and
/// [`run_fresh`] without any mutex orchestration.
///
/// Treats every task as if its mutex were always compatible.
/// Callers that need `EXEC-006` condition 3 / `EXEC-007` mutex
/// semantics (i.e., the scheduler) MUST drive the three phases
/// directly.
///
/// Fires [`RunObserver::on_task_started`] before the lookup step
/// and [`RunObserver::on_task_finished`] after the record is in
/// hand.
///
/// # Errors
///
/// Surfaces the first error encountered across the composed
/// phases; see the docs on each underlying function for the
/// variants reachable from each phase.
pub async fn run_task<F, S, O>(
    ctx: &RunContext<'_, F, S, O>,
    task: &TaskId,
    predecessor_streams: &BTreeMap<TaskId, PredecessorStreamHashes>,
    created_at_unix: u64,
) -> Result<CompletedRecord, RunTaskError>
where
    F: WritableFilesystem,
    S: ProcessSpawner,
    O: RunObserver,
{
    ctx.observer.on_task_started(task);

    let lookup = cache_lookup_phase(ctx, task, predecessor_streams)?;

    let record = if let Some(manifest) = lookup.manifest.as_ref() {
        restore_from_hit(ctx, task, manifest)?
    } else {
        run_fresh(
            ctx,
            task,
            lookup.project,
            lookup.task_def,
            &lookup.key,
            created_at_unix,
        )
        .await?
    };

    ctx.observer.on_task_finished(task, &record);
    Ok(record)
}

/// Cache-hit branch: restore the manifest's outputs, emit the
/// recorded streams to the observer, and build the matching
/// [`CompletedRecord`] (`EXEC-007` step 2, `EXEC-017`,
/// `CACHE-019`).
///
/// Per `MUTEX-007`, this function MUST NOT touch any mutex hold
/// set: a cache hit reproduces the recorded effect via file
/// writes and stream emission and never touches the resource the
/// mutex protects.
///
/// Does NOT fire [`RunObserver::on_task_started`] or
/// [`RunObserver::on_task_finished`]; the caller drives the
/// observer lifecycle so the start/finish pair stays balanced
/// across all three phases.
///
/// # Errors
///
/// - [`RunTaskError::RestoreFailed`] when the cache restore step
///   fails (a `manifest` field references a blob that cannot be
///   read, a destination cannot be written, etc.).
pub fn restore_from_hit<F, S, O>(
    ctx: &RunContext<'_, F, S, O>,
    task: &TaskId,
    manifest: &haz_cache::Manifest,
) -> Result<CompletedRecord, RunTaskError>
where
    F: WritableFilesystem,
    S: ProcessSpawner,
    O: RunObserver,
{
    let restored = ctx.cache.restore(manifest).context(RestoreFailedSnafu)?;
    ctx.observer.on_stdout(task, &restored.stdout);
    ctx.observer.on_stderr(task, &restored.stderr);
    let materialised_outputs = manifest
        .outputs
        .iter()
        .map(|blob| blob.workspace_absolute_path.clone())
        .collect();
    Ok(CompletedRecord {
        task: task.clone(),
        source: RunSource::CacheHit,
        state: RunState::Succeeded,
        exit_status: None,
        stdout_hash: manifest.stdout_hash,
        stderr_hash: manifest.stderr_hash,
        materialised_outputs,
    })
}

/// Cache-miss branch (`EXEC-007` step 3, compatible mutex path):
/// spawn the task's command, capture both streams, classify per
/// `EXEC-009`, and on success persist the run to the cache.
///
/// Reader futures run concurrently with `process.wait()` so a child
/// that fills its stdout/stderr pipe buffer never blocks waiting for
/// the parent to drain (the classic pipe-deadlock shape). Stream
/// hashes are computed under [`RunContext::algo`] from the captured
/// byte buffers (`CACHE-011` corollary for fresh runs).
///
/// Per `MUTEX-006`, the caller MUST acquire the task's mutex
/// before invoking this function and MUST release it after the
/// returned future resolves (regardless of success or failure).
/// This function does not consult any hold set itself.
///
/// Does NOT fire [`RunObserver::on_task_started`] or
/// [`RunObserver::on_task_finished`]; the caller drives the
/// observer lifecycle.
///
/// # Errors
///
/// - [`RunTaskError::SpawnFailed`] when the process spawner
///   rejects the [`SpawnPlan`].
/// - [`RunTaskError::WaitFailed`] when waiting on the spawned
///   process surfaces an OS error.
/// - [`RunTaskError::CapturedStreamReadFailed`] when reading
///   either captured stream errors.
/// - The output-resolution and store error families when the run
///   succeeded and the cache-store step fails.
pub async fn run_fresh<F, S, O>(
    ctx: &RunContext<'_, F, S, O>,
    task: &TaskId,
    project: &Project,
    task_def: &Task,
    key: &haz_cache::CacheKey,
    created_at_unix: u64,
) -> Result<CompletedRecord, RunTaskError>
where
    F: WritableFilesystem,
    S: ProcessSpawner,
    O: RunObserver,
{
    let plan = build_spawn_plan(
        ctx.workspace.root.as_path(),
        project,
        task_def,
        ctx.host_env,
    );

    let Spawned {
        mut process,
        mut stdout,
        mut stderr,
    } = ctx.spawner.spawn(&plan).await.context(SpawnFailedSnafu)?;

    let mut stdout_bytes = Vec::new();
    let mut stderr_bytes = Vec::new();
    let grace = ctx.workspace.settings.execution.cancel_grace.as_duration();
    let ((wait_result, was_cancelled), stdout_result, stderr_result) = tokio::join!(
        await_exit_with_cancel(&mut process, ctx.cancel, grace),
        AsyncReadExt::read_to_end(&mut stdout, &mut stdout_bytes),
        AsyncReadExt::read_to_end(&mut stderr, &mut stderr_bytes),
    );

    let exit_status = wait_result.context(WaitFailedSnafu)?;
    stdout_result.context(CapturedStreamReadFailedSnafu {
        stream: CapturedStream::Stdout,
    })?;
    stderr_result.context(CapturedStreamReadFailedSnafu {
        stream: CapturedStream::Stderr,
    })?;

    ctx.observer.on_stdout(task, &stdout_bytes);
    ctx.observer.on_stderr(task, &stderr_bytes);

    let stdout_hash = hash_bytes(ctx.algo, &stdout_bytes);
    let stderr_hash = hash_bytes(ctx.algo, &stderr_bytes);

    let (state, materialised_outputs) = if was_cancelled {
        // `EXEC-015`: cancelled runs never hit the store branch
        // even on a zero exit code; cancellation is neither
        // success nor a half-success.
        (RunState::Cancelled, Vec::new())
    } else if exit_status.success() {
        let outputs_owned =
            resolve_output_files(ctx.fs, ctx.workspace, project, &task_def.outputs)?;
        let materialised = canonical_paths_from_owned(&outputs_owned)?;
        store_successful_run(
            ctx,
            key,
            &outputs_owned,
            &stdout_bytes,
            &stderr_bytes,
            created_at_unix,
        )?;
        (RunState::Succeeded, materialised)
    } else {
        (RunState::Failed, Vec::new())
    };

    Ok(CompletedRecord {
        task: task.clone(),
        source: RunSource::FreshRun,
        state,
        exit_status: Some(exit_status),
        stdout_hash,
        stderr_hash,
        materialised_outputs,
    })
}

/// Wait for the spawned child to exit, observing the run's
/// cancellation token.
///
/// On the normal path the function delegates to [`Process::wait`]
/// and returns `(Ok(status), false)`. When the cancellation token
/// fires while the child is still running, the function:
///
/// 1. Sends [`Signal::Terminate`] (best-effort; the error is
///    swallowed so Windows hosts that surface
///    [`std::io::ErrorKind::Unsupported`] still escalate).
/// 2. Races a fresh [`Process::wait`] against
///    [`tokio::time::sleep`] of `grace`.
/// 3. If the sleep wins, sends [`Signal::Kill`] (best-effort) and
///    awaits the child.
///
/// In both cancellation paths the second element of the returned
/// tuple is `true`; the caller classifies the run as
/// [`RunState::Cancelled`]. A `grace` of [`Duration::ZERO`]
/// collapses the SIGTERM-to-SIGKILL window: SIGKILL fires
/// immediately after the SIGTERM (`EXEC-014`).
async fn await_exit_with_cancel<P>(
    process: &mut P,
    cancel: &CancellationToken,
    grace: std::time::Duration,
) -> (Result<ExitStatus, ProcessError>, bool)
where
    P: Process,
{
    tokio::select! {
        biased;
        () = cancel.cancelled() => {
            let _ = process.send_signal(Signal::Terminate);
            tokio::select! {
                result = process.wait() => (result, true),
                () = tokio::time::sleep(grace) => {
                    let _ = process.send_signal(Signal::Kill);
                    (process.wait().await, true)
                }
            }
        }
        result = process.wait() => (result, false),
    }
}

/// Persist the successful run as a cache entry under `key`
/// (`CACHE-017`, `CACHE-018`).
///
/// Called only when the fresh run's exit status was zero; the
/// caller has already classified the run as succeeded and has
/// already resolved the task's declared outputs (so the same
/// resolved Vec can drive both the cache store and the
/// scheduler's materialised-outputs view).
fn store_successful_run<F, S, O>(
    ctx: &RunContext<'_, F, S, O>,
    key: &haz_cache::CacheKey,
    outputs_owned: &[OwnedOutputFile],
    stdout: &[u8],
    stderr: &[u8],
    created_at_unix: u64,
) -> Result<(), RunTaskError>
where
    F: WritableFilesystem,
    S: ProcessSpawner,
    O: RunObserver,
{
    let stored_outputs: Vec<StoredOutput<'_>> = outputs_owned
        .iter()
        .map(|f| StoredOutput {
            workspace_absolute_path: &f.workspace_absolute_path,
            on_disk_path: &f.on_disk_path,
            mode: f.mode,
        })
        .collect();
    let inputs = StoreInputs {
        outputs: &stored_outputs,
        stdout,
        stderr,
        created_at_unix,
    };
    ctx.cache.store(key, &inputs).context(StoreFailedSnafu)
}

/// Parse the workspace-absolute path string of each
/// [`OwnedOutputFile`] into a [`CanonicalPath`] for the
/// scheduler's runtime DAG validation pass.
///
/// The strings come from
/// [`haz_cache_key::pattern_walk::workspace_absolute_string_from_segments`]
/// where every segment was validated upstream by
/// [`PathSegment`]'s constructors, so the parse should never
/// fail in practice; the [`Result`] exists so a hypothetical
/// invariant violation surfaces as a typed
/// [`RunTaskError::MaterialisedOutputPathInvalid`] rather than
/// a panic.
fn canonical_paths_from_owned(
    files: &[OwnedOutputFile],
) -> Result<Vec<CanonicalPath>, RunTaskError> {
    files
        .iter()
        .map(|f| {
            CanonicalPath::parse_workspace_absolute(&f.workspace_absolute_path).context(
                MaterialisedOutputPathInvalidSnafu {
                    path: f.workspace_absolute_path.clone(),
                },
            )
        })
        .collect()
}

/// Hash `bytes` under the active [`HashAlgo`]. Used to compute the
/// captured-stream hashes recorded on a fresh-run
/// [`CompletedRecord`].
fn hash_bytes(algo: HashAlgo, bytes: &[u8]) -> [u8; 32] {
    let mut hasher = Hasher::new(algo);
    hasher.update(bytes);
    hasher.finalize()
}

/// Assemble the [`SpawnPlan`] that [`run_task`]'s miss branch hands
/// to the [`ProcessSpawner`].
///
/// Three pieces are derived from `task_def`:
///
/// - `program` and `args` from `task_def.action`. A
///   [`TaskAction::Command`] splits as `program = argv.head`,
///   `args = argv.tail`. A [`TaskAction::Shell`] becomes
///   `program = <shell-binary-name>`, `args = ["-c", script]`.
/// - `cwd` is the project's host path: `workspace_host` walked
///   through the project root's segments. An implicit-mode
///   project ([`ProjectRoot::WorkspaceRoot`], per `DISC-003`) lands
///   at `workspace_host` itself with no further pushing.
/// - `env` is the effective env per `CACHE-008` runtime view:
///   names in `task_def.env.from_host` propagate the value the
///   host snapshot recorded for them, names absent from the host
///   simply do not appear in the env vector; `task_def.env.overrides`
///   entries are layered on top, override winning on collision.
///   The std backend's `env_clear()` ensures the child sees this
///   set and nothing else.
fn build_spawn_plan(
    workspace_host: &Path,
    project: &Project,
    task_def: &Task,
    host_env: &BTreeMap<EnvVarName, String>,
) -> SpawnPlan {
    let (program, args) = program_and_args(&task_def.action);
    SpawnPlan {
        program,
        args,
        env: build_env_vec(&task_def.env, host_env),
        cwd: project_cwd(workspace_host, &project.root),
    }
}

/// Walk `workspace_host` through the project root's segments. For
/// [`ProjectRoot::WorkspaceRoot`] (implicit-mode project per
/// `DISC-003`) returns the workspace host path verbatim.
fn project_cwd(workspace_host: &Path, project_root: &ProjectRoot) -> PathBuf {
    let mut p = workspace_host.to_path_buf();
    if let ProjectRoot::Nested(canonical) = project_root {
        for seg in canonical.segments() {
            p.push(seg.as_str());
        }
    }
    p
}

/// Convert a [`TaskAction`] into the
/// (`program`, `args`) pair the spawner consumes.
fn program_and_args(action: &TaskAction) -> (OsString, Vec<OsString>) {
    match action {
        TaskAction::Command(nonempty_argv) => {
            let program = OsString::from(nonempty_argv.head.as_str());
            let args = nonempty_argv.tail.iter().map(OsString::from).collect();
            (program, args)
        }
        TaskAction::Shell { script, shell } => {
            let program = OsString::from(shell_binary_name(shell));
            let args = vec![OsString::from("-c"), OsString::from(script)];
            (program, args)
        }
    }
}

/// Resolve [`ShellType`] to the binary name passed as `argv[0]` to
/// `exec`.
///
/// First-class variants emit their canonical identifiers (`sh`,
/// `bash`); [`ShellType::Other`] emits the validated name verbatim.
/// The shell is resolved through `PATH` at exec time, matching how
/// `system(3)` and `/bin/sh -c` behave on POSIX hosts.
fn shell_binary_name(shell: &ShellType) -> &str {
    match shell {
        ShellType::Sh => "sh",
        ShellType::Bash => "bash",
        ShellType::Other(name) => AsRef::<str>::as_ref(name.as_ref()),
    }
}

/// Assemble the effective env vector per `CACHE-008` runtime view.
///
/// Steps:
///
/// 1. For each name in `env.from_host`, look it up in `host_env`.
///    Present names contribute their host value; absent names
///    contribute nothing (the child simply will not see that name).
/// 2. For each entry in `env.overrides`, set the name to the
///    override value, replacing any same-name entry from step 1
///    (`CACHE-008` "overrides wins on collision").
///
/// Names not in either map do not appear; the std spawner's
/// `env_clear()` ensures the child's env is exactly the returned
/// set.
///
/// Order in the returned vector is lexicographic by name (the
/// [`BTreeMap`] keeps it so). The std spawner preserves order on
/// `Command::env`; either order works on POSIX hosts because the
/// child sees a set, not a sequence, but lexicographic ordering
/// makes test assertions deterministic.
fn build_env_vec(
    env: &EnvSettings,
    host_env: &BTreeMap<EnvVarName, String>,
) -> Vec<(OsString, OsString)> {
    let mut effective: BTreeMap<OsString, OsString> = BTreeMap::new();
    for name in &env.from_host {
        if let Some(value) = host_env.get(name) {
            effective.insert(
                OsString::from(AsRef::<str>::as_ref(name.as_ref())),
                OsString::from(value),
            );
        }
    }
    for (name, value) in &env.overrides {
        effective.insert(
            OsString::from(AsRef::<str>::as_ref(name.as_ref())),
            OsString::from(value),
        );
    }
    effective.into_iter().collect()
}

/// One file resolved from a task's `outputs` patterns, paired with
/// the host filesystem path it lives at and the Unix permission bits
/// recorded for it.
///
/// `workspace_absolute_path` is the workspace-anchored string (rooted
/// at `/`) the cache library records in the manifest;
/// `on_disk_path` is the real host path the cache reads to ingest the
/// blob; `mode` is the value [`haz_cache::StoredOutput::mode`]
/// receives on store. Borrowed views of these fields drop straight
/// into [`haz_cache::StoredOutput`].
#[derive(Debug, Clone, PartialEq, Eq)]
struct OwnedOutputFile {
    workspace_absolute_path: String,
    on_disk_path: PathBuf,
    mode: u32,
}

/// Resolve every [`OutputSpec`] declared on a task against `fs` and
/// record each match's host path and Unix mode.
///
/// Output-side mirror of the input resolver in
/// [`haz_cache_key::resolve_input_files`]:
/// a [`PathPattern::Literal`] contributes exactly one entry on a
/// regular-file match and a typed failure otherwise; a
/// [`PathPattern::Glob`] contributes zero or more entries through
/// the [`GlobWalk`] machinery. Errors surface the first failure
/// encountered while walking, reading metadata, or reading the
/// permission bits; later outputs in the same task are not
/// attempted.
///
/// Diagnostic divergence from the input resolver: a literal output
/// that resolves to [`FsError::NotFound`] surfaces as
/// [`RunTaskError::OutputDeclaredButNotProduced`] rather than
/// [`RunTaskError::OutputPatternResolutionFailed`]. Outputs are a
/// contract the task is supposed to honour, so a missing literal
/// output is a task-level failure, not a filesystem accident.
///
/// Symlink semantics follow the input resolver: a symlink-to-file
/// contributes the link's own workspace-absolute path, while the
/// mode bits come from the canonical target via
/// [`Filesystem::permissions`] (which follows symlinks).
fn resolve_output_files<F: Filesystem>(
    fs: &F,
    workspace: &Workspace,
    project: &Project,
    outputs: &[OutputSpec],
) -> Result<Vec<OwnedOutputFile>, RunTaskError> {
    let workspace_host = workspace.root.as_path();
    let action = OutputAction;
    let mut out = Vec::new();

    for spec in outputs {
        match spec.pattern() {
            PathPattern::Literal(haz_path) => {
                resolve_literal_output(
                    fs,
                    workspace_host,
                    &project.root,
                    haz_path,
                    &action,
                    &mut out,
                )?;
            }
            PathPattern::Glob(glob_pattern) => {
                let glob = glob_pattern.compile();
                let matcher = glob.compile_matcher();
                let (walk_host, workspace_prefix, candidate_prefix) =
                    glob_walk_origin(workspace_host, &project.root, glob_pattern.anchor());
                let walker = GlobWalk {
                    fs,
                    matcher: &matcher,
                    candidate_prefix,
                    workspace_prefix,
                    action: &action,
                };
                let mut walk_rel: Vec<String> = Vec::new();
                walker.walk(&walk_host, &mut walk_rel, &mut out)?;
            }
        }
    }

    Ok(out)
}

/// Resolve a literal [`OutputSpec`] under the supplied project root
/// and dispatch to the per-match action on a regular-file match.
///
/// Splits the failure space three ways:
///
/// - [`FsError::NotFound`] becomes
///   [`RunTaskError::OutputDeclaredButNotProduced`].
/// - Any other [`FsError`] from the metadata read becomes
///   [`RunTaskError::OutputPatternResolutionFailed`].
/// - A non-regular-file kind becomes
///   [`RunTaskError::OutputNotARegularFile`].
fn resolve_literal_output<F: Filesystem>(
    fs: &F,
    workspace_host: &Path,
    project_root: &ProjectRoot,
    haz_path: &haz_domain::path::HazPath,
    action: &OutputAction,
    out: &mut Vec<OwnedOutputFile>,
) -> Result<(), RunTaskError> {
    let ws_segments = literal_workspace_segments(haz_path, project_root);
    let host = host_path_from_segments(workspace_host, &ws_segments);

    let meta = match fs.metadata(&host) {
        Ok(m) => m,
        Err(FsError::NotFound { path }) => {
            return Err(RunTaskError::OutputDeclaredButNotProduced { path });
        }
        Err(source) => {
            return Err(RunTaskError::OutputPatternResolutionFailed { root: host, source });
        }
    };
    if meta.kind != EntryKind::File {
        return Err(RunTaskError::OutputNotARegularFile { path: host });
    }

    let workspace_absolute_path = workspace_absolute_string_from_segments(&ws_segments);
    action.on_match(fs, &host, workspace_absolute_path, out)
}

/// Per-match action for output resolution: record the matched file's
/// host path and Unix mode. Walk-level [`FsError`]s surface as
/// [`RunTaskError::OutputPatternResolutionFailed`]; mode-read failures
/// surface as [`RunTaskError::OutputModeReadFailed`].
///
/// The action carries no state: the host snapshot, project root,
/// and workspace path projection are all threaded through the
/// [`GlobWalk`] machinery.
struct OutputAction;

impl<F: Filesystem> GlobMatchAction<F> for OutputAction {
    type Output = OwnedOutputFile;
    type Error = RunTaskError;

    fn map_walk_error(&self, root: PathBuf, source: FsError) -> RunTaskError {
        RunTaskError::OutputPatternResolutionFailed { root, source }
    }

    fn on_match(
        &self,
        fs: &F,
        host_path: &Path,
        workspace_absolute_path: String,
        out: &mut Vec<OwnedOutputFile>,
    ) -> Result<(), RunTaskError> {
        let mode = fs
            .permissions(host_path)
            .context(OutputModeReadFailedSnafu {
                path: host_path.to_path_buf(),
            })?;
        out.push(OwnedOutputFile {
            workspace_absolute_path,
            on_disk_path: host_path.to_path_buf(),
            mode,
        });
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use std::collections::{BTreeMap, BTreeSet};
    use std::ffi::OsString;
    use std::path::{Path, PathBuf};
    use std::str::FromStr;

    use nonempty::NonEmpty;

    use haz_domain::action::{ShellType, TaskAction};
    use haz_domain::env::{EnvSettings, EnvVarName};
    use haz_domain::name::{ProjectName, TaskName};
    use haz_domain::path::{CanonicalPath, HazPath, InputSpec, ProjectRoot};
    use haz_domain::project::Project;
    use haz_domain::task::Task;

    use super::{build_env_vec, build_spawn_plan, project_cwd, shell_binary_name};

    fn env_name(s: &str) -> EnvVarName {
        EnvVarName::try_new(s).unwrap()
    }

    fn project_name(s: &str) -> ProjectName {
        ProjectName::try_new(s).unwrap()
    }

    fn task_name(s: &str) -> TaskName {
        TaskName::try_new(s).unwrap()
    }

    fn nested_project(name: &str, root: &str) -> Project {
        Project {
            name: project_name(name),
            root: ProjectRoot::Nested(
                CanonicalPath::from_absolute(&HazPath::parse(root).unwrap()).unwrap(),
            ),
            tags: BTreeSet::new(),
            tasks: BTreeMap::new(),
        }
    }

    fn implicit_project(name: &str) -> Project {
        Project {
            name: project_name(name),
            root: ProjectRoot::WorkspaceRoot,
            tags: BTreeSet::new(),
            tasks: BTreeMap::new(),
        }
    }

    fn task_with(action: TaskAction, env: EnvSettings) -> Task {
        Task {
            name: task_name("run"),
            action,
            inputs: Vec::<InputSpec>::new(),
            outputs: Vec::new(),
            deps: Vec::new(),
            weak_deps: Vec::new(),
            mutex: None,
            env,
        }
    }

    fn command(argv: &[&str]) -> TaskAction {
        TaskAction::Command(
            NonEmpty::from_vec(argv.iter().map(|s| (*s).to_owned()).collect())
                .expect("non-empty argv"),
        )
    }

    fn shell(script: &str, shell_type: ShellType) -> TaskAction {
        TaskAction::Shell {
            script: script.to_owned(),
            shell: shell_type,
        }
    }

    // ---- shell_binary_name ----

    #[test]
    fn shell_binary_name_for_sh() {
        assert_eq!(shell_binary_name(&ShellType::Sh), "sh");
    }

    #[test]
    fn shell_binary_name_for_bash() {
        assert_eq!(shell_binary_name(&ShellType::Bash), "bash");
    }

    #[test]
    fn shell_binary_name_for_other_uses_validated_name() {
        let name = haz_domain::action::NonEmptyAsciiName::from_str("zsh").unwrap();
        assert_eq!(shell_binary_name(&ShellType::Other(name)), "zsh");
    }

    // ---- project_cwd ----

    #[test]
    fn project_cwd_for_nested_appends_segments() {
        let p = nested_project("lib_core", "/lib_core");
        let cwd = project_cwd(Path::new("/abs/ws"), &p.root);
        assert_eq!(cwd, PathBuf::from("/abs/ws/lib_core"));
    }

    #[test]
    fn project_cwd_for_deep_nested_walks_every_segment() {
        let p = nested_project("frontend", "/web/frontend");
        let cwd = project_cwd(Path::new("/abs/ws"), &p.root);
        assert_eq!(cwd, PathBuf::from("/abs/ws/web/frontend"));
    }

    #[test]
    fn project_cwd_for_implicit_mode_is_workspace_host() {
        let p = implicit_project("root");
        let cwd = project_cwd(Path::new("/abs/ws"), &p.root);
        assert_eq!(cwd, PathBuf::from("/abs/ws"));
    }

    // ---- build_env_vec ----

    fn env_settings(from_host: &[&str], overrides: &[(&str, &str)]) -> EnvSettings {
        EnvSettings {
            from_host: from_host.iter().map(|s| env_name(s)).collect(),
            overrides: overrides
                .iter()
                .map(|(k, v)| (env_name(k), (*v).to_owned()))
                .collect(),
        }
    }

    fn host_snapshot(entries: &[(&str, &str)]) -> BTreeMap<EnvVarName, String> {
        entries
            .iter()
            .map(|(k, v)| (env_name(k), (*v).to_owned()))
            .collect()
    }

    fn osstr_pairs(slice: &[(&str, &str)]) -> Vec<(OsString, OsString)> {
        slice
            .iter()
            .map(|(k, v)| (OsString::from(*k), OsString::from(*v)))
            .collect()
    }

    #[test]
    fn build_env_vec_from_host_present_name_propagates() {
        let settings = env_settings(&["PATH"], &[]);
        let host = host_snapshot(&[("PATH", "/usr/bin")]);
        let got = build_env_vec(&settings, &host);
        assert_eq!(got, osstr_pairs(&[("PATH", "/usr/bin")]));
    }

    #[test]
    fn build_env_vec_from_host_absent_name_is_excluded() {
        // CACHE-008: a from_host name absent from the host
        // snapshot contributes the `0x00` absent marker to the
        // cache key, but does NOT appear in the child's runtime
        // env. The child simply does not see the variable.
        let settings = env_settings(&["NEVER_SET"], &[]);
        let host = host_snapshot(&[("OTHER", "v")]);
        let got = build_env_vec(&settings, &host);
        assert!(
            got.is_empty(),
            "absent from_host names must not enter the child env, got {got:?}",
        );
    }

    #[test]
    fn build_env_vec_override_wins_on_collision() {
        // CACHE-008 "overrides wins on collision": at runtime the
        // child sees the override value, not the host value.
        let settings = env_settings(&["X"], &[("X", "override-val")]);
        let host = host_snapshot(&[("X", "host-val")]);
        let got = build_env_vec(&settings, &host);
        assert_eq!(got, osstr_pairs(&[("X", "override-val")]));
    }

    #[test]
    fn build_env_vec_override_only_entries_propagate() {
        let settings = env_settings(&[], &[("HAZ_MODE", "ci")]);
        let host = host_snapshot(&[]);
        let got = build_env_vec(&settings, &host);
        assert_eq!(got, osstr_pairs(&[("HAZ_MODE", "ci")]));
    }

    #[test]
    fn build_env_vec_unrelated_host_names_do_not_appear() {
        // Cache soundness corollary: a host variable NOT named in
        // either from_host or overrides MUST NOT enter the child's
        // env, even though it sits in the host snapshot.
        let settings = env_settings(&["PATH"], &[]);
        let host = host_snapshot(&[("PATH", "/usr/bin"), ("HOME", "/home/me")]);
        let got = build_env_vec(&settings, &host);
        assert_eq!(got, osstr_pairs(&[("PATH", "/usr/bin")]));
    }

    #[test]
    fn build_env_vec_result_is_lexicographic() {
        let settings = env_settings(&["ZULU", "ALPHA", "BRAVO"], &[]);
        let host = host_snapshot(&[("ALPHA", "a"), ("BRAVO", "b"), ("ZULU", "z")]);
        let got = build_env_vec(&settings, &host);
        let names: Vec<&OsString> = got.iter().map(|(k, _)| k).collect();
        assert_eq!(
            names,
            vec![
                &OsString::from("ALPHA"),
                &OsString::from("BRAVO"),
                &OsString::from("ZULU"),
            ]
        );
    }

    #[test]
    fn build_env_vec_empty_value_is_kept() {
        // An empty-string value is a legitimate POSIX env entry,
        // and distinct from absence (which CACHE-008 separates via
        // the 0x00 marker). Make sure we propagate the empty string
        // rather than collapsing it to "absent".
        let settings = env_settings(&["X"], &[]);
        let host = host_snapshot(&[("X", "")]);
        let got = build_env_vec(&settings, &host);
        assert_eq!(got, osstr_pairs(&[("X", "")]));
    }

    // ---- build_spawn_plan ----

    #[test]
    fn build_spawn_plan_command_maps_argv_head_and_tail() {
        let p = nested_project("proj", "/proj");
        let t = task_with(
            command(&["cargo", "build", "--release"]),
            EnvSettings::default(),
        );
        let plan = build_spawn_plan(Path::new("/ws"), &p, &t, &BTreeMap::new());

        assert_eq!(plan.program, OsString::from("cargo"));
        assert_eq!(
            plan.args,
            vec![OsString::from("build"), OsString::from("--release")],
        );
    }

    #[test]
    fn build_spawn_plan_command_with_single_arg_has_empty_args() {
        let p = nested_project("proj", "/proj");
        let t = task_with(command(&["true"]), EnvSettings::default());
        let plan = build_spawn_plan(Path::new("/ws"), &p, &t, &BTreeMap::new());

        assert_eq!(plan.program, OsString::from("true"));
        assert!(plan.args.is_empty());
    }

    #[test]
    fn build_spawn_plan_shell_uses_dash_c_and_script() {
        let p = nested_project("proj", "/proj");
        let t = task_with(shell("echo hi", ShellType::Sh), EnvSettings::default());
        let plan = build_spawn_plan(Path::new("/ws"), &p, &t, &BTreeMap::new());

        assert_eq!(plan.program, OsString::from("sh"));
        assert_eq!(
            plan.args,
            vec![OsString::from("-c"), OsString::from("echo hi")],
        );
    }

    #[test]
    fn build_spawn_plan_shell_other_uses_named_binary() {
        let p = nested_project("proj", "/proj");
        let other = haz_domain::action::NonEmptyAsciiName::from_str("zsh").unwrap();
        let t = task_with(
            shell("echo hi", ShellType::Other(other)),
            EnvSettings::default(),
        );
        let plan = build_spawn_plan(Path::new("/ws"), &p, &t, &BTreeMap::new());

        assert_eq!(plan.program, OsString::from("zsh"));
        assert_eq!(
            plan.args,
            vec![OsString::from("-c"), OsString::from("echo hi")],
        );
    }

    #[test]
    fn build_spawn_plan_sets_cwd_to_project_host_path_for_nested() {
        let p = nested_project("frontend", "/web/frontend");
        let t = task_with(command(&["true"]), EnvSettings::default());
        let plan = build_spawn_plan(Path::new("/abs/ws"), &p, &t, &BTreeMap::new());

        assert_eq!(plan.cwd, PathBuf::from("/abs/ws/web/frontend"));
    }

    #[test]
    fn build_spawn_plan_sets_cwd_to_workspace_host_for_implicit_project() {
        let p = implicit_project("root");
        let t = task_with(command(&["true"]), EnvSettings::default());
        let plan = build_spawn_plan(Path::new("/abs/ws"), &p, &t, &BTreeMap::new());

        assert_eq!(plan.cwd, PathBuf::from("/abs/ws"));
    }

    #[test]
    fn build_spawn_plan_env_reflects_from_host_overrides_and_excludes_unrelated() {
        // End-to-end: feed a representative env config into the
        // whole helper and confirm the SpawnPlan.env matches the
        // CACHE-008 runtime view exactly.
        let settings = env_settings(
            &["PATH", "MISSING"],
            &[("HAZ_MODE", "ci"), ("PATH", "/override/bin")],
        );
        let p = nested_project("proj", "/proj");
        let t = task_with(command(&["true"]), settings);

        let host = host_snapshot(&[("PATH", "/usr/bin"), ("UNRELATED", "should-not-appear")]);
        let plan = build_spawn_plan(Path::new("/ws"), &p, &t, &host);

        assert_eq!(
            plan.env,
            osstr_pairs(&[("HAZ_MODE", "ci"), ("PATH", "/override/bin"),]),
            "override wins over from_host; UNRELATED and MISSING (absent) are excluded",
        );
    }

    // ---- resolve_output_files ----

    mod output_resolution {
        use std::collections::{BTreeMap, BTreeSet};
        use std::path::PathBuf;

        use haz_domain::path::{
            CanonicalPath, HazPath, OutputSpec, ProjectRoot, WorkspaceRootPath,
        };
        use haz_domain::project::Project;
        use haz_domain::settings::WorkspaceSettings;
        use haz_domain::workspace::Workspace;
        use haz_vfs::WritableFilesystem;
        use haz_vfs_testing::MemFilesystem;

        use super::super::{OwnedOutputFile, RunTaskError, resolve_output_files};

        const WORKSPACE_HOST: &str = "/ws";
        const PROJECT_HOST: &str = "/ws/proj";

        fn nested_project() -> Project {
            Project {
                name: haz_domain::name::ProjectName::try_new("proj").unwrap(),
                root: ProjectRoot::Nested(
                    CanonicalPath::from_absolute(&HazPath::parse("/proj").unwrap()).unwrap(),
                ),
                tags: BTreeSet::new(),
                tasks: BTreeMap::new(),
            }
        }

        fn implicit_project() -> Project {
            Project {
                name: haz_domain::name::ProjectName::try_new("root").unwrap(),
                root: ProjectRoot::WorkspaceRoot,
                tags: BTreeSet::new(),
                tasks: BTreeMap::new(),
            }
        }

        fn workspace_with(project: &Project) -> Workspace {
            let mut projects = BTreeMap::new();
            projects.insert(project.name.clone(), project.clone());
            Workspace {
                root: WorkspaceRootPath::try_new(PathBuf::from(WORKSPACE_HOST)).unwrap(),
                projects,
                overlays: BTreeMap::new(),
                settings: WorkspaceSettings::default(),
            }
        }

        fn paths_of(files: &[OwnedOutputFile]) -> BTreeSet<String> {
            files
                .iter()
                .map(|f| f.workspace_absolute_path.clone())
                .collect()
        }

        fn host_paths_of(files: &[OwnedOutputFile]) -> BTreeSet<PathBuf> {
            files.iter().map(|f| f.on_disk_path.clone()).collect()
        }

        fn modes_by_workspace_path(files: &[OwnedOutputFile]) -> BTreeMap<String, u32> {
            files
                .iter()
                .map(|f| (f.workspace_absolute_path.clone(), f.mode))
                .collect()
        }

        #[test]
        fn literal_hit_returns_one_output_with_recorded_mode() {
            let mut fs = MemFilesystem::new();
            fs.add_dir(PROJECT_HOST).unwrap();
            fs.add_file_with_mode(format!("{PROJECT_HOST}/bundle.js"), b"data".to_vec(), 0o640)
                .unwrap();

            let project = nested_project();
            let workspace = workspace_with(&project);
            let outputs = vec![OutputSpec::parse("bundle.js").unwrap()];

            let result = resolve_output_files(&fs, &workspace, &project, &outputs).unwrap();
            assert_eq!(result.len(), 1);
            assert_eq!(result[0].workspace_absolute_path, "/proj/bundle.js");
            assert_eq!(result[0].on_disk_path, PathBuf::from("/ws/proj/bundle.js"));
            assert_eq!(result[0].mode & 0o7777, 0o640);
        }

        #[test]
        fn literal_workspace_absolute_resolves_under_workspace_root() {
            let mut fs = MemFilesystem::new();
            fs.add_dir("/ws/dist").unwrap();
            fs.add_file_with_mode("/ws/dist/main.js", b"x".to_vec(), 0o755)
                .unwrap();
            fs.add_dir(PROJECT_HOST).unwrap();

            let project = nested_project();
            let workspace = workspace_with(&project);
            let outputs = vec![OutputSpec::parse("/dist/main.js").unwrap()];

            let result = resolve_output_files(&fs, &workspace, &project, &outputs).unwrap();
            assert_eq!(result.len(), 1);
            assert_eq!(result[0].workspace_absolute_path, "/dist/main.js");
            assert_eq!(result[0].on_disk_path, PathBuf::from("/ws/dist/main.js"));
            assert_eq!(result[0].mode & 0o7777, 0o755);
        }

        #[test]
        fn literal_missing_surfaces_output_declared_but_not_produced() {
            // Outputs are a contract the task is supposed to honour;
            // a missing literal must NOT collapse into the input-side
            // PatternResolutionFailed shape.
            let mut fs = MemFilesystem::new();
            fs.add_dir(PROJECT_HOST).unwrap();

            let project = nested_project();
            let workspace = workspace_with(&project);
            let outputs = vec![OutputSpec::parse("absent.txt").unwrap()];

            match resolve_output_files(&fs, &workspace, &project, &outputs) {
                Err(RunTaskError::OutputDeclaredButNotProduced { path }) => {
                    assert_eq!(path, PathBuf::from("/ws/proj/absent.txt"));
                }
                other => panic!("expected OutputDeclaredButNotProduced, got {other:?}"),
            }
        }

        #[test]
        fn literal_pointing_at_directory_surfaces_output_not_a_regular_file() {
            let mut fs = MemFilesystem::new();
            fs.add_dir(format!("{PROJECT_HOST}/subdir")).unwrap();

            let project = nested_project();
            let workspace = workspace_with(&project);
            let outputs = vec![OutputSpec::parse("subdir").unwrap()];

            match resolve_output_files(&fs, &workspace, &project, &outputs) {
                Err(RunTaskError::OutputNotARegularFile { path }) => {
                    assert_eq!(path, PathBuf::from("/ws/proj/subdir"));
                }
                other => panic!("expected OutputNotARegularFile, got {other:?}"),
            }
        }

        #[test]
        fn glob_multi_match_collects_every_matching_file_with_its_mode() {
            let mut fs = MemFilesystem::new();
            fs.add_dir(PROJECT_HOST).unwrap();
            fs.add_file_with_mode(format!("{PROJECT_HOST}/a.js"), b"a".to_vec(), 0o644)
                .unwrap();
            fs.add_file_with_mode(format!("{PROJECT_HOST}/b.js"), b"b".to_vec(), 0o600)
                .unwrap();
            // A non-matching neighbour MUST NOT contribute.
            fs.add_file_with_mode(
                format!("{PROJECT_HOST}/keep.txt"),
                b"ignored".to_vec(),
                0o644,
            )
            .unwrap();

            let project = nested_project();
            let workspace = workspace_with(&project);
            let outputs = vec![OutputSpec::parse("*.js").unwrap()];

            let result = resolve_output_files(&fs, &workspace, &project, &outputs).unwrap();
            assert_eq!(result.len(), 2);
            assert_eq!(
                paths_of(&result),
                BTreeSet::from(["/proj/a.js".to_owned(), "/proj/b.js".to_owned()]),
            );
            let modes = modes_by_workspace_path(&result);
            assert_eq!(
                modes.get("/proj/a.js").copied().map(|m| m & 0o7777),
                Some(0o644)
            );
            assert_eq!(
                modes.get("/proj/b.js").copied().map(|m| m & 0o7777),
                Some(0o600)
            );
        }

        #[test]
        fn glob_no_match_returns_empty_contribution() {
            let mut fs = MemFilesystem::new();
            fs.add_dir(PROJECT_HOST).unwrap();
            fs.add_file(format!("{PROJECT_HOST}/only.txt"), b"x".to_vec())
                .unwrap();

            let project = nested_project();
            let workspace = workspace_with(&project);
            let outputs = vec![OutputSpec::parse("*.js").unwrap()];

            let result = resolve_output_files(&fs, &workspace, &project, &outputs).unwrap();
            assert!(result.is_empty());
        }

        #[test]
        fn glob_nested_double_star_recurses_into_subdirectories() {
            let mut fs = MemFilesystem::new();
            fs.add_dir(format!("{PROJECT_HOST}/dist")).unwrap();
            fs.add_dir(format!("{PROJECT_HOST}/dist/inner")).unwrap();
            fs.add_file(format!("{PROJECT_HOST}/dist/top.js"), b"top".to_vec())
                .unwrap();
            fs.add_file(
                format!("{PROJECT_HOST}/dist/inner/deep.js"),
                b"deep".to_vec(),
            )
            .unwrap();
            // Outside the `dist/` prefix; MUST NOT be matched.
            fs.add_file(format!("{PROJECT_HOST}/other.js"), b"other".to_vec())
                .unwrap();

            let project = nested_project();
            let workspace = workspace_with(&project);
            let outputs = vec![OutputSpec::parse("dist/**/*.js").unwrap()];

            let result = resolve_output_files(&fs, &workspace, &project, &outputs).unwrap();
            assert_eq!(result.len(), 2);
            assert_eq!(
                paths_of(&result),
                BTreeSet::from([
                    "/proj/dist/top.js".to_owned(),
                    "/proj/dist/inner/deep.js".to_owned(),
                ]),
            );
            assert_eq!(
                host_paths_of(&result),
                BTreeSet::from([
                    PathBuf::from("/ws/proj/dist/top.js"),
                    PathBuf::from("/ws/proj/dist/inner/deep.js"),
                ]),
            );
        }

        #[test]
        fn glob_symlink_to_file_records_link_path_with_target_mode() {
            // Mirrors the input-side symlink contract: both the
            // real file and the symlink that points at it are
            // distinct contributions; each one's mode comes from
            // Filesystem::permissions (which follows symlinks),
            // so both entries report the target file's mode bits.
            let mut fs = MemFilesystem::new();
            fs.add_dir(PROJECT_HOST).unwrap();
            fs.add_file_with_mode(
                format!("{PROJECT_HOST}/real.txt"),
                b"real bytes".to_vec(),
                0o600,
            )
            .unwrap();
            fs.add_symlink(
                format!("{PROJECT_HOST}/link.txt"),
                format!("{PROJECT_HOST}/real.txt"),
            )
            .unwrap();

            let project = nested_project();
            let workspace = workspace_with(&project);
            let outputs = vec![OutputSpec::parse("*.txt").unwrap()];

            let result = resolve_output_files(&fs, &workspace, &project, &outputs).unwrap();
            assert_eq!(
                result.len(),
                2,
                "both the real file and the symlink to it are distinct contributions",
            );
            assert_eq!(
                paths_of(&result),
                BTreeSet::from(["/proj/real.txt".to_owned(), "/proj/link.txt".to_owned(),]),
            );
            // Both entries see the target's mode bits, because
            // permissions follows the symlink.
            for file in &result {
                assert_eq!(
                    file.mode & 0o7777,
                    0o600,
                    "{} should report target's mode",
                    file.workspace_absolute_path,
                );
            }
        }

        #[test]
        fn implicit_mode_project_relative_literal_is_workspace_absolute() {
            // ProjectRoot::WorkspaceRoot: project-relative output
            // lands at the workspace root directly, with no extra
            // prefix.
            let mut fs = MemFilesystem::new();
            fs.add_dir(WORKSPACE_HOST).unwrap();
            fs.add_file_with_mode(
                format!("{WORKSPACE_HOST}/manifest.json"),
                b"{}".to_vec(),
                0o644,
            )
            .unwrap();

            let project = implicit_project();
            let workspace = workspace_with(&project);
            let outputs = vec![OutputSpec::parse("manifest.json").unwrap()];

            let result = resolve_output_files(&fs, &workspace, &project, &outputs).unwrap();
            assert_eq!(result.len(), 1);
            assert_eq!(result[0].workspace_absolute_path, "/manifest.json");
            assert_eq!(result[0].on_disk_path, PathBuf::from("/ws/manifest.json"));
            assert_eq!(result[0].mode & 0o7777, 0o644);
        }

        #[test]
        fn glob_walk_error_surfaces_output_pattern_resolution_failed() {
            // The project root directory does not exist on disk;
            // the glob walker's read_dir fails on entry and the
            // action's map_walk_error wraps it as
            // OutputPatternResolutionFailed (NOT the literal-side
            // OutputDeclaredButNotProduced shape, which only
            // applies to literal patterns).
            let fs = MemFilesystem::new();

            let project = nested_project();
            let workspace = workspace_with(&project);
            let outputs = vec![OutputSpec::parse("*.js").unwrap()];

            match resolve_output_files(&fs, &workspace, &project, &outputs) {
                Err(RunTaskError::OutputPatternResolutionFailed { root, .. }) => {
                    assert_eq!(root, PathBuf::from("/ws/proj"));
                }
                other => panic!("expected OutputPatternResolutionFailed, got {other:?}"),
            }
        }

        #[test]
        fn permissions_round_trip_through_set_permissions() {
            // Sanity check: a freshly-set mode is faithfully
            // surfaced via the output resolver. Guards against
            // future MemFilesystem regressions where set_permissions
            // and permissions get out of sync.
            let mut fs = MemFilesystem::new();
            fs.add_dir(PROJECT_HOST).unwrap();
            fs.add_file(format!("{PROJECT_HOST}/bin"), b"!".to_vec())
                .unwrap();
            fs.set_permissions(std::path::Path::new("/ws/proj/bin"), 0o751)
                .unwrap();

            let project = nested_project();
            let workspace = workspace_with(&project);
            let outputs = vec![OutputSpec::parse("bin").unwrap()];

            let result = resolve_output_files(&fs, &workspace, &project, &outputs).unwrap();
            assert_eq!(result.len(), 1);
            assert_eq!(result[0].mode & 0o7777, 0o751);
        }
    }

    // ---- end-to-end run_task ----

    mod end_to_end {
        use std::collections::{BTreeMap, BTreeSet};
        use std::path::PathBuf;
        use std::sync::Mutex;

        use nonempty::NonEmpty;

        use haz_cache::CacheWriter;
        use haz_dag::graph::TaskGraph;
        use haz_domain::action::TaskAction;
        use haz_domain::env::{EnvSettings, EnvVarName};
        use haz_domain::name::{ProjectName, TaskName};
        use haz_domain::path::{
            CanonicalPath, HazPath, OutputSpec, ProjectRoot, WorkspaceRootPath,
        };
        use haz_domain::project::Project;
        use haz_domain::settings::WorkspaceSettings;
        use haz_domain::settings::cache::HashAlgo;
        use haz_domain::task::Task;
        use haz_domain::task_id::TaskId;
        use haz_domain::workspace::Workspace;
        use haz_vfs::{Filesystem, FsError, WritableFilesystem};
        use haz_vfs_testing::MemFilesystem;

        use crate::mock_impl::{MockProcessSpawner, MockSpec};
        use haz_cache_key::BuildKeyError;

        use super::super::{
            CancelledRecord, CompletedRecord, RunContext, RunObserver, RunSource, RunState,
            RunTaskError, SkipRecord, run_task,
        };
        use tokio_util::sync::CancellationToken;

        const WORKSPACE_HOST: &str = "/ws";
        const PROJECT_HOST: &str = "/ws/proj";

        fn env_var(s: &str) -> EnvVarName {
            EnvVarName::try_new(s).unwrap()
        }

        fn task_id_for(project: &str, task: &str) -> TaskId {
            TaskId {
                project: ProjectName::try_new(project).unwrap(),
                task: TaskName::try_new(task).unwrap(),
            }
        }

        fn command(argv: &[&str]) -> TaskAction {
            TaskAction::Command(
                NonEmpty::from_vec(argv.iter().map(|s| (*s).to_owned()).collect())
                    .expect("non-empty argv"),
            )
        }

        /// Event captured by [`Recorder`]. The cache-hit branch and
        /// the fresh-run branch both flow through the same observer
        /// surface, so the captured event stream is mode-agnostic.
        #[derive(Debug, Clone, PartialEq, Eq)]
        enum RecordedEvent {
            Started(TaskId),
            Stdout(TaskId, Vec<u8>),
            Stderr(TaskId, Vec<u8>),
            Finished(TaskId, CompletedRecord),
        }

        /// Test-double [`RunObserver`]: records every lifecycle
        /// callback into an in-memory log the assertions inspect.
        #[derive(Default)]
        struct Recorder {
            events: Mutex<Vec<RecordedEvent>>,
        }

        impl Recorder {
            fn events(&self) -> Vec<RecordedEvent> {
                self.events.lock().unwrap().clone()
            }
        }

        impl RunObserver for Recorder {
            fn on_task_started(&self, task: &TaskId) {
                self.events
                    .lock()
                    .unwrap()
                    .push(RecordedEvent::Started(task.clone()));
            }
            fn on_stdout(&self, task: &TaskId, bytes: &[u8]) {
                self.events
                    .lock()
                    .unwrap()
                    .push(RecordedEvent::Stdout(task.clone(), bytes.to_vec()));
            }
            fn on_stderr(&self, task: &TaskId, bytes: &[u8]) {
                self.events
                    .lock()
                    .unwrap()
                    .push(RecordedEvent::Stderr(task.clone(), bytes.to_vec()));
            }
            fn on_task_finished(&self, task: &TaskId, record: &CompletedRecord) {
                self.events
                    .lock()
                    .unwrap()
                    .push(RecordedEvent::Finished(task.clone(), record.clone()));
            }
            fn on_task_skipped(&self, _task: &TaskId, _record: &SkipRecord) {
                // run_task never produces a Skipped, so this
                // recorder leaves the path unrecorded.
            }
            fn on_task_cancelled(&self, _task: &TaskId, _record: &CancelledRecord) {
                // run_task never produces a Cancelled until the
                // spawn-step cancellation flow lands; recorder
                // leaves the path unrecorded for now.
            }
        }

        /// Common end-to-end fixture: one `proj:build` task with the
        /// supplied action/env/outputs, a workspace + graph + cache
        /// backed by a fresh [`MemFilesystem`] rooted at `/ws/proj`.
        struct Fixture {
            cache: CacheWriter<MemFilesystem>,
            workspace: Workspace,
            graph: TaskGraph,
            task_id: TaskId,
            host_env: BTreeMap<EnvVarName, String>,
            cancel: CancellationToken,
        }

        impl Fixture {
            fn new(action: TaskAction, env: EnvSettings, outputs: Vec<OutputSpec>) -> Self {
                let mut fs = MemFilesystem::new();
                fs.add_dir(WORKSPACE_HOST).unwrap();
                fs.add_dir(PROJECT_HOST).unwrap();

                let task = Task {
                    name: TaskName::try_new("build").unwrap(),
                    action,
                    inputs: Vec::new(),
                    outputs,
                    deps: Vec::new(),
                    weak_deps: Vec::new(),
                    mutex: None,
                    env,
                };
                let project = Project {
                    name: ProjectName::try_new("proj").unwrap(),
                    root: ProjectRoot::Nested(
                        CanonicalPath::from_absolute(&HazPath::parse("/proj").unwrap()).unwrap(),
                    ),
                    tags: BTreeSet::new(),
                    tasks: BTreeMap::from([(task.name.clone(), task)]),
                };
                let task_id = task_id_for("proj", "build");
                let workspace = Workspace {
                    root: WorkspaceRootPath::try_new(PathBuf::from(WORKSPACE_HOST)).unwrap(),
                    projects: BTreeMap::from([(project.name.clone(), project)]),
                    overlays: BTreeMap::new(),
                    settings: WorkspaceSettings::default(),
                };
                let graph = TaskGraph {
                    nodes: BTreeSet::from([task_id.clone()]),
                    edges: BTreeSet::new(),
                };
                let cache =
                    CacheWriter::new(fs, std::path::Path::new(WORKSPACE_HOST), HashAlgo::Blake3);
                Self {
                    cache,
                    workspace,
                    graph,
                    task_id,
                    host_env: BTreeMap::new(),
                    cancel: CancellationToken::new(),
                }
            }
        }

        fn make_ctx<'a>(
            fixture: &'a Fixture,
            spawner: &'a MockProcessSpawner,
            observer: &'a Recorder,
        ) -> RunContext<'a, MemFilesystem, MockProcessSpawner, Recorder> {
            RunContext {
                fs: fixture.cache.fs(),
                cache: &fixture.cache,
                spawner,
                observer,
                workspace: &fixture.workspace,
                graph: &fixture.graph,
                host_env: &fixture.host_env,
                algo: HashAlgo::Blake3,
                cancel: &fixture.cancel,
            }
        }

        #[tokio::test]
        async fn cache_miss_success_records_observer_events_and_persists_manifest() {
            let fixture = Fixture::new(command(&["true"]), EnvSettings::default(), Vec::new());
            let spawner = MockProcessSpawner::new();
            spawner.push_spec(MockSpec {
                stdout: b"out\n".to_vec(),
                stderr: b"err\n".to_vec(),
                exit_code: 0,
                ..MockSpec::default()
            });
            let observer = Recorder::default();
            let ctx = make_ctx(&fixture, &spawner, &observer);

            let outcome = run_task(&ctx, &fixture.task_id, &BTreeMap::new(), 1_700_000_000)
                .await
                .expect("baseline run should succeed");

            assert_eq!(outcome.source, RunSource::FreshRun);
            assert_eq!(outcome.state, RunState::Succeeded);
            assert!(outcome.exit_status.is_some_and(|s| s.success()));
            assert_eq!(outcome.task, fixture.task_id);

            let events = observer.events();
            assert_eq!(events.len(), 4, "expected 4 events, got {events:?}");
            assert_eq!(events[0], RecordedEvent::Started(fixture.task_id.clone()));
            assert_eq!(
                events[1],
                RecordedEvent::Stdout(fixture.task_id.clone(), b"out\n".to_vec()),
            );
            assert_eq!(
                events[2],
                RecordedEvent::Stderr(fixture.task_id.clone(), b"err\n".to_vec()),
            );
            match &events[3] {
                RecordedEvent::Finished(id, finished_outcome) => {
                    assert_eq!(id, &fixture.task_id);
                    assert_eq!(finished_outcome, &outcome);
                }
                other => panic!("expected Finished event, got {other:?}"),
            }
        }

        #[tokio::test]
        async fn cache_miss_then_second_call_hits_cache() {
            let fixture = Fixture::new(command(&["true"]), EnvSettings::default(), Vec::new());
            let spawner = MockProcessSpawner::new();
            spawner.push_spec(MockSpec {
                stdout: b"out\n".to_vec(),
                stderr: b"err\n".to_vec(),
                exit_code: 0,
                ..MockSpec::default()
            });

            let observer1 = Recorder::default();
            let ctx1 = make_ctx(&fixture, &spawner, &observer1);
            let first = run_task(&ctx1, &fixture.task_id, &BTreeMap::new(), 1)
                .await
                .unwrap();
            assert_eq!(first.source, RunSource::FreshRun);

            // Second call: same key, manifest present, expect hit.
            let observer2 = Recorder::default();
            let ctx2 = make_ctx(&fixture, &spawner, &observer2);
            let second = run_task(&ctx2, &fixture.task_id, &BTreeMap::new(), 2)
                .await
                .unwrap();

            assert_eq!(second.source, RunSource::CacheHit);
            assert_eq!(second.state, RunState::Succeeded);
            assert!(second.exit_status.is_none());
            assert_eq!(second.stdout_hash, first.stdout_hash);
            assert_eq!(second.stderr_hash, first.stderr_hash);

            // The hit branch surfaces the recorded streams through
            // the observer (EXEC-017).
            let events = observer2.events();
            assert!(
                events.iter().any(|e| matches!(
                    e,
                    RecordedEvent::Stdout(_, bytes) if bytes == b"out\n"
                )),
                "missing recorded stdout in hit-path events: {events:?}",
            );
            assert!(
                events.iter().any(|e| matches!(
                    e,
                    RecordedEvent::Stderr(_, bytes) if bytes == b"err\n"
                )),
                "missing recorded stderr in hit-path events: {events:?}",
            );

            // The mock was only called once; the second run did NOT
            // respawn.
            assert_eq!(
                spawner.spawns().len(),
                1,
                "second call must not respawn, got {} total spawns",
                spawner.spawns().len(),
            );
        }

        #[tokio::test]
        async fn cache_018_failed_run_does_not_store_a_cache_entry() {
            // CACHE-018: a run whose process exit status is non-zero
            // (or signalled, or timed out) MUST NOT produce a cache
            // entry. The cache root MUST remain absent and a follow-up
            // invocation MUST observe a fresh run (not a hit) at the
            // same key.
            let fixture = Fixture::new(command(&["false"]), EnvSettings::default(), Vec::new());
            let spawner = MockProcessSpawner::new();
            spawner.push_spec(MockSpec {
                stdout: Vec::new(),
                stderr: b"error\n".to_vec(),
                exit_code: 1,
                ..MockSpec::default()
            });
            let observer = Recorder::default();
            let ctx = make_ctx(&fixture, &spawner, &observer);

            let outcome = run_task(&ctx, &fixture.task_id, &BTreeMap::new(), 1)
                .await
                .unwrap();

            assert_eq!(outcome.source, RunSource::FreshRun);
            assert_eq!(outcome.state, RunState::Failed);
            assert!(outcome.exit_status.is_some_and(|s| !s.success()));

            // The captured failure-side stderr still flowed through
            // the observer.
            assert!(
                observer.events().iter().any(|e| matches!(
                    e,
                    RecordedEvent::Stderr(_, bytes) if bytes == b"error\n"
                )),
                "failure-side stderr must reach the observer",
            );

            // Direct CACHE-018 check: the cache root MUST NOT have
            // been created. CACHE-010 mandates lazy creation on first
            // store, and a failed run is not a store; the cache
            // directory therefore stays absent.
            let cache_root_meta = fixture.cache.fs().metadata(fixture.cache.cache_root());
            assert!(
                matches!(cache_root_meta, Err(FsError::NotFound { .. })),
                "cache root must remain absent after a failed run; got {cache_root_meta:?}",
            );

            // Second call: nothing stored, expect another miss.
            spawner.push_spec(MockSpec {
                stdout: Vec::new(),
                stderr: b"again\n".to_vec(),
                exit_code: 1,
                ..MockSpec::default()
            });
            let observer2 = Recorder::default();
            let ctx2 = make_ctx(&fixture, &spawner, &observer2);
            let outcome2 = run_task(&ctx2, &fixture.task_id, &BTreeMap::new(), 2)
                .await
                .unwrap();
            assert_eq!(outcome2.source, RunSource::FreshRun);
            assert_eq!(spawner.spawns().len(), 2);
        }

        #[tokio::test]
        async fn cache_miss_success_stores_outputs_and_restore_overwrites_target() {
            // The task declares one output; pre-write it to
            // simulate the command producing it. After a fresh run,
            // the cache holds the output; a subsequent hit restores
            // the cached bytes even when the workspace copy has been
            // tampered with.
            let fixture = Fixture::new(
                command(&["true"]),
                EnvSettings::default(),
                vec![OutputSpec::parse("artifact.bin").unwrap()],
            );
            fixture
                .cache
                .fs()
                .write_file(
                    std::path::Path::new("/ws/proj/artifact.bin"),
                    b"artifact-bytes",
                )
                .unwrap();

            let spawner = MockProcessSpawner::new();
            spawner.push_spec(MockSpec::default());
            let observer = Recorder::default();
            let ctx = make_ctx(&fixture, &spawner, &observer);

            let first = run_task(&ctx, &fixture.task_id, &BTreeMap::new(), 1)
                .await
                .unwrap();
            assert_eq!(first.source, RunSource::FreshRun);
            assert_eq!(first.state, RunState::Succeeded);

            // Tamper with the target. The hit branch's restore
            // (CACHE-019) should overwrite this with the cached
            // bytes.
            fixture
                .cache
                .fs()
                .write_file(std::path::Path::new("/ws/proj/artifact.bin"), b"garbage")
                .unwrap();

            let observer2 = Recorder::default();
            let ctx2 = make_ctx(&fixture, &spawner, &observer2);
            let second = run_task(&ctx2, &fixture.task_id, &BTreeMap::new(), 2)
                .await
                .unwrap();
            assert_eq!(second.source, RunSource::CacheHit);

            let restored_bytes = fixture
                .cache
                .fs()
                .read(std::path::Path::new("/ws/proj/artifact.bin"))
                .unwrap();
            assert_eq!(restored_bytes, b"artifact-bytes");
        }

        #[tokio::test]
        async fn cache_023_hit_is_observationally_equivalent_to_fresh_run() {
            // CACHE-023: a cache hit MUST produce, from the
            // perspective of every external observer that consults
            // only declared outputs and captured streams, a state
            // indistinguishable from a successful fresh run.
            //
            // Setup: one task with one declared output and a mock
            // command that emits non-empty stdout AND stderr. Run
            // once (miss -> fresh). Delete the output file so the
            // second run cannot accidentally observe stale bytes.
            // Run again (hit -> restore). Compare the three
            // externally-visible byte streams.
            const OUTPUT_BYTES: &[u8] = b"\x00built\x01artifact\xFF";
            const STDOUT_BYTES: &[u8] = b"hello from the task\n";
            const STDERR_BYTES: &[u8] = b"warning: be careful\n";

            let fixture = Fixture::new(
                command(&["true"]),
                EnvSettings::default(),
                vec![OutputSpec::parse("artifact.bin").unwrap()],
            );

            // Pre-write the workspace file so the fresh run can
            // hash it into the cache as the task's "produced"
            // output.
            let artifact_path = std::path::Path::new("/ws/proj/artifact.bin");
            fixture
                .cache
                .fs()
                .write_file(artifact_path, OUTPUT_BYTES)
                .unwrap();

            let spawner = MockProcessSpawner::new();
            spawner.push_spec(MockSpec {
                stdout: STDOUT_BYTES.to_vec(),
                stderr: STDERR_BYTES.to_vec(),
                exit_code: 0,
                ..MockSpec::default()
            });

            // Fresh run.
            let observer_fresh = Recorder::default();
            let ctx_fresh = make_ctx(&fixture, &spawner, &observer_fresh);
            let fresh = run_task(&ctx_fresh, &fixture.task_id, &BTreeMap::new(), 1)
                .await
                .unwrap();
            assert_eq!(fresh.source, RunSource::FreshRun);
            assert_eq!(fresh.state, RunState::Succeeded);

            let fresh_output = fixture.cache.fs().read(artifact_path).unwrap();
            let fresh_stdout = concat_stdout(&observer_fresh.events());
            let fresh_stderr = concat_stderr(&observer_fresh.events());

            // Overwrite the workspace artefact with distinctive
            // garbage so the second run cannot accidentally
            // observe stale-but-equal bytes; the assertion below
            // proves the restore put the cached bytes back. The
            // cache entry under .haz/cache/ is untouched.
            fixture
                .cache
                .fs()
                .write_file(artifact_path, b"garbage-should-be-overwritten")
                .unwrap();

            // Second run: cache hit, restoration only.
            let observer_hit = Recorder::default();
            let ctx_hit = make_ctx(&fixture, &spawner, &observer_hit);
            let hit = run_task(&ctx_hit, &fixture.task_id, &BTreeMap::new(), 2)
                .await
                .unwrap();
            assert_eq!(hit.source, RunSource::CacheHit);
            assert_eq!(hit.state, RunState::Succeeded);

            let hit_output = fixture.cache.fs().read(artifact_path).unwrap();
            let hit_stdout = concat_stdout(&observer_hit.events());
            let hit_stderr = concat_stderr(&observer_hit.events());

            // The three externally-visible byte sequences MUST
            // match the fresh-run values exactly.
            assert_eq!(
                fresh_output, hit_output,
                "CACHE-023: declared output bytes differ between fresh and hit",
            );
            assert_eq!(
                fresh_stdout, hit_stdout,
                "CACHE-023: captured stdout differs between fresh and hit",
            );
            assert_eq!(
                fresh_stderr, hit_stderr,
                "CACHE-023: captured stderr differs between fresh and hit",
            );

            // And the bytes themselves agree with the source-of-
            // truth values, not some accidentally-equal corruption.
            assert_eq!(hit_output, OUTPUT_BYTES);
            assert_eq!(hit_stdout, STDOUT_BYTES);
            assert_eq!(hit_stderr, STDERR_BYTES);

            // Exactly one spawn happened (the fresh run); the hit
            // did not re-spawn.
            assert_eq!(spawner.spawns().len(), 1);
        }

        fn concat_stdout(events: &[RecordedEvent]) -> Vec<u8> {
            let mut acc = Vec::new();
            for event in events {
                if let RecordedEvent::Stdout(_, bytes) = event {
                    acc.extend_from_slice(bytes);
                }
            }
            acc
        }

        fn concat_stderr(events: &[RecordedEvent]) -> Vec<u8> {
            let mut acc = Vec::new();
            for event in events {
                if let RecordedEvent::Stderr(_, bytes) = event {
                    acc.extend_from_slice(bytes);
                }
            }
            acc
        }

        #[tokio::test]
        async fn task_not_in_workspace_surfaces_build_key_failed() {
            let fixture = Fixture::new(command(&["true"]), EnvSettings::default(), Vec::new());
            let spawner = MockProcessSpawner::new();
            let observer = Recorder::default();
            let ctx = make_ctx(&fixture, &spawner, &observer);

            let absent = task_id_for("absent_project", "build");

            match run_task(&ctx, &absent, &BTreeMap::new(), 1).await {
                Err(RunTaskError::BuildKeyFailed {
                    source: BuildKeyError::TaskNotInWorkspace { task },
                }) => assert_eq!(task, absent),
                other => panic!("expected BuildKeyFailed(TaskNotInWorkspace), got {other:?}"),
            }

            // on_task_started fires before the lookup phase: the
            // wrapper marks the lifecycle as entered before any
            // workspace resolution attempt. on_task_finished does
            // NOT fire because no CompletedRecord was produced.
            let events = observer.events();
            assert_eq!(
                events.len(),
                1,
                "expected one Started event, got {events:?}"
            );
            assert!(
                matches!(events[0], RecordedEvent::Started(ref id) if id == &absent),
                "expected Started(absent), got {events:?}",
            );
            // No spawn either.
            assert!(spawner.spawns().is_empty());
        }

        #[tokio::test]
        async fn spawn_plan_env_reflects_cache_008_runtime_view() {
            let env = EnvSettings {
                from_host: BTreeSet::from([env_var("PATH"), env_var("MISSING")]),
                overrides: BTreeMap::from([
                    (env_var("HAZ_MODE"), "ci".to_owned()),
                    (env_var("PATH"), "/override/bin".to_owned()),
                ]),
            };
            let mut fixture = Fixture::new(command(&["true"]), env, Vec::new());
            fixture
                .host_env
                .insert(env_var("PATH"), "/usr/bin".to_owned());
            fixture
                .host_env
                .insert(env_var("UNRELATED"), "should-not-appear".to_owned());

            let spawner = MockProcessSpawner::new();
            spawner.push_spec(MockSpec::default());
            let observer = Recorder::default();
            let ctx = make_ctx(&fixture, &spawner, &observer);

            let _ = run_task(&ctx, &fixture.task_id, &BTreeMap::new(), 1)
                .await
                .unwrap();

            let records = spawner.spawns();
            assert_eq!(records.len(), 1);
            let env_vec = &records[0].plan.env;
            let env_names: BTreeSet<String> = env_vec
                .iter()
                .map(|(k, _)| k.to_str().unwrap().to_owned())
                .collect();
            // Override-only entries propagate; from_host present
            // names propagate (with override winning on
            // collision); from_host absent names are excluded;
            // unrelated host names are excluded.
            assert_eq!(
                env_names,
                BTreeSet::from(["HAZ_MODE".to_owned(), "PATH".to_owned()]),
            );

            let path_value = env_vec
                .iter()
                .find(|(k, _)| k.to_str() == Some("PATH"))
                .map(|(_, v)| v.to_str().unwrap().to_owned());
            assert_eq!(path_value.as_deref(), Some("/override/bin"));
        }
    }
}