haz-exec 0.1.0

//! Cache-key derivation for one task in a workspace.
//!
//! [`build_cache_key`] is the executor-side wrapper around
//! [`haz_cache::CacheKeyBuilder`]. It gathers the four normative
//! ingredients of `CACHE-005..008` from the validated workspace
//! state, the dependency graph, the host environment snapshot, and
//! the supplied per-predecessor stream hashes; hands them to the
//! cache library; and returns the resulting [`CacheKey`].
//!
//! The function is pure and synchronous. Filesystem access happens
//! through the supplied [`Filesystem`] handle; no `tokio` runtime is
//! required. Callers that want to off-load the file-hashing work
//! from an async context can wrap the call in
//! [`tokio::task::spawn_blocking`].

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

use snafu::{ResultExt, Snafu};

use haz_cache::{
    CacheKey, CacheKeyBuilder, CacheKeyInputs, EnvContribution, Hasher, InputFile,
    PredecessorStreams,
};
use haz_dag::edge::EdgeKind;
use haz_dag::graph::TaskGraph;
use haz_domain::env::{EnvSettings, EnvVarName};
use haz_domain::path::{HazPath, InputSpec, PathPattern, ProjectRoot};
use haz_domain::project::Project;
use haz_domain::settings::cache::HashAlgo;
use haz_domain::task_id::TaskId;
use haz_domain::workspace::Workspace;
use haz_vfs::{EntryKind, Filesystem, FsError};

use crate::pattern_walk::{
    GlobMatchAction, GlobWalk, glob_walk_origin, host_path_from_segments,
    literal_workspace_segments, workspace_absolute_string_from_segments,
};

/// Hashes of one hard-edge predecessor's captured stdout and stderr,
/// as recorded by its prior successful run (`CACHE-007`).
///
/// The executor maintains a `BTreeMap<TaskId, PredecessorStreamHashes>`
/// as tasks complete and threads it into [`build_cache_key`]; this
/// type is a small newtype around the pair so the entries the map
/// carries are self-describing.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct PredecessorStreamHashes {
    /// Hash of the predecessor's stdout under the active hash
    /// function.
    pub stdout_hash: [u8; 32],
    /// Hash of the predecessor's stderr under the active hash
    /// function.
    pub stderr_hash: [u8; 32],
}

/// Failure modes of [`build_cache_key`].
#[derive(Debug, Snafu)]
#[snafu(visibility(pub(crate)))]
pub enum BuildKeyError {
    /// The supplied [`TaskId`] does not name a task that exists in
    /// the supplied [`Workspace`]. Either the project is unknown or
    /// the project does not declare the task.
    #[snafu(display("task not in workspace: {task}"))]
    TaskNotInWorkspace {
        /// The unresolved task identifier.
        task: TaskId,
    },

    /// One of the target task's hard-edge predecessors does not have
    /// an entry in the supplied predecessor-stream-hash map.
    /// Typically a scheduler bug: the predecessor must complete (and
    /// therefore have its stream hashes recorded) before its
    /// downstream task is keyed.
    #[snafu(display("missing recorded stream hashes for hard-edge predecessor: {predecessor}"))]
    PredecessorStreamsMissing {
        /// The predecessor that lacked recorded stream hashes.
        predecessor: TaskId,
    },

    /// Walking the filesystem to resolve a task's `inputs` patterns
    /// failed. Surfaces the originating [`FsError`] and the absolute
    /// path the walk was attempting.
    #[snafu(display(
        "failed to resolve input patterns under: {}: {source}",
        root.display()
    ))]
    InputPatternResolutionFailed {
        /// The absolute path that was being walked when the failure
        /// occurred (typically the project root or the workspace
        /// root, depending on the spec form).
        root: PathBuf,
        /// Underlying filesystem error.
        source: FsError,
    },

    /// Reading the bytes of one matched input file failed.
    #[snafu(display(
        "failed to read input file at: {}: {source}",
        path.display()
    ))]
    InputContentReadFailed {
        /// Absolute path of the file the executor tried to read.
        path: PathBuf,
        /// Underlying filesystem error.
        source: FsError,
    },

    /// One of the matched input paths is a non-regular file kind the
    /// executor cannot hash (a socket, a FIFO, a block or character
    /// device). Symbolic links to regular files are followed
    /// transparently per `CACHE-006` and do NOT trigger this
    /// variant.
    #[snafu(display("input path is not a regular file: {}", path.display()))]
    InputNotARegularFile {
        /// Absolute path of the offending entry.
        path: PathBuf,
    },
}

/// Derive the cache key for one task in `workspace` per
/// `CACHE-001..009`.
///
/// The function gathers the four canonical ingredients of `CACHE-005`
/// to `CACHE-008` from the supplied state:
///
/// - The task's [`TaskAction`](haz_domain::action::TaskAction), via
///   workspace lookup.
/// - The files matched by the task's `inputs` patterns, with each
///   file's content hash under `algo`. Patterns are resolved through
///   `fs`.
/// - The hard-edge predecessors enumerated from `graph`, joined with
///   their captured stream hashes from `predecessor_streams`.
/// - The environment contribution: the task's `from_host` allow-list
///   resolved against `host_env`, and the task's `overrides` map.
///
/// `host_env` is the caller-snapshotted host environment, restricted
/// to names that pass [`EnvVarName`] validation. Only names actually
/// present in the host process appear in the map; a name in the
/// task's `from_host` allow-list that does not appear in `host_env`
/// is treated as absent (`CACHE-008`'s `0x00` marker). The function
/// does NOT read `std::env`. Callers SHOULD snapshot the host
/// environment once per `haz` run and thread the same snapshot into
/// every cache-key derivation in that run.
///
/// `predecessor_streams` is the executor's running record of which
/// predecessor produced which stream hashes. It MUST contain an
/// entry for every hard-edge predecessor of `task` per `graph`, or
/// the function returns [`BuildKeyError::PredecessorStreamsMissing`].
///
/// # Errors
///
/// See [`BuildKeyError`] for the failure modes.
pub fn build_cache_key<F: Filesystem>(
    fs: &F,
    workspace: &Workspace,
    graph: &TaskGraph,
    task: &TaskId,
    host_env: &BTreeMap<EnvVarName, String>,
    predecessor_streams: &BTreeMap<TaskId, PredecessorStreamHashes>,
    algo: HashAlgo,
) -> Result<CacheKey, BuildKeyError> {
    let project = workspace
        .projects
        .get(&task.project)
        .ok_or_else(|| BuildKeyError::TaskNotInWorkspace { task: task.clone() })?;
    let task_def = project
        .tasks
        .get(&task.task)
        .ok_or_else(|| BuildKeyError::TaskNotInWorkspace { task: task.clone() })?;

    let from_host_resolved = resolve_from_host_values(&task_def.env, host_env);
    let predecessors_owned = collect_hard_predecessor_streams(graph, task, predecessor_streams)?;
    let input_files_owned = resolve_input_files(fs, workspace, project, &task_def.inputs, algo)?;

    let input_files: Vec<InputFile<'_>> = input_files_owned
        .iter()
        .map(|f| InputFile {
            workspace_absolute_path: &f.workspace_absolute_path,
            content_hash: f.content_hash,
        })
        .collect();

    let hard_predecessors: Vec<PredecessorStreams<'_>> = predecessors_owned
        .iter()
        .map(|(id, hashes)| PredecessorStreams {
            project: &id.project,
            task: &id.task,
            stdout_hash: hashes.stdout_hash,
            stderr_hash: hashes.stderr_hash,
        })
        .collect();

    let env = EnvContribution {
        from_host: &from_host_resolved,
        overrides: &task_def.env.overrides,
    };

    let inputs = CacheKeyInputs {
        action: &task_def.action,
        input_files: &input_files,
        hard_predecessors: &hard_predecessors,
        env: &env,
    };

    Ok(CacheKeyBuilder::new(algo).finish(&inputs))
}

/// Enumerate the hard-edge predecessors of `target` in `graph` and
/// pair each with its recorded stream hashes from `streams`.
///
/// Order in the returned vector follows iteration order of
/// `graph.edges`; the cache library re-sorts predecessors into
/// canonical `(ProjectName, TaskName)` order during contribution
/// (`CACHE-009`), so the caller MAY pass them as enumerated.
///
/// # Errors
///
/// Returns [`BuildKeyError::PredecessorStreamsMissing`] when one of
/// the enumerated predecessors does not have an entry in `streams`.
fn collect_hard_predecessor_streams<'g>(
    graph: &'g TaskGraph,
    target: &TaskId,
    streams: &BTreeMap<TaskId, PredecessorStreamHashes>,
) -> Result<Vec<(&'g TaskId, PredecessorStreamHashes)>, BuildKeyError> {
    graph
        .edges
        .iter()
        .filter(|e| e.kind == EdgeKind::Hard && &e.to == target)
        .map(|e| {
            streams
                .get(&e.from)
                .copied()
                .map(|hashes| (&e.from, hashes))
                .ok_or_else(|| BuildKeyError::PredecessorStreamsMissing {
                    predecessor: e.from.clone(),
                })
        })
        .collect()
}

/// Resolve a task's `from_host` allow-list against a host-environment
/// snapshot, producing the [`Option<String>`]-valued map that
/// [`haz_cache::EnvContribution`] expects.
///
/// For every name in `env.from_host`, the result carries the value
/// found in `host_env` (`Some(value)`) or [`None`] when the name is
/// not present in the snapshot. The `overrides` map of `env` is NOT
/// consulted here: the cache library applies the precedence rule of
/// `CACHE-008` (overrides win on name collision) inside its own
/// canonical-serialisation step, so the caller may pass `from_host`
/// untouched.
fn resolve_from_host_values(
    env: &EnvSettings,
    host_env: &BTreeMap<EnvVarName, String>,
) -> BTreeMap<EnvVarName, Option<String>> {
    env.from_host
        .iter()
        .map(|name| (name.clone(), host_env.get(name).cloned()))
        .collect()
}

/// One file resolved from a task's `inputs` patterns, paired with the
/// hash of its content under the active hash function.
///
/// The `workspace_absolute_path` carries the workspace-absolute string
/// form (e.g. `/lib_core/src/main.rs`), suitable for direct conversion
/// into [`haz_cache::InputFile`]'s borrowed `&str` field. The cache
/// library re-sorts inputs into canonical order per `CACHE-009`, so the
/// order in which [`resolve_input_files`] appends matches is
/// unconstrained.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OwnedInputFile {
    /// Workspace-absolute path of the matched input file.
    pub workspace_absolute_path: String,
    /// Content hash of the file's bytes under the active hash
    /// function.
    pub content_hash: [u8; 32],
}

/// Resolve every [`InputSpec`] declared on a task against `fs` and
/// hash the bytes of each matched file under `algo`.
///
/// Each spec contributes zero or more [`OwnedInputFile`]s to the
/// returned vector. A [`PathPattern::Literal`] always contributes
/// exactly one entry on success; a non-matching glob contributes
/// none. Errors surface the first failure encountered while walking
/// or reading; later inputs in the same task are not attempted.
///
/// Symlink semantics follow `CACHE-006`: the workspace-absolute path
/// recorded is the matching entry's own path (so two symlinks to the
/// same target produce two distinct contributions), while the bytes
/// fed into the hash come from the symlink's canonical target
/// ([`Filesystem::read`] follows symlinks).
///
/// # Errors
///
/// Returns a [`BuildKeyError`] variant describing the first
/// failure encountered: a literal input that resolves to a missing
/// path or non-regular-file, a glob walk that fails on metadata or
/// recursion, or a content read that fails. Caller responsibility
/// to surface as a user-facing diagnostic.
pub fn resolve_input_files<F: Filesystem>(
    fs: &F,
    workspace: &Workspace,
    project: &Project,
    inputs: &[InputSpec],
    algo: HashAlgo,
) -> Result<Vec<OwnedInputFile>, BuildKeyError> {
    let workspace_host = workspace.root.as_path();
    let action = InputAction { algo };
    let mut out = Vec::new();

    for spec in inputs {
        match spec.pattern() {
            PathPattern::Literal(haz_path) => {
                resolve_literal(
                    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)
}

fn resolve_literal<F: Filesystem>(
    fs: &F,
    workspace_host: &Path,
    project_root: &ProjectRoot,
    haz_path: &HazPath,
    action: &InputAction,
    out: &mut Vec<OwnedInputFile>,
) -> Result<(), BuildKeyError> {
    let ws_segments = literal_workspace_segments(haz_path, project_root);
    let host = host_path_from_segments(workspace_host, &ws_segments);

    let meta = fs
        .metadata(&host)
        .context(InputPatternResolutionFailedSnafu { root: host.clone() })?;
    if meta.kind != EntryKind::File {
        return Err(BuildKeyError::InputNotARegularFile { 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 input resolution: read the matched file's
/// bytes and fold them into [`OwnedInputFile::content_hash`] under
/// the active hash algorithm. Walk-level [`FsError`]s surface as
/// [`BuildKeyError::InputPatternResolutionFailed`].
struct InputAction {
    algo: HashAlgo,
}

impl<F: Filesystem> GlobMatchAction<F> for InputAction {
    type Output = OwnedInputFile;
    type Error = BuildKeyError;

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

    fn on_match(
        &self,
        fs: &F,
        host_path: &Path,
        workspace_absolute_path: String,
        out: &mut Vec<OwnedInputFile>,
    ) -> Result<(), BuildKeyError> {
        let bytes = fs.read(host_path).context(InputContentReadFailedSnafu {
            path: host_path.to_path_buf(),
        })?;
        let mut hasher = Hasher::new(self.algo);
        hasher.update(&bytes);
        out.push(OwnedInputFile {
            workspace_absolute_path,
            content_hash: hasher.finalize(),
        });
        Ok(())
    }
}

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

    use haz_dag::edge::{Edge, EdgeKind};
    use haz_dag::graph::TaskGraph;
    use haz_domain::env::{EnvSettings, EnvVarName};
    use haz_domain::name::{ProjectName, TaskName};
    use haz_domain::task_id::TaskId;

    use super::{
        BuildKeyError, PredecessorStreamHashes, collect_hard_predecessor_streams,
        resolve_from_host_values,
    };

    fn name(s: &str) -> EnvVarName {
        EnvVarName::try_new(s).expect("valid env var name")
    }

    fn settings(from_host: &[&str], overrides: &[(&str, &str)]) -> EnvSettings {
        EnvSettings {
            from_host: from_host.iter().map(|s| name(s)).collect::<BTreeSet<_>>(),
            overrides: overrides
                .iter()
                .map(|(k, v)| (name(k), (*v).to_owned()))
                .collect::<BTreeMap<_, _>>(),
        }
    }

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

    #[test]
    fn empty_from_host_yields_empty_map() {
        let env = settings(&[], &[]);
        let result = resolve_from_host_values(&env, &host(&[]));
        assert!(result.is_empty());
    }

    #[test]
    fn name_present_in_host_resolves_to_some() {
        let env = settings(&["PATH"], &[]);
        let result = resolve_from_host_values(&env, &host(&[("PATH", "/usr/bin")]));
        assert_eq!(
            result.get(&name("PATH")).cloned(),
            Some(Some("/usr/bin".to_owned()))
        );
    }

    #[test]
    fn name_absent_from_host_resolves_to_none() {
        let env = settings(&["NEVER_SET"], &[]);
        let result = resolve_from_host_values(&env, &host(&[("OTHER", "v")]));
        assert_eq!(result.get(&name("NEVER_SET")).cloned(), Some(None));
    }

    #[test]
    fn empty_string_in_host_is_distinct_from_absent() {
        let env = settings(&["X"], &[]);
        let result_empty = resolve_from_host_values(&env, &host(&[("X", "")]));
        let result_absent = resolve_from_host_values(&env, &host(&[]));
        assert_eq!(
            result_empty.get(&name("X")).cloned(),
            Some(Some(String::new()))
        );
        assert_eq!(result_absent.get(&name("X")).cloned(), Some(None));
        assert_ne!(
            result_empty.get(&name("X")).unwrap(),
            result_absent.get(&name("X")).unwrap(),
            "empty value MUST NOT collapse with absence"
        );
    }

    #[test]
    fn from_host_iteration_is_lexicographically_sorted() {
        // BTreeMap collection preserves the sort, but the assertion
        // is the load-bearing one for CACHE-009: future-me changing
        // the result type away from BTreeMap would break key
        // determinism silently.
        let env = settings(&["ZULU", "ALPHA", "BRAVO"], &[]);
        let h = host(&[("ALPHA", "a"), ("BRAVO", "b"), ("ZULU", "z")]);
        let result = resolve_from_host_values(&env, &h);
        let names: Vec<&str> = result
            .keys()
            .map(|n| AsRef::<str>::as_ref(n.as_ref()))
            .collect();
        assert_eq!(names, vec!["ALPHA", "BRAVO", "ZULU"]);
    }

    #[test]
    fn overrides_are_not_consulted_by_this_helper() {
        // resolve_from_host_values intentionally ignores env.overrides;
        // applying the override-wins precedence is haz_cache's job.
        // The result here MUST contain the from_host name regardless
        // of whether the same name is also overridden.
        let env = settings(&["X"], &[("X", "override-val")]);
        let result = resolve_from_host_values(&env, &host(&[("X", "host-val")]));
        assert_eq!(
            result.get(&name("X")).cloned(),
            Some(Some("host-val".to_owned()))
        );
    }

    #[test]
    fn unrelated_host_entries_do_not_appear() {
        let env = settings(&["WANTED"], &[]);
        let h = host(&[("WANTED", "yes"), ("UNRELATED", "no")]);
        let result = resolve_from_host_values(&env, &h);
        assert_eq!(result.len(), 1);
        assert!(!result.contains_key(&name("UNRELATED")));
    }

    // ---- collect_hard_predecessor_streams ----

    fn task_id(project: &str, task: &str) -> TaskId {
        TaskId {
            project: ProjectName::from_str(project).expect("project name"),
            task: TaskName::from_str(task).expect("task name"),
        }
    }

    fn edge(from: TaskId, to: TaskId, kind: EdgeKind) -> Edge {
        Edge { from, to, kind }
    }

    fn streams(stdout: u8, stderr: u8) -> PredecessorStreamHashes {
        PredecessorStreamHashes {
            stdout_hash: [stdout; 32],
            stderr_hash: [stderr; 32],
        }
    }

    fn graph(nodes: &[TaskId], edges: &[Edge]) -> TaskGraph {
        TaskGraph {
            nodes: nodes.iter().cloned().collect::<BTreeSet<_>>(),
            edges: edges.iter().cloned().collect::<BTreeSet<_>>(),
        }
    }

    #[test]
    fn no_predecessors_yields_empty_vec() {
        let target = task_id("p", "alone");
        let g = graph(std::slice::from_ref(&target), &[]);
        let recorded: BTreeMap<TaskId, PredecessorStreamHashes> = BTreeMap::new();
        let result = collect_hard_predecessor_streams(&g, &target, &recorded).unwrap();
        assert!(result.is_empty());
    }

    #[test]
    fn single_hard_predecessor_is_returned() {
        let pred = task_id("p", "pre");
        let target = task_id("p", "main");
        let g = graph(
            &[pred.clone(), target.clone()],
            &[edge(pred.clone(), target.clone(), EdgeKind::Hard)],
        );
        let mut recorded = BTreeMap::new();
        recorded.insert(pred.clone(), streams(0xAA, 0xBB));

        let result = collect_hard_predecessor_streams(&g, &target, &recorded).unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].0, &pred);
        assert_eq!(result[0].1, streams(0xAA, 0xBB));
    }

    #[test]
    fn soft_and_producer_matching_edges_do_not_count() {
        let target = task_id("p", "main");
        let soft_pred = task_id("p", "soft");
        let pm_pred = task_id("p", "pm");
        let g = graph(
            &[target.clone(), soft_pred.clone(), pm_pred.clone()],
            &[
                edge(soft_pred, target.clone(), EdgeKind::Soft),
                edge(pm_pred, target.clone(), EdgeKind::ProducerMatching),
            ],
        );
        let recorded: BTreeMap<TaskId, PredecessorStreamHashes> = BTreeMap::new();
        let result = collect_hard_predecessor_streams(&g, &target, &recorded).unwrap();
        assert!(
            result.is_empty(),
            "only Hard edges contribute to CACHE-007; got {} predecessors",
            result.len(),
        );
    }

    #[test]
    fn outgoing_hard_edges_are_not_predecessors() {
        let target = task_id("p", "main");
        let dep_of_target = task_id("p", "downstream");
        let g = graph(
            &[target.clone(), dep_of_target.clone()],
            &[edge(target.clone(), dep_of_target, EdgeKind::Hard)],
        );
        let recorded: BTreeMap<TaskId, PredecessorStreamHashes> = BTreeMap::new();
        let result = collect_hard_predecessor_streams(&g, &target, &recorded).unwrap();
        assert!(result.is_empty());
    }

    #[test]
    fn multiple_hard_predecessors_are_all_returned() {
        let target = task_id("p", "main");
        let a = task_id("alpha", "build");
        let b = task_id("beta", "build");
        let g = graph(
            &[target.clone(), a.clone(), b.clone()],
            &[
                edge(a.clone(), target.clone(), EdgeKind::Hard),
                edge(b.clone(), target.clone(), EdgeKind::Hard),
            ],
        );
        let mut recorded = BTreeMap::new();
        recorded.insert(a.clone(), streams(0x01, 0x02));
        recorded.insert(b.clone(), streams(0x03, 0x04));

        let result = collect_hard_predecessor_streams(&g, &target, &recorded).unwrap();
        assert_eq!(result.len(), 2);
        let names: BTreeSet<TaskId> = result.iter().map(|(id, _)| (*id).clone()).collect();
        assert!(names.contains(&a));
        assert!(names.contains(&b));
    }

    #[test]
    fn missing_stream_hashes_yield_predecessor_streams_missing() {
        let pred = task_id("p", "pre");
        let target = task_id("p", "main");
        let g = graph(
            &[pred.clone(), target.clone()],
            &[edge(pred.clone(), target.clone(), EdgeKind::Hard)],
        );
        let recorded: BTreeMap<TaskId, PredecessorStreamHashes> = BTreeMap::new();

        match collect_hard_predecessor_streams(&g, &target, &recorded) {
            Err(BuildKeyError::PredecessorStreamsMissing { predecessor }) => {
                assert_eq!(predecessor, pred);
            }
            Err(other) => panic!("expected PredecessorStreamsMissing, got {other:?}"),
            Ok(v) => panic!("expected error, got Ok with {} entries", v.len()),
        }
    }

    #[test]
    fn mixed_edge_kinds_into_target_select_only_hard() {
        let target = task_id("p", "main");
        let hard_pred = task_id("p", "hard");
        let soft_pred = task_id("p", "soft");
        let g = graph(
            &[target.clone(), hard_pred.clone(), soft_pred.clone()],
            &[
                edge(hard_pred.clone(), target.clone(), EdgeKind::Hard),
                edge(soft_pred, target.clone(), EdgeKind::Soft),
            ],
        );
        let mut recorded = BTreeMap::new();
        recorded.insert(hard_pred.clone(), streams(0x10, 0x20));

        let result = collect_hard_predecessor_streams(&g, &target, &recorded).unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].0, &hard_pred);
    }

    // ---- resolve_input_files ----

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

        use haz_cache::Hasher;
        use haz_domain::path::{CanonicalPath, HazPath, InputSpec, ProjectRoot, WorkspaceRootPath};
        use haz_domain::project::Project;
        use haz_domain::settings::WorkspaceSettings;
        use haz_domain::settings::cache::HashAlgo;
        use haz_domain::workspace::Workspace;
        use haz_vfs::{FsError, MemFilesystem};

        use super::super::BuildKeyError;
        use super::super::{OwnedInputFile, resolve_input_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 hash_of(algo: HashAlgo, bytes: &[u8]) -> [u8; 32] {
            let mut h = Hasher::new(algo);
            h.update(bytes);
            h.finalize()
        }

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

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

            let project = nested_project();
            let workspace = workspace_with(&project);
            let inputs = vec![InputSpec::parse("file.txt").unwrap()];

            let result =
                resolve_input_files(&fs, &workspace, &project, &inputs, HashAlgo::Blake3).unwrap();
            assert_eq!(result.len(), 1);
            assert_eq!(result[0].workspace_absolute_path, "/proj/file.txt");
            assert_eq!(result[0].content_hash, hash_of(HashAlgo::Blake3, b"hello"));
        }

        #[test]
        fn literal_workspace_absolute_resolves_under_workspace_root() {
            // A workspace-absolute literal (`/other/data.bin`) bypasses
            // the bearing project's root entirely; the file is read
            // from <workspace_root>/other/data.bin.
            let mut fs = MemFilesystem::new();
            fs.add_dir("/ws/other").unwrap();
            fs.add_file("/ws/other/data.bin", b"\xDE\xAD\xBE\xEF".to_vec())
                .unwrap();
            // Project root is /ws/proj; the absolute literal must not
            // be interpreted relative to it.
            fs.add_dir(PROJECT_HOST).unwrap();

            let project = nested_project();
            let workspace = workspace_with(&project);
            let inputs = vec![InputSpec::parse("/other/data.bin").unwrap()];

            let result =
                resolve_input_files(&fs, &workspace, &project, &inputs, HashAlgo::Blake3).unwrap();
            assert_eq!(result.len(), 1);
            assert_eq!(result[0].workspace_absolute_path, "/other/data.bin");
            assert_eq!(
                result[0].content_hash,
                hash_of(HashAlgo::Blake3, b"\xDE\xAD\xBE\xEF")
            );
        }

        #[test]
        fn literal_missing_file_surfaces_pattern_resolution_failure() {
            let mut fs = MemFilesystem::new();
            fs.add_dir(PROJECT_HOST).unwrap();

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

            match resolve_input_files(&fs, &workspace, &project, &inputs, HashAlgo::Blake3) {
                Err(BuildKeyError::InputPatternResolutionFailed { root, source }) => {
                    assert_eq!(root, PathBuf::from("/ws/proj/absent.txt"));
                    assert!(
                        matches!(source, FsError::NotFound { .. }),
                        "expected NotFound source, got {source:?}",
                    );
                }
                other => panic!("expected InputPatternResolutionFailed, got {other:?}"),
            }
        }

        #[test]
        fn literal_pointing_at_directory_surfaces_input_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 inputs = vec![InputSpec::parse("subdir").unwrap()];

            match resolve_input_files(&fs, &workspace, &project, &inputs, HashAlgo::Blake3) {
                Err(BuildKeyError::InputNotARegularFile { path }) => {
                    assert_eq!(path, PathBuf::from("/ws/proj/subdir"));
                }
                other => panic!("expected InputNotARegularFile, got {other:?}"),
            }
        }

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

            let project = nested_project();
            let workspace = workspace_with(&project);
            let inputs = vec![InputSpec::parse("*.rs").unwrap()];

            let result =
                resolve_input_files(&fs, &workspace, &project, &inputs, HashAlgo::Blake3).unwrap();
            assert_eq!(result.len(), 2);
            assert_eq!(
                paths_of(&result),
                BTreeSet::from(["/proj/a.rs".to_owned(), "/proj/b.rs".to_owned()]),
            );
            for file in &result {
                let bytes: &[u8] = if file.workspace_absolute_path == "/proj/a.rs" {
                    b"a"
                } else {
                    b"b"
                };
                assert_eq!(file.content_hash, hash_of(HashAlgo::Blake3, bytes));
            }
        }

        #[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 inputs = vec![InputSpec::parse("*.rs").unwrap()];

            let result =
                resolve_input_files(&fs, &workspace, &project, &inputs, HashAlgo::Blake3).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}/src")).unwrap();
            fs.add_dir(format!("{PROJECT_HOST}/src/inner")).unwrap();
            fs.add_file(
                format!("{PROJECT_HOST}/src/top.rs"),
                b"top contents".to_vec(),
            )
            .unwrap();
            fs.add_file(
                format!("{PROJECT_HOST}/src/inner/deep.rs"),
                b"deep contents".to_vec(),
            )
            .unwrap();
            // Outside the `src/` prefix; MUST NOT be matched.
            fs.add_file(format!("{PROJECT_HOST}/other.rs"), b"other".to_vec())
                .unwrap();

            let project = nested_project();
            let workspace = workspace_with(&project);
            let inputs = vec![InputSpec::parse("src/**/*.rs").unwrap()];

            let result =
                resolve_input_files(&fs, &workspace, &project, &inputs, HashAlgo::Blake3).unwrap();
            assert_eq!(result.len(), 2);
            assert_eq!(
                paths_of(&result),
                BTreeSet::from([
                    "/proj/src/top.rs".to_owned(),
                    "/proj/src/inner/deep.rs".to_owned(),
                ]),
            );
        }

        #[test]
        fn glob_symlink_to_file_records_link_path_with_target_bytes_hash() {
            let mut fs = MemFilesystem::new();
            fs.add_dir(PROJECT_HOST).unwrap();
            fs.add_file(format!("{PROJECT_HOST}/real.txt"), b"real bytes".to_vec())
                .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 inputs = vec![InputSpec::parse("*.txt").unwrap()];

            let result =
                resolve_input_files(&fs, &workspace, &project, &inputs, HashAlgo::Blake3).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(),]),
            );
            // CACHE-006: the symlink contributes the canonical
            // target's content hash, but its own path.
            let expected_hash = hash_of(HashAlgo::Blake3, b"real bytes");
            for file in &result {
                assert_eq!(file.content_hash, expected_hash);
            }
        }

        #[test]
        fn implicit_mode_project_relative_literal_is_workspace_absolute() {
            // ProjectRoot::WorkspaceRoot means the project lives at the
            // workspace root; project-relative paths land at the
            // workspace root directly, with no extra prefix.
            let mut fs = MemFilesystem::new();
            fs.add_dir(WORKSPACE_HOST).unwrap();
            fs.add_file(format!("{WORKSPACE_HOST}/at_root.txt"), b"r".to_vec())
                .unwrap();

            let project = implicit_project();
            let workspace = workspace_with(&project);
            let inputs = vec![InputSpec::parse("at_root.txt").unwrap()];

            let result =
                resolve_input_files(&fs, &workspace, &project, &inputs, HashAlgo::Blake3).unwrap();
            assert_eq!(result.len(), 1);
            assert_eq!(result[0].workspace_absolute_path, "/at_root.txt");
            assert_eq!(result[0].content_hash, hash_of(HashAlgo::Blake3, b"r"));
        }
    }

    // ---- end-to-end build_cache_key ----

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

        use nonempty::NonEmpty;

        use haz_cache::CacheKey;
        use haz_dag::edge::{Edge, EdgeKind};
        use haz_dag::graph::TaskGraph;
        use haz_domain::action::TaskAction;
        use haz_domain::env::{EnvSettings, EnvVarName};
        use haz_domain::mutex::{Mutex, MutexMode, MutexScope};
        use haz_domain::name::{MutexName, ProjectName, TaskName};
        use haz_domain::path::{CanonicalPath, HazPath, InputSpec, 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::{MemFilesystem, WritableFilesystem};

        use super::super::{BuildKeyError, PredecessorStreamHashes, build_cache_key};

        struct E2eState {
            fs: MemFilesystem,
            workspace: Workspace,
            graph: TaskGraph,
            task_id: TaskId,
            host_env: BTreeMap<EnvVarName, String>,
            predecessor_streams: BTreeMap<TaskId, PredecessorStreamHashes>,
        }

        impl E2eState {
            fn key(&self) -> CacheKey {
                build_cache_key(
                    &self.fs,
                    &self.workspace,
                    &self.graph,
                    &self.task_id,
                    &self.host_env,
                    &self.predecessor_streams,
                    HashAlgo::Blake3,
                )
                .expect("baseline state must yield a valid key")
            }
        }

        fn env_var(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()
        }

        /// Default state used by every e2e test.
        ///
        /// Exercises all four CACHE-005..008 ingredients in one shot:
        /// a Command action, a single literal input file, one
        /// hard-edge predecessor with recorded stream hashes, and one
        /// from_host-allowed env var present in the host snapshot.
        fn baseline() -> E2eState {
            let mut fs = MemFilesystem::new();
            fs.add_dir("/ws/proj").unwrap();
            fs.add_file("/ws/proj/file.txt", b"hello".to_vec()).unwrap();

            let task = Task {
                name: task_name("build"),
                action: TaskAction::Command(NonEmpty::from_vec(vec!["true".to_owned()]).unwrap()),
                inputs: vec![InputSpec::parse("file.txt").unwrap()],
                outputs: vec![],
                deps: vec![],
                weak_deps: vec![],
                mutex: None,
                env: EnvSettings {
                    from_host: BTreeSet::from([env_var("PATH")]),
                    overrides: BTreeMap::new(),
                },
            };

            let project = Project {
                name: project_name("proj"),
                root: ProjectRoot::Nested(
                    CanonicalPath::from_absolute(&HazPath::parse("/proj").unwrap()).unwrap(),
                ),
                tags: BTreeSet::new(),
                tasks: BTreeMap::from([(task.name.clone(), task)]),
            };

            let task_id = TaskId {
                project: project_name("proj"),
                task: task_name("build"),
            };
            let pred = TaskId {
                project: project_name("lib"),
                task: task_name("compile"),
            };

            let workspace = Workspace {
                root: WorkspaceRootPath::try_new(PathBuf::from("/ws")).unwrap(),
                projects: BTreeMap::from([(project.name.clone(), project)]),
                overlays: BTreeMap::new(),
                settings: WorkspaceSettings::default(),
            };

            let graph = TaskGraph {
                nodes: BTreeSet::from([task_id.clone(), pred.clone()]),
                edges: BTreeSet::from([Edge {
                    from: pred.clone(),
                    to: task_id.clone(),
                    kind: EdgeKind::Hard,
                }]),
            };

            let predecessor_streams = BTreeMap::from([(
                pred,
                PredecessorStreamHashes {
                    stdout_hash: [0xAA; 32],
                    stderr_hash: [0xBB; 32],
                },
            )]);

            let host_env = BTreeMap::from([(env_var("PATH"), "/usr/bin".to_owned())]);

            E2eState {
                fs,
                workspace,
                graph,
                task_id,
                host_env,
                predecessor_streams,
            }
        }

        fn mutate_task(state: &mut E2eState, f: impl FnOnce(&mut Task)) {
            let proj = state
                .workspace
                .projects
                .get_mut(&state.task_id.project)
                .expect("baseline contains the target project");
            let t = proj
                .tasks
                .get_mut(&state.task_id.task)
                .expect("baseline contains the target task");
            f(t);
        }

        #[test]
        fn identical_inputs_yield_identical_keys() {
            // CACHE-001 determinism: two fresh builders fed
            // byte-equivalent state produce the same key.
            assert_eq!(baseline().key(), baseline().key());
        }

        #[test]
        fn task_action_change_changes_key() {
            let baseline_key = baseline().key();
            let mut perturbed = baseline();
            mutate_task(&mut perturbed, |t| {
                t.action =
                    TaskAction::Command(NonEmpty::from_vec(vec!["false".to_owned()]).unwrap());
            });
            assert_ne!(baseline_key, perturbed.key());
        }

        #[test]
        fn dag_018_input_file_content_change_changes_key() {
            // DAG-018: file contents at any path matched by `A`'s
            // `inputs` MUST contribute to `A`'s cache key. The
            // operational mirror is CACHE-006.
            let baseline_key = baseline().key();
            let perturbed = baseline();
            perturbed
                .fs
                .write_file(Path::new("/ws/proj/file.txt"), b"world")
                .unwrap();
            assert_ne!(baseline_key, perturbed.key());
        }

        #[test]
        fn dag_018_adding_a_soft_edge_predecessor_does_not_change_the_key() {
            // DAG-018 second clause: soft edges (DAG-009) and the
            // producer relation (DAG-013) contribute NOTHING
            // directly to `A`'s cache key. The producer's identity
            // is not in the key; the producer's effect reaches `A`
            // only via file contents at `A`'s declared input paths
            // (which the first DAG-018 test pins).
            let baseline_key = baseline().key();
            let mut perturbed = baseline();
            let soft_pred = TaskId {
                project: project_name("soft_producer"),
                task: task_name("emit"),
            };
            perturbed.graph.nodes.insert(soft_pred.clone());
            perturbed.graph.edges.insert(Edge {
                from: soft_pred,
                to: perturbed.task_id.clone(),
                kind: EdgeKind::Soft,
            });
            assert_eq!(
                baseline_key,
                perturbed.key(),
                "DAG-018: soft-edge predecessors MUST NOT contribute to the key",
            );
        }

        #[test]
        fn dag_018_adding_a_producer_matching_edge_does_not_change_the_key() {
            // DAG-018 second clause (producer relation arm): a
            // producer-matching edge alone (not paired with a hard
            // edge) MUST NOT contribute to the consumer's key.
            let baseline_key = baseline().key();
            let mut perturbed = baseline();
            let pm_pred = TaskId {
                project: project_name("pm_producer"),
                task: task_name("emit"),
            };
            perturbed.graph.nodes.insert(pm_pred.clone());
            perturbed.graph.edges.insert(Edge {
                from: pm_pred,
                to: perturbed.task_id.clone(),
                kind: EdgeKind::ProducerMatching,
            });
            assert_eq!(
                baseline_key,
                perturbed.key(),
                "DAG-018: producer-matching edges MUST NOT contribute to the key",
            );
        }

        #[test]
        fn dag_017_predecessor_stream_hash_change_changes_key() {
            let baseline_key = baseline().key();
            let mut perturbed = baseline();
            let pred = TaskId {
                project: project_name("lib"),
                task: task_name("compile"),
            };
            perturbed.predecessor_streams.insert(
                pred,
                PredecessorStreamHashes {
                    stdout_hash: [0xCC; 32],
                    stderr_hash: [0xBB; 32],
                },
            );
            assert_ne!(baseline_key, perturbed.key());
        }

        #[test]
        fn env_value_change_changes_key() {
            let baseline_key = baseline().key();
            let mut perturbed = baseline();
            perturbed
                .host_env
                .insert(env_var("PATH"), "/usr/local/bin".to_owned());
            assert_ne!(baseline_key, perturbed.key());
        }

        #[test]
        fn unrelated_host_env_does_not_affect_key() {
            // A host env entry whose name is NOT in the task's
            // from_host allow-list MUST NOT enter the key (CACHE-008).
            let baseline_key = baseline().key();
            let mut perturbed = baseline();
            perturbed
                .host_env
                .insert(env_var("HOME"), "/home/user".to_owned());
            assert_eq!(baseline_key, perturbed.key());
        }

        #[test]
        fn override_change_changes_key() {
            // CACHE-008 overrides: changing the task's override
            // value for a name changes the key.
            let baseline_key = baseline().key();
            let mut perturbed = baseline();
            mutate_task(&mut perturbed, |t| {
                t.env
                    .overrides
                    .insert(env_var("HAZ_ENV"), "alpha".to_owned());
            });
            let with_alpha = perturbed.key();
            assert_ne!(baseline_key, with_alpha);

            mutate_task(&mut perturbed, |t| {
                t.env
                    .overrides
                    .insert(env_var("HAZ_ENV"), "beta".to_owned());
            });
            assert_ne!(with_alpha, perturbed.key());
        }

        #[test]
        fn mutex_008_adding_a_mutex_does_not_change_the_key() {
            // MUTEX-008: a task's (scope, name, mode) declaration
            // MUST NOT contribute to the cache key. Two tasks
            // identical in command, inputs, hard-edge predecessors,
            // and environment but differing only by `mutex` produce
            // identical keys.
            let baseline_key = baseline().key();
            let mut perturbed = baseline();
            mutate_task(&mut perturbed, |t| {
                t.mutex = Some(Mutex {
                    scope: MutexScope::Workspace,
                    name: MutexName::try_new("db").unwrap(),
                    mode: MutexMode::Exclusive,
                });
            });
            assert_eq!(baseline_key, perturbed.key());
        }

        #[test]
        fn mutex_008_changing_mutex_scope_does_not_change_the_key() {
            // MUTEX-008: the scope axis of the (scope, name, mode)
            // tuple is also out of the key.
            let mut workspace_scoped = baseline();
            mutate_task(&mut workspace_scoped, |t| {
                t.mutex = Some(Mutex {
                    scope: MutexScope::Workspace,
                    name: MutexName::try_new("db").unwrap(),
                    mode: MutexMode::Exclusive,
                });
            });
            let mut project_scoped = baseline();
            mutate_task(&mut project_scoped, |t| {
                t.mutex = Some(Mutex {
                    scope: MutexScope::Project,
                    name: MutexName::try_new("db").unwrap(),
                    mode: MutexMode::Exclusive,
                });
            });
            assert_eq!(workspace_scoped.key(), project_scoped.key());
        }

        #[test]
        fn mutex_008_changing_mutex_name_does_not_change_the_key() {
            // MUTEX-008: the name axis is also out of the key.
            let mut named_db = baseline();
            mutate_task(&mut named_db, |t| {
                t.mutex = Some(Mutex {
                    scope: MutexScope::Workspace,
                    name: MutexName::try_new("db").unwrap(),
                    mode: MutexMode::Exclusive,
                });
            });
            let mut named_gpu = baseline();
            mutate_task(&mut named_gpu, |t| {
                t.mutex = Some(Mutex {
                    scope: MutexScope::Workspace,
                    name: MutexName::try_new("gpu").unwrap(),
                    mode: MutexMode::Exclusive,
                });
            });
            assert_eq!(named_db.key(), named_gpu.key());
        }

        #[test]
        fn mutex_008_changing_mutex_mode_does_not_change_the_key() {
            // MUTEX-008: the mode axis is also out of the key.
            let mut exclusive = baseline();
            mutate_task(&mut exclusive, |t| {
                t.mutex = Some(Mutex {
                    scope: MutexScope::Workspace,
                    name: MutexName::try_new("db").unwrap(),
                    mode: MutexMode::Exclusive,
                });
            });
            let mut shared = baseline();
            mutate_task(&mut shared, |t| {
                t.mutex = Some(Mutex {
                    scope: MutexScope::Workspace,
                    name: MutexName::try_new("db").unwrap(),
                    mode: MutexMode::Shared,
                });
            });
            assert_eq!(exclusive.key(), shared.key());
        }

        #[test]
        fn task_not_in_workspace_surfaces_error() {
            let mut state = baseline();
            state.task_id = TaskId {
                project: project_name("absent"),
                task: task_name("build"),
            };
            match build_cache_key(
                &state.fs,
                &state.workspace,
                &state.graph,
                &state.task_id,
                &state.host_env,
                &state.predecessor_streams,
                HashAlgo::Blake3,
            ) {
                Err(BuildKeyError::TaskNotInWorkspace { task }) => {
                    assert_eq!(task, state.task_id);
                }
                other => panic!("expected TaskNotInWorkspace, got {other:?}"),
            }
        }
    }
}