haz_cli/commands/

run.rs

//! `haz run` subcommand.
//!
//! Wires the upstream layers (`crate::workspace_load`,
//! `crate::bearing`, `crate::target_resolve`) to either the
//! scheduler ([`haz_exec::run_graph::run_graph`]) or to a passive
//! dry-run report per `EXEC-030`.
//!
//! Current scope:
//!
//! - Real execution uses [`StdProcessSpawner`], a buffered output
//!   observer, and a never-cancelled token. Live mode and signal
//!   handling are not yet wired here.
//! - Dry-run prints the canonical-ordered selected task set to
//!   stdout in a human-readable form. Cache-state predictions are
//!   permitted by `EXEC-030` but not yet emitted; v1 emits the
//!   plain "would run N tasks" report.

use std::collections::{BTreeMap, BTreeSet};
use std::io::{self, Write};
use std::sync::{Arc, Mutex};
use std::time::{Instant, SystemTime, UNIX_EPOCH};

use haz_dag::edge::Edge;
use haz_dag::graph::TaskGraph;
use haz_domain::env::EnvVarName;
use haz_domain::settings::output::OutputMode;
use haz_domain::task_id::TaskId;
use haz_exec::exit_code::{CancellationSignal, exit_code_for};
use haz_exec::output::{BufferedOutputObserver, LiveOutputObserver};
use haz_exec::presenter::TaskPresenter;
use haz_exec::run_graph::{RunGraphOutcome, run_graph};
use haz_exec::run_task::RunContext;
use haz_exec::std_impl::StdProcessSpawner;
use miette::{IntoDiagnostic, Result, WrapErr};
use tokio_util::sync::CancellationToken;

use haz_target::{TargetSpec, resolve_targets};
use haz_vcs::workspace_changeset;

use crate::cli::RunArgs;
use crate::context::Context;
use crate::output::color_choice::ColorChoice;
use crate::output::epilogue::write_epilogue;
use crate::output::presenter::CliPresenter;

/// Execute the `haz run` subcommand against the supplied [`Context`].
///
/// Returns the process exit code that the binary should propagate.
///
/// # Errors
///
/// Returns a [`miette::Report`] on workspace-load, target-
/// resolution, or scheduler-level failures. Task-level failures
/// reach this function as a [`haz_exec::run_graph::RunGraphOutcome`]
/// whose `task_errors` / `outcomes` carry the per-task detail; the
/// returned exit code (via [`exit_code_for`]) signals the class.
pub fn execute(args: &RunArgs, ctx: &Context) -> Result<i32> {
    let bearing = ctx.bearing();

    // EXEC-026: compute the git changeset before target
    // resolution when a filter is active. The set lives as long
    // as `resolve_targets` needs it (one borrow); declaring it
    // here keeps the borrow lifetime obvious.
    let changeset = match &args.git_filter {
        Some(filter) => Some(
            workspace_changeset(filter, ctx.workspace().root.as_path())
                .into_diagnostic()
                .wrap_err("git-changeset enumeration failed")?,
        ),
        None => None,
    };
    let spec = TargetSpec {
        targets: args.targets.clone(),
        scope: args.scope,
    };
    let targets = resolve_targets(
        &spec,
        bearing,
        ctx.workspace(),
        ctx.graph(),
        changeset.as_ref(),
    )
    .into_diagnostic()
    .wrap_err("target resolution failed")?;

    if args.dry_run {
        let mut out = io::stdout().lock();
        return print_dry_run_plan(&targets, &mut out)
            .into_diagnostic()
            .wrap_err("failed to write dry-run plan")
            .map(|()| 0);
    }

    // `EXEC-016` precedence: CLI flag > workspace setting >
    // default (`OutputMode::Live` per `CFG-030`). The workspace
    // setting reads from `workspace.settings.output.mode`; the
    // workspace's defaults are already populated by
    // `WorkspaceSettings::default` when the explicit-mode
    // `workspace.yml` omits the `output:` mapping (`CFG-030`).
    let mode = args
        .output_mode
        .unwrap_or(ctx.workspace().settings.output.mode);
    let exec_graph = restrict_graph_to_targets(ctx.graph(), &targets);
    let color = ColorChoice::detect();
    run_to_exit(ctx, exec_graph, mode, color)
}

/// Emit the human-readable dry-run plan per the v1 format.
///
/// Tasks are emitted in canonical `(ProjectName, TaskName)` order
/// per `EXEC-003` (a [`BTreeSet`] of [`TaskId`] iterates in this
/// order natively).
fn print_dry_run_plan(targets: &BTreeSet<TaskId>, out: &mut impl Write) -> io::Result<()> {
    if targets.is_empty() {
        return writeln!(out, "would run 0 tasks.");
    }
    let suffix = if targets.len() == 1 { "" } else { "s" };
    writeln!(out, "would run {} task{suffix}:", targets.len())?;
    for id in targets {
        writeln!(out, "  - {id}")?;
    }
    Ok(())
}

/// Project the full [`TaskGraph`] onto the resolved target set.
///
/// The scheduler runs every node in its graph. To make `--all`
/// and explicit-positional selection actually narrow the run, the
/// graph passed to [`run_graph`] is pruned to the targets and
/// their cross-target edges. Per `EXEC-001` / `EXEC-002`, the
/// target set is already closed under hard predecessors, so the
/// pruned graph is admission-complete: every node's hard
/// predecessors are still in the graph.
fn restrict_graph_to_targets(graph: &TaskGraph, targets: &BTreeSet<TaskId>) -> TaskGraph {
    let nodes = targets.clone();
    let edges: BTreeSet<Edge> = graph
        .edges
        .iter()
        .filter(|e| nodes.contains(&e.from) && nodes.contains(&e.to))
        .cloned()
        .collect();
    TaskGraph { nodes, edges }
}

/// Build the scheduler context and run, returning the exit code.
///
/// An empty graph is treated as a successful zero-task run without
/// spinning up the async runtime (`EXEC-021` first bullet,
/// `EXEC-025` empty-workspace clause).
fn run_to_exit(
    ctx: &Context,
    graph: TaskGraph,
    mode: OutputMode,
    color: ColorChoice,
) -> Result<i32> {
    if graph.nodes.is_empty() {
        return Ok(0);
    }
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .into_diagnostic()
        .wrap_err("failed to build the tokio runtime")?;
    runtime.block_on(async move { dispatch_run_graph(ctx, graph, mode, color).await })
}

/// The async core of [`run_to_exit`].
///
/// Holds the cache, spawner, observer, and host-env snapshot
/// stack-locally so their borrows feed a single [`RunContext`]
/// value. Installs `SIGINT` / `SIGTERM` listeners on Unix that
/// fire the run's [`CancellationToken`] and record which signal
/// arrived; the recorded signal is then handed to
/// [`exit_code_for`] for POSIX `128 + signum` exit-code mapping
/// per `EXEC-021`.
async fn dispatch_run_graph(
    ctx: &Context,
    graph: TaskGraph,
    mode: OutputMode,
    color: ColorChoice,
) -> Result<i32> {
    let cache = ctx.cache_writer();
    let algo = cache.hash_algo();
    let spawner = StdProcessSpawner::new();
    let host_env = host_env_snapshot();
    let cancel = CancellationToken::new();
    let captured_signal: Arc<Mutex<Option<CancellationSignal>>> = Arc::new(Mutex::new(None));
    let signal_handle = spawn_signal_listener(cancel.clone(), Arc::clone(&captured_signal));
    let created_at_unix = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |d| d.as_secs());
    let presenter: Arc<dyn TaskPresenter> = Arc::new(CliPresenter::new(color));

    // Wall-clock pair captured around the scheduler call so the
    // run-level epilogue can report total elapsed time, whatever
    // path each individual task took (cache hit, fresh run, skip,
    // cancel).
    let started_at = Instant::now();

    // `EXEC-016` observer selection. Both observers carry their
    // own `Mutex` internally for cross-task line / block atomicity;
    // both write directly to `io::stdout` / `io::stderr`.
    // Branching inside `dispatch_run_graph` keeps the run-context
    // borrow lifetimes obvious without erasing the observer type.
    let outcome = match mode {
        OutputMode::Live => {
            let observer = LiveOutputObserver::with_presenter(
                io::stdout(),
                io::stderr(),
                Arc::clone(&presenter),
            );
            let run_ctx = RunContext {
                fs: cache.fs(),
                cache: &cache,
                spawner: &spawner,
                observer: &observer,
                workspace: ctx.workspace(),
                graph: &graph,
                host_env: &host_env,
                algo,
                cancel: &cancel,
            };
            run_graph(&run_ctx, created_at_unix).await
        }
        OutputMode::Buffered => {
            let observer = BufferedOutputObserver::with_presenter(
                io::stdout(),
                io::stderr(),
                Arc::clone(&presenter),
            );
            let run_ctx = RunContext {
                fs: cache.fs(),
                cache: &cache,
                spawner: &spawner,
                observer: &observer,
                workspace: ctx.workspace(),
                graph: &graph,
                host_env: &host_env,
                algo,
                cancel: &cancel,
            };
            run_graph(&run_ctx, created_at_unix).await
        }
    }
    .map_err(|err| miette::miette!("scheduler failed: {err}"))?;

    let elapsed = started_at.elapsed();

    // Whether or not a signal arrived, the listener task is no
    // longer needed once the scheduler returns. Abort it so it
    // does not outlive this function. (On Unix the inner
    // `select!` blocks on signal streams indefinitely; we MUST
    // abort or the runtime would not shut down cleanly.)
    signal_handle.abort();

    emit_run_epilogue(&outcome, elapsed, color);

    let signal = *captured_signal
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    Ok(exit_code_for(&outcome, signal))
}

/// Write the run-level epilogue to stderr.
///
/// The epilogue is presentation-only and follows every task's
/// per-task summary line; emitting it on stderr (rather than
/// stdout) keeps the stdout stream a strict echo of the tasks'
/// captured bytes per `EXEC-016`.
fn emit_run_epilogue(outcome: &RunGraphOutcome, elapsed: std::time::Duration, color: ColorChoice) {
    let mut stderr = io::stderr().lock();
    let _ = write_epilogue(&mut stderr, outcome, elapsed, color);
}

/// Spawn the signal-listener background task.
///
/// On Unix the task awaits the first `SIGINT` or `SIGTERM`,
/// records the corresponding [`CancellationSignal`] in
/// `captured`, and fires `cancel` so the scheduler stops
/// admitting new tasks and signals in-flight children
/// (`EXEC-012`..`EXEC-014`). The task is aborted by the caller
/// once the scheduler returns; it does NOT race itself to
/// completion.
///
/// On non-Unix targets the task is a no-op: signal-driven
/// cancellation is not yet implemented for those platforms, so
/// runs there always observe a never-fired token. A future
/// revision MAY add Windows `Ctrl+C` handling.
fn spawn_signal_listener(
    cancel: CancellationToken,
    captured: Arc<Mutex<Option<CancellationSignal>>>,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(listen_for_signals(cancel, captured))
}

#[cfg(unix)]
async fn listen_for_signals(
    cancel: CancellationToken,
    captured: Arc<Mutex<Option<CancellationSignal>>>,
) {
    use tokio::signal::unix::{SignalKind, signal};
    // If signal handlers cannot be installed (an extremely narrow
    // OS-level failure), the listener silently degrades to a
    // no-op. The run is still cancellable via the OS default-
    // action path (which would terminate the process); the
    // cooperative cancellation path simply is not available.
    let (Ok(mut sigint), Ok(mut sigterm)) = (
        signal(SignalKind::interrupt()),
        signal(SignalKind::terminate()),
    ) else {
        return;
    };
    tokio::select! {
        _ = sigint.recv() => {
            record_signal(&captured, CancellationSignal::Interrupt);
            cancel.cancel();
        }
        _ = sigterm.recv() => {
            record_signal(&captured, CancellationSignal::Terminate);
            cancel.cancel();
        }
    }
}

#[cfg(not(unix))]
async fn listen_for_signals(
    _cancel: CancellationToken,
    _captured: Arc<Mutex<Option<CancellationSignal>>>,
) {
    // Platform without `tokio::signal::unix`. A future revision
    // MAY add `tokio::signal::ctrl_c` handling for Windows;
    // until then, runs on non-Unix observe a never-fired token.
}

#[cfg(unix)]
fn record_signal(captured: &Mutex<Option<CancellationSignal>>, sig: CancellationSignal) {
    let mut slot = captured
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    *slot = Some(sig);
}

/// Snapshot the host environment into the form `haz-exec` expects.
///
/// Variables whose names violate `EnvVarName`'s identifier rules
/// are silently dropped: such names cannot appear in any cache
/// manifest or task `env` mapping anyway, so they are
/// unreferenceable from the workspace.
fn host_env_snapshot() -> BTreeMap<EnvVarName, String> {
    let mut out = BTreeMap::new();
    for (key, value) in std::env::vars() {
        if let Ok(name) = EnvVarName::try_new(key) {
            out.insert(name, value);
        }
    }
    out
}

#[cfg(test)]
mod tests {
    use super::{execute, print_dry_run_plan};
    use crate::cli::{RunArgs, Scope, TaskRef};
    use crate::context::Context;
    use haz_domain::name::{ProjectName, TaskName};
    use haz_domain::task_id::TaskId;
    use miette::Result;
    use std::collections::BTreeSet;
    use std::fs;
    use std::path::Path;
    use std::str::FromStr;

    fn write_file(path: &Path, contents: &str) {
        if let Some(parent) = path.parent() {
            fs::create_dir_all(parent).unwrap();
        }
        fs::write(path, contents).unwrap();
    }

    fn run_run(args: &RunArgs, cwd: &Path) -> Result<i32> {
        execute(args, &Context::open(cwd)?)
    }

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

    fn dry_run_args() -> RunArgs {
        RunArgs {
            targets: Vec::new(),
            scope: Scope::Default,
            git_filter: None,
            dry_run: true,
            output_mode: None,
        }
    }

    fn run_args_with_targets(targets: &[&str]) -> RunArgs {
        RunArgs {
            targets: targets.iter().map(|s| TaskRef::parse(s).unwrap()).collect(),
            scope: Scope::Default,
            git_filter: None,
            dry_run: false,
            output_mode: None,
        }
    }

    #[test]
    fn dry_run_plan_empty_targets() {
        let targets: BTreeSet<TaskId> = BTreeSet::new();
        let mut out = Vec::<u8>::new();
        print_dry_run_plan(&targets, &mut out).unwrap();
        assert_eq!(String::from_utf8(out).unwrap(), "would run 0 tasks.\n");
    }

    #[test]
    fn dry_run_plan_single_target_singular_count() {
        let mut targets = BTreeSet::new();
        targets.insert(id("lib", "build"));
        let mut out = Vec::<u8>::new();
        print_dry_run_plan(&targets, &mut out).unwrap();
        assert_eq!(
            String::from_utf8(out).unwrap(),
            "would run 1 task:\n  - lib:build\n",
        );
    }

    #[test]
    fn dry_run_plan_multiple_targets_canonical_order() {
        let mut targets = BTreeSet::new();
        targets.insert(id("z", "build"));
        targets.insert(id("a", "test"));
        targets.insert(id("a", "build"));
        let mut out = Vec::<u8>::new();
        print_dry_run_plan(&targets, &mut out).unwrap();
        assert_eq!(
            String::from_utf8(out).unwrap(),
            "would run 3 tasks:\n  - a:build\n  - a:test\n  - z:build\n",
        );
    }

    /// Path to a `true(1)` binary the smoke test can spawn. The
    /// host-env snapshot allow-lists by name and the spawner clears
    /// the child's env, so a bare `"true"` would fail PATH lookup
    /// in the child; using an absolute path sidesteps the issue.
    /// Linux + macOS both ship `/usr/bin/true`; the smoke test is
    /// skipped on hosts that lack it.
    fn true_path() -> Option<&'static str> {
        const CANDIDATES: &[&str] = &["/usr/bin/true", "/bin/true"];
        CANDIDATES.iter().copied().find(|p| Path::new(p).exists())
    }

    #[test]
    fn execute_dry_run_on_singleton_workspace_returns_zero() {
        let tempdir = tempfile::TempDir::new().unwrap();
        let root = tempdir.path();
        fs::create_dir_all(root.join(".git")).unwrap();
        write_file(
            &root.join("haz.yml"),
            "name: only_proj\ntasks:\n  default:\n    command: [\"true\"]\n",
        );
        let args = dry_run_args();
        let code = run_run(&args, root).unwrap();
        assert_eq!(code, 0);
    }

    #[test]
    fn execute_runs_trivial_task_to_success() {
        let Some(true_bin) = true_path() else {
            eprintln!("skipping: no `true` binary at /usr/bin/true or /bin/true");
            return;
        };
        let tempdir = tempfile::TempDir::new().unwrap();
        let root = tempdir.path();
        fs::create_dir_all(root.join(".git")).unwrap();
        write_file(
            &root.join("haz.yml"),
            &format!("name: only_proj\ntasks:\n  default:\n    command: [\"{true_bin}\"]\n"),
        );
        // `output_mode: None` falls back to the workspace setting,
        // whose default per `CFG-030` is `OutputMode::Live`.
        let args = run_args_with_targets(&["~:default"]);
        let code = run_run(&args, root).unwrap();
        assert_eq!(code, 0);
    }

    #[test]
    fn execute_runs_trivial_task_in_buffered_mode() {
        // Explicit `--output-mode buffered` exercises the
        // `OutputMode::Buffered` arm of `dispatch_run_graph`'s
        // observer selection. The wiring is the only change under
        // test; the smoke target is the same `true`-spawning
        // workspace.
        let Some(true_bin) = true_path() else {
            eprintln!("skipping: no `true` binary at /usr/bin/true or /bin/true");
            return;
        };
        let tempdir = tempfile::TempDir::new().unwrap();
        let root = tempdir.path();
        fs::create_dir_all(root.join(".git")).unwrap();
        write_file(
            &root.join("haz.yml"),
            &format!("name: only_proj\ntasks:\n  default:\n    command: [\"{true_bin}\"]\n"),
        );
        let mut args = run_args_with_targets(&["~:default"]);
        args.output_mode = Some(crate::cli::OutputMode::Buffered);
        let code = run_run(&args, root).unwrap();
        assert_eq!(code, 0);
    }

    #[test]
    fn exec_023_execute_default_target_without_bearing_errors() {
        // cwd at the workspace root but in explicit mode means no
        // project encloses it; `EXEC-023`'s NoBearingForDefault
        // arm fires.
        let tempdir = tempfile::TempDir::new().unwrap();
        let root = tempdir.path();
        write_file(&root.join(".haz/workspace.yml"), "projects:\n  - /lib\n");
        write_file(
            &root.join("lib/haz.yml"),
            "name: lib\ntasks:\n  default:\n    command: [\"true\"]\n",
        );
        let args = RunArgs {
            targets: Vec::new(),
            scope: Scope::Default,
            git_filter: None,
            dry_run: false,
            output_mode: None,
        };
        let err = run_run(&args, root).unwrap_err();
        assert!(
            format!("{err:?}").contains("no bearing project"),
            "diagnostic should mention the no-bearing case: {err:?}",
        );
    }

    // --- Git changeset wiring (EXEC-026..029) ---

    use crate::cli::GitFilter;
    use std::process::Command;

    fn git(cwd: &Path, args: &[&str]) {
        let output = Command::new("git")
            .arg("-C")
            .arg(cwd)
            .args(args)
            .output()
            .expect("git binary must be available on PATH for these tests");
        assert!(
            output.status.success(),
            "git {} failed:\nstdout: {}\nstderr: {}",
            args.join(" "),
            String::from_utf8_lossy(&output.stdout),
            String::from_utf8_lossy(&output.stderr),
        );
    }

    fn init_repo(root: &Path) {
        git(root, &["init", "--quiet", "--initial-branch=main"]);
        git(root, &["config", "user.email", "test@example.invalid"]);
        git(root, &["config", "user.name", "Test"]);
        git(root, &["config", "commit.gpgsign", "false"]);
    }

    fn dry_run_with_filter(filter: GitFilter) -> RunArgs {
        RunArgs {
            targets: Vec::new(),
            scope: Scope::Workspace,
            git_filter: Some(filter),
            dry_run: true,
            output_mode: None,
        }
    }

    /// Build the smoke-test workspace: two sibling projects under
    /// a shared git repository, each declaring a single task whose
    /// input is its own `src/main.rs`.
    fn write_two_project_workspace(root: &Path) {
        write_file(
            &root.join(".haz/workspace.yml"),
            "projects:\n  - /lib_a\n  - /lib_b\n",
        );
        write_file(
            &root.join("lib_a/haz.yml"),
            "name: lib_a\ntasks:\n  build:\n    command: [\"true\"]\n    inputs:\n      - src/main.rs\n",
        );
        write_file(
            &root.join("lib_b/haz.yml"),
            "name: lib_b\ntasks:\n  build:\n    command: [\"true\"]\n    inputs:\n      - src/main.rs\n",
        );
        write_file(&root.join("lib_a/src/main.rs"), "fn main() {}\n");
        write_file(&root.join("lib_b/src/main.rs"), "fn main() {}\n");
    }

    #[test]
    fn execute_git_changed_dry_run_succeeds_when_one_input_changes() {
        // Two projects sharing a git repo; only `lib_a/src/main.rs`
        // is modified after the initial commit. With `--git-changed
        // --dry-run --workspace`, target resolution must narrow to
        // `lib_a:build` (the closure on the singleton seed adds
        // nothing). The smoke test asserts the wiring runs to exit
        // 0; the per-branch matching invariants are covered by the
        // unit tests in target_resolve and git_changeset.
        let tempdir = tempfile::TempDir::new().unwrap();
        let root = tempdir.path();
        init_repo(root);
        write_two_project_workspace(root);
        git(root, &["add", "."]);
        git(root, &["commit", "--quiet", "-m", "init"]);

        // Modify only lib_a's input.
        write_file(&root.join("lib_a/src/main.rs"), "fn main() { /* v2 */ }\n");

        let args = dry_run_with_filter(GitFilter::Changed);
        let code = run_run(&args, root).unwrap();
        assert_eq!(code, 0);
    }

    #[test]
    fn execute_git_changed_dry_run_succeeds_when_no_input_changes() {
        // No file is modified after the initial commit. The
        // changeset is empty; the target set is empty; the dry-run
        // plan emits "would run 0 tasks." and exits 0.
        let tempdir = tempfile::TempDir::new().unwrap();
        let root = tempdir.path();
        init_repo(root);
        write_two_project_workspace(root);
        git(root, &["add", "."]);
        git(root, &["commit", "--quiet", "-m", "init"]);

        let args = dry_run_with_filter(GitFilter::Changed);
        let code = run_run(&args, root).unwrap();
        assert_eq!(code, 0);
    }

    #[test]
    fn execute_git_changed_surfaces_not_a_repo_error() {
        // Workspace without a git repo: the filter MUST fail with
        // a typed error before any task is selected.
        let tempdir = tempfile::TempDir::new().unwrap();
        let root = tempdir.path();
        write_two_project_workspace(root);

        let args = dry_run_with_filter(GitFilter::Changed);
        let err = run_run(&args, root).unwrap_err();
        let rendered = format!("{err:?}");
        assert!(
            rendered.contains("not inside a git repository")
                || rendered.contains("git-changeset enumeration"),
            "diagnostic should name the missing-repo failure: {rendered}",
        );
    }
}