mise 2026.8.16

use crate::config::Settings;
use crate::config::env_directive::EnvDirective;
use crate::task::task_fetcher::TaskFetcher;
use crate::task::{Task, TaskRunPhase, dep_has_usage_ref, parse_usage_values_from_task};
use crate::{config::Config, task::task_list::resolve_depends};
use itertools::Itertools;
use petgraph::Direction;
use petgraph::algo::kosaraju_scc;
use petgraph::graph::{DiGraph, NodeIndex};
use std::{
    collections::{HashMap, HashSet, VecDeque},
    fmt,
    sync::Arc,
};
use tokio::sync::mpsc;

/// Unique key for a task occurrence, including name, args, env vars, and phase.
pub(crate) type TaskKey = (String, Vec<String>, Vec<(String, String)>, TaskRunPhase);

fn env_key(task: &Task) -> Vec<(String, String)> {
    task.env
        .0
        .iter()
        .filter_map(|d| match d {
            EnvDirective::Val(k, v, _) => Some((k.clone(), v.clone())),
            _ => None,
        })
        .sorted()
        .collect()
}

pub(crate) struct TaskCycleError {
    paths: Vec<Vec<String>>,
    keys: Vec<Vec<TaskKey>>,
}

impl TaskCycleError {
    pub(crate) fn path(&self) -> &[String] {
        self.paths.first().map(Vec::as_slice).unwrap_or_default()
    }

    pub(crate) fn paths(&self) -> &[Vec<String>] {
        &self.paths
    }

    pub(crate) fn keys(&self) -> &[Vec<TaskKey>] {
        &self.keys
    }
}

impl fmt::Debug for TaskCycleError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("TaskCycleError")
            .field("paths", &self.paths)
            .finish_non_exhaustive()
    }
}

impl fmt::Display for TaskCycleError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "circular dependency detected: {}",
            self.path().iter().join(" -> ")
        )
    }
}

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

#[derive(Debug, Clone, Default, PartialEq, Eq)]
/// State contributed by a task's completed direct dependencies.
pub(crate) struct TaskDependencyState {
    /// Stable artifact identities to include in the task's cache key.
    pub cache_keys: Vec<String>,
    /// Whether any dependency executed or restored outputs.
    pub any_did_work: bool,
    /// Whether any dependency did work without publishing a stable artifact identity.
    pub any_unkeyed_did_work: bool,
}

#[derive(Debug, Clone, Default)]
/// Completed task state that can be propagated into a nested task graph.
pub(crate) struct TaskCompletionState {
    completed: HashSet<TaskKey>,
    did_work: HashSet<TaskKey>,
    cache_keys: HashMap<TaskKey, String>,
}

impl TaskCompletionState {
    /// Merge state returned by a completed nested task graph.
    pub(crate) fn merge(&mut self, other: Self) {
        self.completed.extend(other.completed);
        self.did_work.extend(other.did_work);
        self.cache_keys.extend(other.cache_keys);
    }
}

#[derive(Debug)]
pub(crate) struct Deps {
    pub graph: DiGraph<Task, ()>,
    sent: HashSet<TaskKey>, // tasks that have already started so should not run again
    removed: HashSet<TaskKey>, // tasks that have already finished to track if we are in an infinite loop
    executed: HashSet<TaskKey>, // tasks that actually began executing (not just scheduled)
    did_work: HashSet<TaskKey>, // tasks that executed or restored outputs (not freshness-skipped)
    cache_keys: HashMap<TaskKey, String>, // stable artifact identities published by completed tasks
    dep_edges: HashMap<TaskKey, HashSet<TaskKey>>, // maps each task to its direct dependency task keys
    post_dep_parents: HashMap<TaskKey, HashSet<TaskKey>>, // maps each post-subtree task to its triggering parents
    creation_order: Vec<TaskKey>, // every node in the order the graph created it, see `all_in_creation_order`
    tx: mpsc::UnboundedSender<Option<Task>>,
    // not clone, notify waiters via tx None
}

/// Extract a hashable key from a task, including env vars set via dependencies
pub(super) fn task_key(task: &Task) -> TaskKey {
    (
        task.name.clone(),
        task.args.clone(),
        env_key(task),
        task.run_phase,
    )
}

fn same_task_without_phase(task: &Task, other: &Task) -> bool {
    task.name == other.name && task.args == other.args && env_key(task) == env_key(other)
}

/// manages a dependency graph of tasks so `mise run` knows what to run next
impl Deps {
    pub(crate) async fn new(config: &Arc<Config>, tasks: Vec<Task>) -> eyre::Result<Self> {
        Self::new_with_cycle_limit(config, tasks, Some(1)).await
    }

    pub(crate) async fn new_for_validation(
        config: &Arc<Config>,
        tasks: Vec<Task>,
    ) -> eyre::Result<Self> {
        Self::new_with_cycle_limit(config, tasks, None).await
    }

    async fn new_with_cycle_limit(
        config: &Arc<Config>,
        tasks: Vec<Task>,
        cycle_limit: Option<usize>,
    ) -> eyre::Result<Self> {
        let mut graph = DiGraph::new();
        let mut indexes = HashMap::new();
        let mut stack = vec![];
        let mut seen = HashSet::new();
        let mut direct_post_parents: HashMap<TaskKey, HashSet<TaskKey>> = HashMap::new();
        let mut dep_edges: HashMap<TaskKey, HashSet<TaskKey>> = HashMap::new();
        let mut wait_edges = Vec::new();

        let mut add_idx = |task: &Task, graph: &mut DiGraph<Task, ()>| {
            *indexes
                .entry(task_key(task))
                .or_insert_with(|| graph.add_node(task.clone()))
        };

        // first we add all tasks to the graph, create a stack of work for this function, and
        // store the index of each task in the graph
        for t in &tasks {
            let t = t.clone().with_run_phase(TaskRunPhase::Normal);
            stack.push(t.clone());
            add_idx(&t, &mut graph);
        }
        let all_tasks_to_run = resolve_depends(config, tasks).await?;
        let no_cache = Settings::get().task.remote_no_cache.unwrap_or(false);
        let fetcher = TaskFetcher::new(no_cache);
        while let Some(mut a) = stack.pop() {
            if seen.contains(&a) {
                // prevent infinite loop
                continue;
            }
            // Fetch remote task files so file-based tasks have local paths
            // before we try to parse their usage specs or execute them.
            if a.file
                .as_ref()
                .is_some_and(|f| TaskFetcher::is_remote_source(&f.to_string_lossy()))
            {
                let mut tasks_to_fetch = vec![a];
                fetcher.fetch_tasks(config, &mut tasks_to_fetch).await?;
                a = tasks_to_fetch.into_iter().next().unwrap();
            }
            // Re-render dependency templates with usage values (including defaults)
            // so {{usage.*}} resolves.
            let has_usage_deps = |raw: &Option<Vec<_>>| {
                raw.as_ref()
                    .is_some_and(|r| r.iter().any(dep_has_usage_ref))
            };
            if has_usage_deps(&a.depends_raw)
                || has_usage_deps(&a.depends_post_raw)
                || has_usage_deps(&a.wait_for_raw)
            {
                let usage_values = parse_usage_values_from_task(config, &a).await?;
                if !usage_values.is_empty() {
                    a.render_depends_with_usage(config, &usage_values).await?;
                }
            }
            let a_idx = add_idx(&a, &mut graph);
            // Update the graph node with the fetched version of the task
            // (add_idx may have returned an existing index with an unfetched task)
            graph[a_idx] = a.clone();
            let resolved = a.resolve_depends(config, &all_tasks_to_run).await?;
            for b in resolved.depends {
                let b = b.with_run_phase(a.run_phase);
                let b_idx = add_idx(&b, &mut graph);
                graph.update_edge(a_idx, b_idx, ());
                dep_edges
                    .entry(task_key(&a))
                    .or_default()
                    .insert(task_key(&b));
                stack.push(b.clone());
            }
            for b in resolved.wait_for {
                wait_edges.push((task_key(&a), b));
            }
            for b in resolved.depends_post {
                let b = b.with_run_phase(TaskRunPhase::Post);
                add_idx(&b, &mut graph);
                direct_post_parents
                    .entry(task_key(&b))
                    .or_default()
                    .insert(task_key(&a));
                stack.push(b.clone());
            }
            seen.insert(a);
        }

        // A post task's regular dependencies are part of the same cleanup
        // subtree. Propagate the triggering parents through those edges so
        // every prerequisite waits for the parent before it can start.
        let mut post_dep_parents: HashMap<TaskKey, HashSet<TaskKey>> = HashMap::new();
        let mut pending = VecDeque::new();
        for (post_key, parents) in direct_post_parents {
            for parent in parents {
                if post_dep_parents
                    .entry(post_key.clone())
                    .or_default()
                    .insert(parent.clone())
                {
                    pending.push_back((post_key.clone(), parent));
                }
            }
        }
        while let Some((post_key, parent)) = pending.pop_front() {
            for dependency in dep_edges.get(&post_key).into_iter().flatten() {
                if dependency.3 != TaskRunPhase::Post || dependency == &parent {
                    continue;
                }
                if post_dep_parents
                    .entry(dependency.clone())
                    .or_default()
                    .insert(parent.clone())
                {
                    pending.push_back((dependency.clone(), parent.clone()));
                }
            }
        }

        // Add the parent-completion barrier to the entire post subtree.
        for (post_key, parents) in &post_dep_parents {
            let post_idx = indexes[post_key];
            for parent in parents {
                graph.update_edge(post_idx, indexes[parent], ());
            }
        }

        // wait_for only links occurrences already introduced by a root or a
        // real dependency. Prefer the current phase, but never create another
        // occurrence solely to satisfy a wait_for declaration.
        for (waiting_key, target) in wait_edges {
            let waiting_idx = indexes[&waiting_key];
            let target_idx = graph
                .node_indices()
                .filter(|&idx| same_task_without_phase(&graph[idx], &target))
                .find(|&idx| graph[idx].run_phase == waiting_key.3)
                .or_else(|| {
                    graph
                        .node_indices()
                        .find(|&idx| same_task_without_phase(&graph[idx], &target))
                });
            if let Some(target_idx) = target_idx {
                graph.update_edge(waiting_idx, target_idx, ());
            }
        }
        let cycles = find_cycles(&graph, cycle_limit);
        if !cycles.is_empty() {
            let paths = cycles
                .iter()
                .map(|cycle| {
                    cycle
                        .iter()
                        .map(|&idx| task_cycle_label(&graph[idx]))
                        .collect()
                })
                .collect();
            let keys = cycles
                .iter()
                .map(|cycle| cycle.iter().map(|&idx| task_key(&graph[idx])).collect())
                .collect();
            return Err(eyre::Report::new(TaskCycleError { paths, keys }));
        }
        let (tx, _) = mpsc::unbounded_channel();
        let sent = HashSet::new();
        let removed = HashSet::new();
        let executed = HashSet::new();
        let did_work = HashSet::new();
        let cache_keys = HashMap::new();
        // Node indices are handed out by `add_idx` alone -- roots in the order
        // they were named, then dependencies as the worklist reaches them -- and
        // nothing has removed a node yet, so index order is creation order here
        // and nowhere later.
        let creation_order = graph
            .node_indices()
            .map(|idx| task_key(&graph[idx]))
            .collect();
        Ok(Self {
            graph,
            tx,
            sent,
            removed,
            executed,
            did_work,
            cache_keys,
            dep_edges,
            post_dep_parents,
            creation_order,
        })
    }

    /// Create a sub-graph that prunes tasks already completed by the caller.
    /// `completed` is a snapshot of task keys that have finished in the parent
    /// graph — these are removed from the sub-graph so they don't run again.
    pub(crate) async fn new_pruned(
        config: &Arc<Config>,
        tasks: Vec<Task>,
        completed: &TaskCompletionState,
    ) -> eyre::Result<Self> {
        let mut deps = Self::new(config, tasks).await?;
        deps.did_work.extend(completed.did_work.iter().cloned());
        deps.cache_keys.extend(completed.cache_keys.clone());
        let mut to_remove = vec![];
        for idx in deps.graph.node_indices() {
            let key = task_key(&deps.graph[idx]);
            if completed.completed.contains(&key) {
                to_remove.push(idx);
            }
        }
        // Remove in reverse index order so petgraph swap-remove
        // doesn't invalidate indices we haven't processed yet
        to_remove.sort_unstable_by(|a, b| b.cmp(a));
        for idx in to_remove {
            deps.graph.remove_node(idx);
        }
        deps.mark_ambiguous_prefixes();
        Ok(deps)
    }

    /// main method to emit tasks that no longer have dependencies being waited on
    fn emit_leaves(&mut self) {
        let leaves = leaves(&self.graph);
        let leaves_is_empty = leaves.is_empty();

        for task in leaves {
            let key = task_key(&task);

            if self.sent.insert(key.clone()) {
                trace!("Scheduling task {0}", task.name);
                if let Err(e) = self.tx.send(Some(task)) {
                    trace!("Error sending task: {e:?}");
                    self.sent.remove(&key);
                }
            }
        }

        if self.is_empty() {
            trace!("All tasks finished");
            if let Err(e) = self.tx.send(None) {
                trace!("Error closing task stream: {e:?}");
            }
        } else if leaves_is_empty && self.sent.len() == self.removed.len() {
            panic!(
                "Infinitive loop detected, all tasks are finished but the graph isn't empty {0} {1:#?}",
                self.all().map(|t| t.name.clone()).join(", "),
                self.graph
            )
        }
    }

    /// listened to by `mise run` which gets a stream of tasks to run
    pub(crate) fn subscribe(&mut self) -> mpsc::UnboundedReceiver<Option<Task>> {
        let (tx, rx) = mpsc::unbounded_channel();
        self.tx = tx;
        self.emit_leaves();
        rx
    }

    pub(crate) fn is_empty(&self) -> bool {
        self.graph.node_count() == 0
    }

    /// Snapshot completed task state for nested task sub-graphs.
    pub(crate) fn completion_state(&self) -> TaskCompletionState {
        TaskCompletionState {
            completed: self.removed.clone(),
            did_work: self.did_work.clone(),
            cache_keys: self.cache_keys.clone(),
        }
    }

    /// Check if a post-dep task should actually run: it must be a post-dependency
    /// AND its parent must have actually started executing (not just been scheduled).
    /// Returns false for non-post-dep tasks or post-deps whose parent was never executed.
    pub(crate) fn is_runnable_post_dep(&self, task: &Task) -> bool {
        let key = task_key(task);
        match self.post_dep_parents.get(&key) {
            Some(parent_keys) => parent_keys.iter().any(|pk| self.executed.contains(pk)),
            None => false,
        }
    }

    /// Whether this task ever started executing. `mark_executed` runs
    /// synchronously before the task is spawned, so a task missing here has no
    /// execution in flight and will never reach the completion callbacks.
    pub(crate) fn has_executed(&self, task: &Task) -> bool {
        self.executed.contains(&task_key(task))
    }

    /// Mark a task as having actually started execution.
    /// This is distinct from being scheduled (sent) — a task may be scheduled as a
    /// graph leaf but then skipped because an earlier task failed.
    pub(crate) fn mark_executed(&mut self, task: &Task) {
        self.executed.insert(task_key(task));
    }

    /// Clear the execution marker when cancellation prevents a scheduled task
    /// from reaching process startup.
    pub(crate) fn unmark_executed(&mut self, task: &Task) {
        self.executed.remove(&task_key(task));
    }

    /// Mark a task as having executed or restored outputs.
    /// Used to invalidate dependent tasks' source freshness checks.
    pub(crate) fn mark_did_work(&mut self, task: &Task) {
        self.did_work.insert(task_key(task));
    }

    /// Record a stable artifact identity produced or reused by a completed task.
    pub(crate) fn mark_cache_key(&mut self, task: &Task, cache_key: String) {
        self.cache_keys.insert(task_key(task), cache_key);
    }

    /// Return the completed dependency state needed for freshness and artifact caching.
    pub(crate) fn dependency_state(&self, task: &Task) -> TaskDependencyState {
        let key = task_key(task);
        let deps = self
            .dep_edges
            .get(&key)
            .into_iter()
            .flatten()
            .chain(self.post_dep_parents.get(&key).into_iter().flatten())
            .collect::<HashSet<_>>();
        let mut cache_keys = deps
            .iter()
            .filter_map(|dep_key| self.cache_keys.get(dep_key).cloned())
            .collect::<Vec<_>>();
        cache_keys.sort();
        cache_keys.dedup();
        TaskDependencyState {
            cache_keys,
            any_did_work: deps.iter().any(|dep_key| self.did_work.contains(dep_key)),
            any_unkeyed_did_work: deps.iter().any(|dep_key| {
                self.did_work.contains(dep_key) && !self.cache_keys.contains_key(dep_key)
            }),
        }
    }

    /// Remove multiple tasks from the graph in a batch, emitting leaves only once at the end.
    /// This prevents intermediate emit_leaves from scheduling tasks that will be removed later.
    pub(crate) fn remove_batch(&mut self, tasks: &[Task]) {
        for task in tasks {
            if let Some(idx) = self.node_idx(task) {
                self.graph.remove_node(idx);
                let key = task_key(task);
                self.removed.insert(key);
            }
        }
        self.emit_leaves();
    }

    // use contracts::{ensures, requires};
    // #[requires(self.graph.node_count() > 0)]
    // #[ensures(self.graph.node_count() == old(self.graph.node_count()) - 1)]
    pub(crate) fn remove(&mut self, task: &Task) {
        if let Some(idx) = self.node_idx(task) {
            self.graph.remove_node(idx);
            let key = task_key(task);
            self.removed.insert(key);
            self.emit_leaves();
        }
    }

    fn node_idx(&self, task: &Task) -> Option<petgraph::graph::NodeIndex> {
        self.graph
            .node_indices()
            .find(|&idx| &self.graph[idx] == task)
    }

    /// Every task still in the graph, in no particular order.
    ///
    /// Node index order is creation order only until the first removal, so
    /// anything that cares about the order wants
    /// [`all_in_creation_order`](Self::all_in_creation_order) instead.
    pub(crate) fn all(&self) -> impl Iterator<Item = &Task> {
        self.graph.node_indices().map(|idx| &self.graph[idx])
    }

    /// Every task still in the graph, in the order the graph created its nodes:
    /// the roots in the order they were named, then each task's `depends` and
    /// `depends_post` in the order the worklist reached them.
    ///
    /// This is the order keep-order hands out its output slots in, so it has to
    /// survive removals — and [`all`](Self::all) does not. petgraph's
    /// `remove_node` swap-removes, moving the last node into the hole, so a
    /// single pruned or finished task is enough to scramble index order.
    ///
    /// Tasks the graph no longer holds are dropped rather than reported: a task
    /// pruned as already complete never runs, so nothing would ever retire the
    /// slot it was given, and an empty slot at the front of the buffer map stops
    /// everything behind it from streaming.
    pub(crate) fn all_in_creation_order(&self) -> Vec<&Task> {
        let mut present: HashMap<TaskKey, &Task> = self
            .graph
            .node_indices()
            .map(|idx| (task_key(&self.graph[idx]), &self.graph[idx]))
            .collect();
        // `add_idx` deduplicates by key, so key to node is one-to-one and
        // removing as we go is just that stated out loud.
        self.creation_order
            .iter()
            .filter_map(|key| present.remove(key))
            .collect()
    }

    /// Mark tasks that share a display_name so their prefix includes args
    /// for disambiguation (e.g. `[test-docker 4.1]` vs `[test-docker 4.2]`).
    pub(crate) fn mark_ambiguous_prefixes(&mut self) {
        let mut name_to_indices: HashMap<String, Vec<petgraph::graph::NodeIndex>> = HashMap::new();
        for idx in self.graph.node_indices() {
            name_to_indices
                .entry(self.graph[idx].display_name.clone())
                .or_default()
                .push(idx);
        }
        for indices in name_to_indices.values() {
            if indices.len() > 1 {
                for &idx in indices {
                    self.graph[idx].show_args_in_prefix = true;
                }
            }
        }
    }

    pub(crate) fn is_linear(&self) -> bool {
        let mut graph = self.graph.clone();
        // pop dependencies off, if we get multiple dependencies at once it's not linear
        loop {
            let leaves = leaves(&graph);
            if leaves.is_empty() {
                return true;
            } else if leaves.len() > 1 {
                return false;
            } else {
                let idx = self
                    .graph
                    .node_indices()
                    .find(|&idx| graph[idx] == leaves[0])
                    .unwrap();
                graph.remove_node(idx);
            }
        }
    }
}

fn leaves(graph: &DiGraph<Task, ()>) -> Vec<Task> {
    graph
        .externals(Direction::Outgoing)
        .map(|idx| graph[idx].clone())
        .collect()
}

pub(crate) fn task_cycle_label(task: &Task) -> String {
    let mut label = if task.args.is_empty() {
        task.name.clone()
    } else {
        format!("{} {}", task.name, task.args.join(" "))
    };
    if task.run_phase == TaskRunPhase::Post {
        label.push_str(" (post)");
    }
    let env_keys = task
        .env
        .0
        .iter()
        .filter_map(|directive| match directive {
            EnvDirective::Val(key, _, _) => Some(key),
            _ => None,
        })
        .sorted()
        .unique()
        .join(", ");
    if env_keys.is_empty() {
        label
    } else {
        format!("{label} [env: {env_keys}]")
    }
}

fn find_cycles(graph: &DiGraph<Task, ()>, limit: Option<usize>) -> Vec<Vec<NodeIndex>> {
    let mut cycles = Vec::new();
    for mut component in kosaraju_scc(graph) {
        component.sort_by_key(|node| node.index());
        let component: HashSet<_> = component.into_iter().collect();
        if component.len() == 1 {
            let node = *component.iter().next().unwrap();
            if graph.find_edge(node, node).is_some() {
                cycles.push(vec![node, node]);
            }
            if limit.is_some_and(|limit| cycles.len() >= limit) {
                return cycles;
            }
            continue;
        }

        let mut starts = component.iter().copied().collect_vec();
        starts.sort_by_key(|node| node.index());
        for start in starts {
            let mut path = vec![start];
            let mut in_path = HashSet::from([start]);
            let mut stack = vec![(
                start,
                graph
                    .neighbors_directed(start, Direction::Outgoing)
                    .filter(|node| component.contains(node))
                    .sorted_by_key(|node| node.index())
                    .collect_vec(),
                0,
            )];

            while !stack.is_empty() {
                let dependency = {
                    let (_, dependencies, next) = stack.last_mut().unwrap();
                    if *next < dependencies.len() {
                        let dependency = dependencies[*next];
                        *next += 1;
                        Some(dependency)
                    } else {
                        None
                    }
                };

                let Some(dependency) = dependency else {
                    let (node, _, _) = stack.pop().unwrap();
                    path.pop();
                    in_path.remove(&node);
                    continue;
                };
                if dependency == start {
                    let mut cycle = path.clone();
                    cycle.push(start);
                    cycles.push(cycle);
                    if limit.is_some_and(|limit| cycles.len() >= limit) {
                        return cycles;
                    }
                } else if dependency.index() >= start.index() && in_path.insert(dependency) {
                    path.push(dependency);
                    stack.push((
                        dependency,
                        graph
                            .neighbors_directed(dependency, Direction::Outgoing)
                            .filter(|node| component.contains(node))
                            .sorted_by_key(|node| node.index())
                            .collect_vec(),
                        0,
                    ));
                }
            }
        }
    }
    cycles
}

#[cfg(test)]
mod tests {
    use super::*;

    fn task(name: &str) -> Task {
        Task {
            name: name.to_string(),
            ..Default::default()
        }
    }

    fn post_task(name: &str) -> Task {
        task(name).with_run_phase(TaskRunPhase::Post)
    }

    fn deps_with_relationships(
        dep_edges: HashMap<TaskKey, HashSet<TaskKey>>,
        post_dep_parents: HashMap<TaskKey, HashSet<TaskKey>>,
    ) -> Deps {
        let (tx, _) = mpsc::unbounded_channel();
        Deps {
            graph: DiGraph::new(),
            sent: HashSet::new(),
            removed: HashSet::new(),
            executed: HashSet::new(),
            did_work: HashSet::new(),
            cache_keys: HashMap::new(),
            dep_edges,
            post_dep_parents,
            creation_order: Vec::new(),
            tx,
        }
    }

    #[test]
    fn unmark_executed_disables_post_dependency_cleanup() {
        let parent = task("parent");
        let cleanup = task("cleanup");
        let mut deps = deps_with_relationships(
            HashMap::new(),
            HashMap::from([(task_key(&cleanup), HashSet::from([task_key(&parent)]))]),
        );

        deps.mark_executed(&parent);
        assert!(deps.is_runnable_post_dep(&cleanup));

        deps.unmark_executed(&parent);
        assert!(!deps.is_runnable_post_dep(&cleanup));
    }

    #[test]
    fn dependency_state_tracks_direct_artifact_identity_and_unkeyed_work() {
        let a = task("a");
        let b = task("b");
        let c = task("c");
        let dep_edges = HashMap::from([
            (task_key(&b), HashSet::from([task_key(&a)])),
            (task_key(&c), HashSet::from([task_key(&b)])),
        ]);
        let mut deps = deps_with_relationships(dep_edges, HashMap::new());

        deps.mark_did_work(&b);
        assert_eq!(
            deps.dependency_state(&c),
            TaskDependencyState {
                cache_keys: vec![],
                any_did_work: true,
                any_unkeyed_did_work: true,
            }
        );

        deps.mark_cache_key(&b, "b-key".to_string());
        assert_eq!(
            deps.dependency_state(&c),
            TaskDependencyState {
                cache_keys: vec!["b-key".to_string()],
                any_did_work: true,
                any_unkeyed_did_work: false,
            }
        );
    }

    #[test]
    fn dependency_state_includes_post_dependency_parents() {
        let parent = task("parent");
        let post = task("post");
        let post_dep_parents =
            HashMap::from([(task_key(&post), HashSet::from([task_key(&parent)]))]);
        let mut deps = deps_with_relationships(HashMap::new(), post_dep_parents);

        deps.mark_did_work(&parent);
        deps.mark_cache_key(&parent, "parent-key".to_string());

        assert_eq!(
            deps.dependency_state(&post),
            TaskDependencyState {
                cache_keys: vec!["parent-key".to_string()],
                any_did_work: true,
                any_unkeyed_did_work: false,
            }
        );
    }

    #[tokio::test]
    async fn new_pruned_preserves_completed_artifact_state() {
        let completed_task = task("completed");
        let key = task_key(&completed_task);
        let completion_state = TaskCompletionState {
            completed: HashSet::from([key.clone()]),
            did_work: HashSet::from([key.clone()]),
            cache_keys: HashMap::from([(key.clone(), "completed-key".to_string())]),
        };
        let config = Config::get().await.unwrap();

        let deps = Deps::new_pruned(&config, vec![completed_task], &completion_state)
            .await
            .unwrap();
        let propagated = deps.completion_state();

        assert!(deps.is_empty());
        assert!(propagated.did_work.contains(&key));
        assert_eq!(
            propagated.cache_keys.get(&key).map(String::as_str),
            Some("completed-key")
        );
    }

    #[tokio::test]
    async fn wait_for_falls_back_to_post_occurrence_without_tracking_dependency_state() {
        let config = Config::get().await.unwrap();
        let tasks = config.tasks().await.unwrap();
        let mut parent = tasks["configtask"].clone();
        parent.depends_post = vec!["lint".parse().unwrap()];
        let mut waiter = tasks["test"].clone();
        waiter.wait_for = vec!["lint".parse().unwrap()];
        let mut deps = Deps::new(&config, vec![parent, waiter.clone()])
            .await
            .unwrap();

        let waiter_idx = deps.node_idx(&waiter).unwrap();
        let post_target_idx = deps
            .graph
            .node_indices()
            .find(|&idx| {
                deps.graph[idx].name == "lint" && deps.graph[idx].run_phase == TaskRunPhase::Post
            })
            .unwrap();
        assert!(deps.graph.find_edge(waiter_idx, post_target_idx).is_some());
        assert!(!deps.graph.node_indices().any(|idx| {
            deps.graph[idx].name == "lint" && deps.graph[idx].run_phase == TaskRunPhase::Normal
        }));

        let post_target = deps.graph[post_target_idx].clone();
        deps.mark_did_work(&post_target);
        deps.mark_cache_key(&post_target, "post-key".to_string());
        assert_eq!(
            deps.dependency_state(&waiter),
            TaskDependencyState::default()
        );
    }

    /// Roots keep the order they were named; a dependency is discovered by the
    /// worklist and lands after every root. Both halves matter: keep-order hands
    /// out its output slots in this order.
    #[tokio::test]
    async fn creation_order_is_roots_then_discovered_dependencies() {
        let config = Config::get().await.unwrap();
        let tasks = config.tasks().await.unwrap();
        let mut root = tasks["configtask"].clone();
        root.depends = vec!["lint".parse().unwrap()];
        let other = tasks["test"].clone();

        let deps = Deps::new(&config, vec![root, other]).await.unwrap();

        assert_eq!(
            deps.all_in_creation_order()
                .iter()
                .map(|t| t.name.as_str())
                .collect_vec(),
            ["configtask", "test", "lint"],
            "roots as named, then what the worklist reached"
        );
    }

    /// petgraph's `remove_node` swap-removes, so one departure is enough to
    /// scramble node index order. That is the whole reason this accessor exists
    /// rather than callers using `all`.
    #[tokio::test]
    async fn creation_order_survives_a_removal() {
        let config = Config::get().await.unwrap();
        let tasks = config.tasks().await.unwrap();
        let mut root = tasks["configtask"].clone();
        root.depends = vec!["lint".parse().unwrap()];
        let other = tasks["test"].clone();
        let mut deps = Deps::new(&config, vec![root.clone(), other]).await.unwrap();

        deps.remove(&root);

        assert_eq!(
            deps.all_in_creation_order()
                .iter()
                .map(|t| t.name.as_str())
                .collect_vec(),
            ["test", "lint"],
            "the survivors, still in the order they were created"
        );
        assert_eq!(
            deps.all().map(|t| t.name.as_str()).collect_vec(),
            ["lint", "test"],
            "while index order has the last node moved into the hole"
        );
    }

    /// A task pruned as already complete never runs, so nothing would retire the
    /// slot it was given and everything behind it would stay buffered.
    #[tokio::test]
    async fn a_pruned_task_is_absent_from_creation_order() {
        let config = Config::get().await.unwrap();
        let tasks = config.tasks().await.unwrap();
        let mut root = tasks["configtask"].clone();
        root.depends = vec!["lint".parse().unwrap()];
        let other = tasks["test"].clone();
        let completion_state = TaskCompletionState {
            completed: HashSet::from([task_key(&other)]),
            ..Default::default()
        };

        let deps = Deps::new_pruned(&config, vec![root, other], &completion_state)
            .await
            .unwrap();

        assert_eq!(
            deps.all_in_creation_order()
                .iter()
                .map(|t| t.name.as_str())
                .collect_vec(),
            ["configtask", "lint"],
            "the completed task is gone, the rest keep their order"
        );
    }

    #[test]
    fn finds_cycle_path() {
        let mut graph = DiGraph::new();
        let a = graph.add_node(task("a"));
        let b = graph.add_node(task("b"));
        let c = graph.add_node(task("c"));
        graph.update_edge(a, b, ());
        graph.update_edge(b, c, ());
        graph.update_edge(c, a, ());

        let cycle = find_cycles(&graph, Some(1)).pop().unwrap();
        let labels = cycle
            .iter()
            .map(|&idx| task_cycle_label(&graph[idx]))
            .collect_vec();
        assert_eq!(labels, ["a", "b", "c", "a"]);
    }

    #[test]
    fn accepts_acyclic_graph() {
        let mut graph = DiGraph::new();
        let a = graph.add_node(task("a"));
        let b = graph.add_node(task("b"));
        graph.update_edge(b, a, ());

        assert!(find_cycles(&graph, None).is_empty());
    }

    #[test]
    fn normal_and_post_occurrences_have_distinct_identity() {
        let normal = task("shared");
        let post = post_task("shared");
        assert_ne!(normal, post);
        assert_ne!(task_key(&normal), task_key(&post));

        let mut graph = DiGraph::new();
        let normal_idx = graph.add_node(normal);
        let parent_idx = graph.add_node(task("parent"));
        let post_idx = graph.add_node(post);
        graph.update_edge(parent_idx, normal_idx, ());
        graph.update_edge(post_idx, parent_idx, ());
        assert!(find_cycles(&graph, None).is_empty());
    }

    #[test]
    fn post_subtree_waits_for_parent() {
        let mut graph = DiGraph::new();
        let parent = graph.add_node(task("parent"));
        let prerequisite = graph.add_node(post_task("prerequisite"));
        let post = graph.add_node(post_task("post"));
        graph.update_edge(post, prerequisite, ());
        graph.update_edge(prerequisite, parent, ());
        graph.update_edge(post, parent, ());

        assert_eq!(leaves(&graph), [task("parent")]);
        graph.remove_node(parent);
        assert_eq!(leaves(&graph), [post_task("prerequisite")]);
        graph.remove_node(prerequisite);
        assert_eq!(leaves(&graph), [post_task("post")]);
    }

    #[test]
    fn shared_post_occurrence_waits_for_all_parents() {
        let mut graph = DiGraph::new();
        let parent_a = graph.add_node(task("parent-a"));
        let parent_b = graph.add_node(task("parent-b"));
        let post = graph.add_node(post_task("post"));
        graph.update_edge(post, parent_a, ());
        graph.update_edge(post, parent_b, ());

        assert_eq!(leaves(&graph).len(), 2);
        graph.remove_node(parent_a);
        assert_eq!(leaves(&graph), [task("parent-b")]);
        graph.remove_node(parent_b);
        assert_eq!(leaves(&graph), [post_task("post")]);
    }

    #[test]
    fn accepts_deep_acyclic_graph() {
        let mut graph = DiGraph::new();
        let nodes = (0..10_000)
            .map(|i| graph.add_node(task(&format!("task-{i}"))))
            .collect_vec();
        for pair in nodes.windows(2) {
            graph.update_edge(pair[0], pair[1], ());
        }

        assert!(find_cycles(&graph, None).is_empty());
    }

    #[test]
    fn finds_overlapping_cycles() {
        let mut graph = DiGraph::new();
        let root = graph.add_node(task("root"));
        let left = graph.add_node(task("left"));
        let right = graph.add_node(task("right"));
        graph.update_edge(root, left, ());
        graph.update_edge(left, root, ());
        graph.update_edge(root, right, ());
        graph.update_edge(right, root, ());

        let cycles = find_cycles(&graph, None)
            .into_iter()
            .map(|cycle| {
                cycle
                    .iter()
                    .map(|&idx| task_cycle_label(&graph[idx]))
                    .collect_vec()
            })
            .collect_vec();

        assert_eq!(
            cycles,
            [["root", "left", "root"], ["root", "right", "root"]]
        );
    }

    #[test]
    fn cycle_label_disambiguates_environment_variants_without_values() {
        let mut task = task("build");
        task.args = vec!["linux".to_string()];
        task.env.0 = vec![
            EnvDirective::Val(
                "TOKEN".to_string(),
                "secret".to_string(),
                Default::default(),
            ),
            EnvDirective::Val(
                "TARGET".to_string(),
                "linux".to_string(),
                Default::default(),
            ),
        ];

        assert_eq!(task_cycle_label(&task), "build linux [env: TARGET, TOKEN]");
    }
}