haz-exec 0.1.0

//! Runtime `EXEC-019` cycle detection.
//!
//! The static cycle check (`DAG-014`) rejects every length-≥2
//! cycle decidable without filesystem state. Cycles that depend
//! on glob-glob producer-matching edges escape it. This module
//! provides the runtime fallback: after every successful task
//! completion, the scheduler calls
//! [`check_and_record_runtime_cycle_for_completion`] with the
//! completed task's materialised outputs. For each output, the
//! helper enumerates the consumer tasks whose `inputs` patterns
//! match it (a freshly-materialised producer-matching edge), adds
//! the edge to the augmented edge set, and BFS-searches for a
//! cycle through the new edge. On detection the helper appends a
//! [`RuntimeInvariantViolation::RuntimeCycle`] entry to the run's
//! diagnostic vector and returns the cycle's node set so the
//! scheduler can drain pending cycle members from the ready set
//! via [`skip_ready_cycle_members`].

use std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque};

use haz_domain::path::pattern::PathAnchor;
use haz_domain::path::{CanonicalPath, PathPattern, ProjectRoot};
use haz_domain::task_id::TaskId;
use haz_domain::workspace::Workspace;

use crate::pattern_walk::{literal_workspace_segments, workspace_absolute_string_from_segments};
use crate::run_graph::RuntimeInvariantViolation;
use crate::run_graph::state::ReadyState;
use crate::run_task::{RunObserver, RunOutcome, SkipCause, SkipRecord};

/// Return the set of task identities whose declared `inputs`
/// patterns match `path`. Used by `EXEC-019` runtime cycle
/// detection to derive new producer-matching edges from a
/// just-completed task's materialised outputs.
///
/// Linear iteration over the workspace's tasks; per-task,
/// short-circuits on the first matching input. Globs are
/// compiled on each call; for the typical workspace size (dozens
/// of tasks with a handful of patterns each) this is well within
/// budget. A precompile cache may be added if profiling shows
/// it's needed.
pub(super) fn tasks_matching_path(workspace: &Workspace, path: &CanonicalPath) -> BTreeSet<TaskId> {
    let path_workspace_absolute = path.to_string();
    let mut matches = BTreeSet::new();
    for (project_name, project) in &workspace.projects {
        for (task_name, task) in &project.tasks {
            for input in &task.inputs {
                if pattern_matches_workspace_absolute_path(
                    input.pattern(),
                    &project.root,
                    &path_workspace_absolute,
                ) {
                    matches.insert(TaskId {
                        project: project_name.clone(),
                        task: task_name.clone(),
                    });
                    break;
                }
            }
        }
    }
    matches
}

/// Test whether `pattern` (with anchor and project root context)
/// matches the workspace-absolute path `path_workspace_absolute`
/// (a string of the form `/seg1/seg2/...`).
fn pattern_matches_workspace_absolute_path(
    pattern: &PathPattern,
    project_root: &ProjectRoot,
    path_workspace_absolute: &str,
) -> bool {
    match pattern {
        PathPattern::Literal(haz_path) => {
            let pattern_segments = literal_workspace_segments(haz_path, project_root);
            let pattern_absolute = workspace_absolute_string_from_segments(&pattern_segments);
            pattern_absolute == path_workspace_absolute
        }
        PathPattern::Glob(glob) => {
            let matcher = glob.compile().compile_matcher();
            match (glob.anchor(), project_root) {
                (PathAnchor::WorkspaceAbsolute, _) => matcher.is_match(path_workspace_absolute),
                (PathAnchor::ProjectRelative, ProjectRoot::WorkspaceRoot) => {
                    let stripped = path_workspace_absolute
                        .strip_prefix('/')
                        .unwrap_or(path_workspace_absolute);
                    matcher.is_match(stripped)
                }
                (PathAnchor::ProjectRelative, ProjectRoot::Nested(proj_root)) => {
                    let mut prefix = String::new();
                    for seg in proj_root.segments().iter() {
                        prefix.push('/');
                        prefix.push_str(seg.as_str());
                    }
                    let Some(remainder) = path_workspace_absolute.strip_prefix(&prefix) else {
                        return false;
                    };
                    let Some(rel) = remainder.strip_prefix('/') else {
                        return false;
                    };
                    if rel.is_empty() {
                        return false;
                    }
                    matcher.is_match(rel)
                }
            }
        }
    }
}

/// BFS from `new_edge_to` through `augmented` looking for
/// `new_edge_from`. Returns the cycle's node set when a path is
/// found, [`None`] otherwise.
///
/// The caller has just inserted the edge `(new_edge_from,
/// new_edge_to)` into `augmented`; a length-≥2 cycle through that
/// edge exists iff `new_edge_to` already reached `new_edge_from`
/// via other edges. Self-loops (`from == to`) are ignored during
/// the walk because `DAG-014` explicitly carves out length-1
/// cycles as allowed (`DAG-013` self-producer is the canonical
/// in-place-transformer case).
fn find_cycle_through_new_edge(
    augmented: &BTreeSet<(TaskId, TaskId)>,
    new_edge_from: &TaskId,
    new_edge_to: &TaskId,
) -> Option<BTreeSet<TaskId>> {
    let mut parent: HashMap<TaskId, TaskId> = HashMap::new();
    let mut visited: BTreeSet<TaskId> = BTreeSet::from([new_edge_to.clone()]);
    let mut queue: VecDeque<TaskId> = VecDeque::from([new_edge_to.clone()]);
    while let Some(node) = queue.pop_front() {
        for (from, to) in augmented {
            if from != &node || from == to || visited.contains(to) {
                continue;
            }
            if to == new_edge_from {
                let mut cycle: BTreeSet<TaskId> = BTreeSet::new();
                cycle.insert(new_edge_from.clone());
                cycle.insert(node.clone());
                let mut cur = node.clone();
                while let Some(p) = parent.get(&cur).cloned() {
                    cycle.insert(p.clone());
                    if &p == new_edge_to {
                        break;
                    }
                    cur = p;
                }
                return Some(cycle);
            }
            parent.insert(to.clone(), node.clone());
            visited.insert(to.clone());
            queue.push_back(to.clone());
        }
    }
    None
}

/// Detect any `EXEC-019` runtime cycle that the just-completed
/// task's `materialised_outputs` introduce.
///
/// For each materialised path of `completing_task`, find every
/// task whose `inputs` patterns match it (the new producer-
/// matching edge `(completing_task, consumer)`). New edges are
/// inserted into `augmented_edges`; for each newly inserted edge
/// the function runs [`find_cycle_through_new_edge`]. On the
/// first detected cycle, a [`RuntimeInvariantViolation::RuntimeCycle`]
/// is appended to `invariant_violations` and the cycle's node
/// set is returned. Self-edges (`DAG-013` self-producer) are
/// added to `augmented_edges` but never trigger a cycle check
/// per `DAG-014`'s length-≥2 carve-out.
///
/// Returns [`None`] when no cycle was introduced.
pub(super) fn check_and_record_runtime_cycle_for_completion(
    augmented_edges: &mut BTreeSet<(TaskId, TaskId)>,
    invariant_violations: &mut Vec<RuntimeInvariantViolation>,
    workspace: &Workspace,
    completing_task: &TaskId,
    materialised_outputs: &[CanonicalPath],
) -> Option<BTreeSet<TaskId>> {
    for path in materialised_outputs {
        for consumer in tasks_matching_path(workspace, path) {
            let edge = (completing_task.clone(), consumer.clone());
            if !augmented_edges.insert(edge.clone()) {
                continue;
            }
            if &consumer == completing_task {
                continue;
            }
            if let Some(cycle) =
                find_cycle_through_new_edge(augmented_edges, completing_task, &consumer)
            {
                invariant_violations.push(RuntimeInvariantViolation::RuntimeCycle {
                    nodes: cycle.clone(),
                    offending_edge: edge,
                });
                return Some(cycle);
            }
        }
    }
    None
}

/// Move any cycle members still sitting in `state.ready` out of
/// the ready set and record them as
/// [`RunOutcome::Skipped`] with [`SkipCause::RuntimeCycle`].
/// Inserts each into `state.skip` so the cancellation drain
/// (`EXEC-013` step 1) does not re-emit a conflicting
/// `RunCancelled` for them.
///
/// Tasks already in flight or already terminal are unaffected:
/// the function targets the ready set only. Cycle members that
/// are in flight reach the same cancellation flow as `EXEC-013`
/// users via the internal cancel token (`EXEC-019` step 3),
/// so the `EXEC-014` SIGTERM / grace / SIGKILL dance applies
/// uniformly; cycle members that already completed keep their
/// truthful outcome.
pub(super) fn skip_ready_cycle_members<O>(
    observer: &O,
    state: &mut ReadyState,
    outcomes: &mut BTreeMap<TaskId, RunOutcome>,
    cycle_nodes: &BTreeSet<TaskId>,
) where
    O: RunObserver,
{
    let drained: Vec<TaskId> = state
        .ready
        .iter()
        .filter(|t| cycle_nodes.contains(*t))
        .cloned()
        .collect();
    for task in drained {
        state.ready.remove(&task);
        state.skip.insert(task.clone());
        let record = SkipRecord {
            task: task.clone(),
            cause: SkipCause::RuntimeCycle,
        };
        observer.on_task_skipped(&task, &record);
        outcomes.insert(task, RunOutcome::Skipped(record));
    }
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeSet;
    use std::path::Path;

    use haz_domain::path::CanonicalPath;
    use haz_domain::settings::WorkspaceSettings;
    use haz_domain::task_id::TaskId;
    use haz_vfs::WritableFilesystem;

    use crate::mock_impl::{MockProcessSpawner, MockSpec};
    use crate::run_graph::RuntimeInvariantViolation;
    use crate::run_graph::cycle::tasks_matching_path;
    use crate::run_graph::scheduler::run_graph;
    use crate::run_graph::test_fixtures::*;
    use crate::run_task::{CancelledRecord, RunOutcome, RunState};

    #[tokio::test]
    async fn exec_019_two_node_runtime_cycle_yields_diagnostic() {
        // A: outputs match B's inputs; B: outputs match A's
        // inputs. Two-node mutual producer cycle, detected at
        // runtime when the second-of-A-or-B completes.
        let task_a = make_task_with_inputs_outputs("a", vec!["b_out"], vec!["a_out"], vec![]);
        let task_b = make_task_with_inputs_outputs("b", vec!["a_out"], vec!["b_out"], vec![]);
        let p = make_project("p", BTreeSet::new(), vec![task_a, task_b]);
        let ws = make_workspace(vec![p], workspace_settings_with(fixed_cap(2)));
        let g = make_graph(vec![tid("p", "a"), tid("p", "b")], vec![]);
        let fixture = Fixture::new(ws, g);
        // Pre-write both input files so both tasks' cache-key
        // derivation can read them; pre-write both output
        // files so their `resolve_output_files` step succeeds.
        fixture
            .cache
            .fs()
            .write_file(Path::new("/ws/a_out"), b"a")
            .unwrap();
        fixture
            .cache
            .fs()
            .write_file(Path::new("/ws/b_out"), b"b")
            .unwrap();
        let spawner = MockProcessSpawner::new();
        push_n_default_specs(&spawner, 2);
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        let result = run_graph(&ctx, 1_700_000_000).await.unwrap();

        // Both tasks completed successfully (the cycle is a
        // run-level diagnostic; per-task outcomes are honest).
        assert_eq!(
            completed_for(&result.outcomes, &tid("p", "a")).state,
            RunState::Succeeded
        );
        assert_eq!(
            completed_for(&result.outcomes, &tid("p", "b")).state,
            RunState::Succeeded
        );

        // Exactly one RuntimeCycle violation. The cycle's node
        // set is {a, b}; the offending_edge direction depends
        // on which task completed last (FuturesUnordered-
        // dependent), so assert membership rather than
        // ordering.
        assert_eq!(result.invariant_violations.len(), 1);
        match &result.invariant_violations[0] {
            RuntimeInvariantViolation::RuntimeCycle {
                nodes,
                offending_edge,
            } => {
                let expected: BTreeSet<TaskId> = BTreeSet::from([tid("p", "a"), tid("p", "b")]);
                assert_eq!(nodes, &expected);
                assert!(expected.contains(&offending_edge.0));
                assert!(expected.contains(&offending_edge.1));
                assert_ne!(offending_edge.0, offending_edge.1);
            }
            other @ RuntimeInvariantViolation::OutputOverlap { .. } => {
                panic!("expected RuntimeCycle, got {other:?}")
            }
        }
    }

    #[tokio::test]
    async fn exec_019_cycle_through_static_hard_plus_runtime_producer() {
        // Static hard edge `a -> b` (b.deps = [a]) gives an
        // `a -> b` edge in augmented_edges from the start.
        // b's outputs match a's inputs at runtime; when b
        // completes, the new edge `b -> a` closes the cycle.
        let task_a = make_task_with_inputs_outputs("a", vec!["b_out"], vec!["a_out"], vec![]);
        let task_b = make_task_with_inputs_outputs("b", vec!["a_out"], vec!["b_out"], vec!["~:a"]);
        let p = make_project("p", BTreeSet::new(), vec![task_a, task_b]);
        let ws = make_workspace(vec![p], workspace_settings_with(fixed_cap(1)));
        let g = make_graph(
            vec![tid("p", "a"), tid("p", "b")],
            vec![h_edge(tid("p", "a"), tid("p", "b"))],
        );
        let fixture = Fixture::new(ws, g);
        fixture
            .cache
            .fs()
            .write_file(Path::new("/ws/a_out"), b"a")
            .unwrap();
        fixture
            .cache
            .fs()
            .write_file(Path::new("/ws/b_out"), b"b")
            .unwrap();
        let spawner = MockProcessSpawner::new();
        push_n_default_specs(&spawner, 2);
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        let result = run_graph(&ctx, 1_700_000_000).await.unwrap();

        // Static dep ensures a -> b ordering: a runs first.
        assert_eq!(
            completed_for(&result.outcomes, &tid("p", "a")).state,
            RunState::Succeeded
        );
        assert_eq!(
            completed_for(&result.outcomes, &tid("p", "b")).state,
            RunState::Succeeded
        );

        assert_eq!(result.invariant_violations.len(), 1);
        match &result.invariant_violations[0] {
            RuntimeInvariantViolation::RuntimeCycle {
                nodes,
                offending_edge,
            } => {
                assert_eq!(nodes, &BTreeSet::from([tid("p", "a"), tid("p", "b")]));
                // The runtime-discovered edge is b -> a (b's
                // outputs match a's inputs; b completes last
                // under the static a -> b ordering).
                assert_eq!(offending_edge, &(tid("p", "b"), tid("p", "a")));
            }
            other @ RuntimeInvariantViolation::OutputOverlap { .. } => {
                panic!("expected RuntimeCycle, got {other:?}")
            }
        }
    }

    #[tokio::test]
    async fn exec_019_three_node_cycle_yields_full_node_set() {
        // 3-cycle: a's outputs -> b's inputs (runtime a -> b),
        // b's outputs -> c's inputs (runtime b -> c), c's
        // outputs -> a's inputs (runtime c -> a). With cap=1
        // and lex admission, the cycle closes at c's
        // completion.
        let task_a = make_task_with_inputs_outputs("a", vec!["c_out"], vec!["a_out"], vec![]);
        let task_b = make_task_with_inputs_outputs("b", vec!["a_out"], vec!["b_out"], vec![]);
        let task_c = make_task_with_inputs_outputs("c", vec!["b_out"], vec!["c_out"], vec![]);
        let p = make_project("p", BTreeSet::new(), vec![task_a, task_b, task_c]);
        let ws = make_workspace(vec![p], workspace_settings_with(fixed_cap(1)));
        let g = make_graph(vec![tid("p", "a"), tid("p", "b"), tid("p", "c")], vec![]);
        let fixture = Fixture::new(ws, g);
        for name in ["a_out", "b_out", "c_out"] {
            fixture
                .cache
                .fs()
                .write_file(&Path::new("/ws").join(name), b"x")
                .unwrap();
        }
        let spawner = MockProcessSpawner::new();
        push_n_default_specs(&spawner, 3);
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        let result = run_graph(&ctx, 1_700_000_000).await.unwrap();

        // All three completed.
        for n in ["a", "b", "c"] {
            assert_eq!(
                completed_for(&result.outcomes, &tid("p", n)).state,
                RunState::Succeeded
            );
        }

        // One cycle diagnostic with all three members.
        assert_eq!(result.invariant_violations.len(), 1);
        match &result.invariant_violations[0] {
            RuntimeInvariantViolation::RuntimeCycle {
                nodes,
                offending_edge,
            } => {
                assert_eq!(
                    nodes,
                    &BTreeSet::from([tid("p", "a"), tid("p", "b"), tid("p", "c")])
                );
                // Lex admission under cap=1 runs a, b, c in
                // order; c completes last and adds c -> a.
                assert_eq!(offending_edge, &(tid("p", "c"), tid("p", "a")));
            }
            other @ RuntimeInvariantViolation::OutputOverlap { .. } => {
                panic!("expected RuntimeCycle, got {other:?}")
            }
        }
    }

    #[tokio::test]
    async fn exec_019_already_completed_member_keeps_succeeded_outcome() {
        // Reuse the two-node cycle: confirm that both members'
        // outcomes are Completed(Succeeded) rather than being
        // overwritten as Failed when the cycle is discovered.
        let task_a = make_task_with_inputs_outputs("a", vec!["b_out"], vec!["a_out"], vec![]);
        let task_b = make_task_with_inputs_outputs("b", vec!["a_out"], vec!["b_out"], vec![]);
        let p = make_project("p", BTreeSet::new(), vec![task_a, task_b]);
        let ws = make_workspace(vec![p], workspace_settings_with(fixed_cap(1)));
        let g = make_graph(vec![tid("p", "a"), tid("p", "b")], vec![]);
        let fixture = Fixture::new(ws, g);
        fixture
            .cache
            .fs()
            .write_file(Path::new("/ws/a_out"), b"a")
            .unwrap();
        fixture
            .cache
            .fs()
            .write_file(Path::new("/ws/b_out"), b"b")
            .unwrap();
        let spawner = MockProcessSpawner::new();
        push_n_default_specs(&spawner, 2);
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        let result = run_graph(&ctx, 1_700_000_000).await.unwrap();

        // R2: honour what actually happened; cycle members
        // that succeeded keep `Completed(Succeeded)`. No
        // rewrite of state to Failed.
        for n in ["a", "b"] {
            let outcome = result.outcomes.get(&tid("p", n));
            match outcome {
                Some(RunOutcome::Completed(record)) => {
                    assert_eq!(record.state, RunState::Succeeded);
                }
                other => panic!("expected Completed(Succeeded) for {n}, got {other:?}"),
            }
        }
        assert_eq!(result.invariant_violations.len(), 1);
    }

    #[tokio::test]
    async fn exec_019_in_flight_non_cycle_task_gets_cancelled() {
        // a and b form the cycle (mutual producer-consumer).
        // c is independent (no inputs overlapping a/b's
        // outputs). c's mock blocks on SIGTERM so it is
        // in-flight when the cycle is detected. EXEC-019
        // step 3's internal cancel token signals c; the
        // `EXEC-014` SIGTERM / grace / SIGKILL flow classifies
        // c as Cancelled(SignaledInFlight).
        let task_a = make_task_with_inputs_outputs("a", vec!["b_out"], vec!["a_out"], vec![]);
        let task_b = make_task_with_inputs_outputs("b", vec!["a_out"], vec!["b_out"], vec![]);
        let task_c = make_task_with_inputs_outputs("c", vec![], vec!["c_out"], vec![]);
        let p = make_project("p", BTreeSet::new(), vec![task_a, task_b, task_c]);
        let ws = make_workspace(vec![p], workspace_settings_with_grace(fixed_cap(3), 0.05));
        let g = make_graph(vec![tid("p", "a"), tid("p", "b"), tid("p", "c")], vec![]);
        let fixture = Fixture::new(ws, g);
        fixture
            .cache
            .fs()
            .write_file(Path::new("/ws/a_out"), b"a")
            .unwrap();
        fixture
            .cache
            .fs()
            .write_file(Path::new("/ws/b_out"), b"b")
            .unwrap();
        fixture
            .cache
            .fs()
            .write_file(Path::new("/ws/c_out"), b"c")
            .unwrap();
        let spawner = MockProcessSpawner::new();
        spawner.push_spec(MockSpec::default()); // a (Immediate)
        spawner.push_spec(MockSpec::default()); // b (Immediate)
        spawner.push_spec(exit_on_terminate_spec(0)); // c (OnTerminate)
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        let result = run_graph(&ctx, 1_700_000_000).await.unwrap();

        // a and b completed successfully (cycle members).
        for n in ["a", "b"] {
            assert_eq!(
                completed_for(&result.outcomes, &tid("p", n)).state,
                RunState::Succeeded
            );
        }

        // c was in flight when the cycle fired and got
        // signalled by the internal cancel token.
        match result.outcomes.get(&tid("p", "c")) {
            Some(RunOutcome::Cancelled(CancelledRecord::SignaledInFlight { .. })) => {}
            other => panic!("expected Cancelled(SignaledInFlight) for c, got {other:?}"),
        }

        // c's spawn (index 2 in the spawner) received at
        // least one signal.
        assert!(!spawner.signals_for(2).unwrap().is_empty());

        // One cycle diagnostic.
        assert_eq!(result.invariant_violations.len(), 1);
    }

    #[tokio::test]
    async fn exec_019_internal_cancel_does_not_pollute_user_token() {
        // After a runtime cycle, the user-supplied
        // CancellationToken (the parent of the scheduler's
        // internal child) MUST remain uncancelled. The
        // scheduler trips a child token only.
        let task_a = make_task_with_inputs_outputs("a", vec!["b_out"], vec!["a_out"], vec![]);
        let task_b = make_task_with_inputs_outputs("b", vec!["a_out"], vec!["b_out"], vec![]);
        let p = make_project("p", BTreeSet::new(), vec![task_a, task_b]);
        let ws = make_workspace(vec![p], workspace_settings_with(fixed_cap(2)));
        let g = make_graph(vec![tid("p", "a"), tid("p", "b")], vec![]);
        let fixture = Fixture::new(ws, g);
        fixture
            .cache
            .fs()
            .write_file(Path::new("/ws/a_out"), b"a")
            .unwrap();
        fixture
            .cache
            .fs()
            .write_file(Path::new("/ws/b_out"), b"b")
            .unwrap();
        let spawner = MockProcessSpawner::new();
        push_n_default_specs(&spawner, 2);
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        // Capture the user's token state before the run.
        assert!(!fixture.cancel.is_cancelled());

        let result = run_graph(&ctx, 1_700_000_000).await.unwrap();
        assert_eq!(result.invariant_violations.len(), 1);

        // The user's token stays uncancelled. The internal
        // child token, scoped to run_graph's body, was the
        // only one tripped.
        assert!(!fixture.cancel.is_cancelled());
    }

    #[tokio::test]
    async fn exec_019_self_edge_does_not_trigger_cycle() {
        // DAG-013 self-producer: a task whose `inputs` and
        // `outputs` patterns intersect is its own producer.
        // DAG-014 carves out length-1 cycles as allowed.
        // EXEC-019 detection MUST NOT report a runtime cycle
        // for the self-edge.
        let task = make_task_with_inputs_outputs("fmt", vec!["src.txt"], vec!["src.txt"], vec![]);
        let p = make_project("p", BTreeSet::new(), vec![task]);
        let ws = make_workspace(vec![p], workspace_settings_with(fixed_cap(1)));
        let g = make_graph(vec![tid("p", "fmt")], vec![]);
        let fixture = Fixture::new(ws, g);
        fixture
            .cache
            .fs()
            .write_file(Path::new("/ws/src.txt"), b"x")
            .unwrap();
        let spawner = MockProcessSpawner::new();
        push_n_default_specs(&spawner, 1);
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        let result = run_graph(&ctx, 1_700_000_000).await.unwrap();

        assert_eq!(
            completed_for(&result.outcomes, &tid("p", "fmt")).state,
            RunState::Succeeded
        );
        assert!(
            result.invariant_violations.is_empty(),
            "self-edge MUST NOT trigger EXEC-019; got {:?}",
            result.invariant_violations
        );
    }

    #[tokio::test]
    async fn exec_018_self_edge_task_runs_exactly_once_per_invocation() {
        // EXEC-018: a task whose declared `inputs` and `outputs`
        // patterns intersect (the canonical formatter case) runs
        // AT MOST ONCE per `haz` invocation. The executor MUST
        // NOT re-run the task in the same invocation even when
        // the task's outputs modify files matched by its inputs.
        //
        // The mock-spawned "command" writes new bytes to the
        // task's own input file; in a hypothetical re-run loop
        // the new contents would change the cache key and the
        // executor would re-spawn. EXEC-018 forbids that: the
        // run completes after a single spawn.
        let task = make_task_with_inputs_outputs("fmt", vec!["src.txt"], vec!["src.txt"], vec![]);
        let p = make_project("p", BTreeSet::new(), vec![task]);
        let ws = make_workspace(vec![p], workspace_settings_with(fixed_cap(1)));
        let g = make_graph(vec![tid("p", "fmt")], vec![]);
        let fixture = Fixture::new(ws, g);
        fixture
            .cache
            .fs()
            .write_file(Path::new("/ws/src.txt"), b"original")
            .unwrap();

        // The spec is silent on what the formatter "did"; we only
        // care that the executor does not loop. The mock command
        // succeeds without touching the filesystem; convergence
        // across invocations is `EXEC-018`'s second paragraph
        // (out of scope for this test).
        let spawner = MockProcessSpawner::new();
        push_n_default_specs(&spawner, 1);
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        let result = run_graph(&ctx, 1_700_000_000).await.unwrap();

        assert_eq!(
            completed_for(&result.outcomes, &tid("p", "fmt")).state,
            RunState::Succeeded,
        );
        assert_eq!(
            spawner.spawns().len(),
            1,
            "EXEC-018: self-edge task must spawn exactly once per invocation",
        );
        assert_eq!(
            count_started(&observer.events(), &tid("p", "fmt")),
            1,
            "EXEC-018: self-edge task must fire exactly one Started event",
        );
        // No invariant should be violated.
        assert!(
            result.invariant_violations.is_empty(),
            "self-edge run should not violate EXEC-019 / EXEC-020; got {:?}",
            result.invariant_violations,
        );
    }

    #[test]
    fn tasks_matching_path_literal_match_returns_consumer_task() {
        let task = make_task_with_inputs_outputs("consumer", vec!["foo.txt"], vec![], vec![]);
        let p = make_project("p", BTreeSet::new(), vec![task]);
        let ws = make_workspace(vec![p], WorkspaceSettings::default());
        let path = CanonicalPath::parse_workspace_absolute("/foo.txt").unwrap();
        let matched = tasks_matching_path(&ws, &path);
        assert_eq!(matched, BTreeSet::from([tid("p", "consumer")]));
    }

    #[test]
    fn tasks_matching_path_glob_match_returns_consumer_task() {
        let task = make_task_with_inputs_outputs("consumer", vec!["*.txt"], vec![], vec![]);
        let p = make_project("p", BTreeSet::new(), vec![task]);
        let ws = make_workspace(vec![p], WorkspaceSettings::default());
        let path = CanonicalPath::parse_workspace_absolute("/foo.txt").unwrap();
        let matched = tasks_matching_path(&ws, &path);
        assert_eq!(matched, BTreeSet::from([tid("p", "consumer")]));
    }

    #[test]
    fn tasks_matching_path_no_match_returns_empty() {
        let task = make_task_with_inputs_outputs("consumer", vec!["foo.txt"], vec![], vec![]);
        let p = make_project("p", BTreeSet::new(), vec![task]);
        let ws = make_workspace(vec![p], WorkspaceSettings::default());
        let path = CanonicalPath::parse_workspace_absolute("/different/path.bin").unwrap();
        let matched = tasks_matching_path(&ws, &path);
        assert!(matched.is_empty());
    }
}