alba-engine 0.2.1

//! Behaviour tests for the parallel scheduler, driven end to end through
//! [`alba_engine::run`] against `alba_executors`' `FakeExecutor` — no real
//! process is ever spawned, so the scheduling semantics (dependency order,
//! the parallelism bound, fail-fast, `keep_going`, `allow_failure`,
//! cancellation, and event ordering) are asserted deterministically.
//!
//! Timing: the scripted delays are chosen with wide margins (tens of
//! milliseconds where an ordering must hold, hundreds where a beam must
//! still be running when another finishes) so a loaded machine does not
//! flip an assertion. Tests whose assertions depend on *which* beam takes
//! the single available permit first deliberately run on the
//! current-thread runtime, where tasks are polled in spawn order; the ones
//! asserting real concurrency run on a multi-threaded one.

use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use alba_core::{BeamId, load_str};
use alba_engine::{
    BeamStatus, EngineError, Executors, RunEvent, RunOptions, RunSummary, Targets, run,
};
use alba_executors::{
    BeamContext, CommandSpec, ExecContext, ExecError, ExecResult, ExecSession, Executor,
    FakeBehavior, FakeEvent, FakeExecutor, OutputLine, Stream,
};
use tokio_util::sync::CancellationToken;

/// Everything one `run` produced: its result and every event it emitted,
/// drained after the run returned (all senders are dropped by then, so
/// nothing can still be in flight).
struct Outcome {
    result: Result<RunSummary, EngineError>,
    events: Vec<RunEvent>,
}

impl Outcome {
    fn summary(&self) -> &RunSummary {
        self.result.as_ref().expect("the run must not fail")
    }

    fn error(&self) -> &EngineError {
        self.result.as_ref().expect_err("the run must fail")
    }
}

fn options(jobs: usize, keep_going: bool) -> RunOptions {
    RunOptions {
        jobs,
        keep_going,
        params: Vec::new(),
        cache: None,
        extra_env: Vec::new(),
    }
}

fn behavior(exit_code: i32, delay_ms: u64) -> FakeBehavior {
    FakeBehavior {
        exit_code,
        delay: Duration::from_millis(delay_ms),
        output_lines: Vec::new(),
    }
}

fn stdout(text: &str) -> OutputLine {
    OutputLine {
        stream: Stream::Stdout,
        text: text.to_string(),
    }
}

async fn run_target(
    source: &str,
    target: &str,
    options: RunOptions,
    executor: Arc<dyn Executor>,
) -> Outcome {
    run_target_with_cancel(source, target, options, executor, CancellationToken::new()).await
}

async fn run_target_with_cancel(
    source: &str,
    target: &str,
    options: RunOptions,
    executor: Arc<dyn Executor>,
    cancel: CancellationToken,
) -> Outcome {
    run_target_with_executors(
        source,
        target,
        options,
        Executors::uniform(executor),
        cancel,
    )
    .await
}

/// Like [`run_target_with_cancel`], but for a scenario that needs the
/// embedded and system slots to be distinguishable executors rather than
/// one executor wearing both hats.
async fn run_target_with_executors(
    source: &str,
    target: &str,
    options: RunOptions,
    executors: Executors,
    cancel: CancellationToken,
) -> Outcome {
    let project = load_str(source).expect("the test Beamfile must load");
    let (events_tx, mut events_rx) = tokio::sync::mpsc::unbounded_channel();

    let result = run(
        &project,
        &Targets::beam(BeamId(target.to_string())),
        options,
        executors,
        events_tx,
        cancel,
    )
    .await;

    let mut events = Vec::new();
    while let Ok(event) = events_rx.try_recv() {
        events.push(event);
    }
    Outcome { result, events }
}

fn ids(beams: &[BeamId]) -> Vec<&str> {
    beams.iter().map(|id| id.0.as_str()).collect()
}

fn commands(executor: &FakeExecutor) -> Vec<String> {
    executor
        .calls()
        .into_iter()
        .map(|call| call.command)
        .collect()
}

fn position(commands: &[String], command: &str) -> usize {
    commands
        .iter()
        .position(|candidate| candidate == command)
        .unwrap_or_else(|| panic!("`{command}` was never executed, ran: {commands:?}"))
}

fn positions(events: &[RunEvent], predicate: impl Fn(&RunEvent) -> bool) -> Vec<usize> {
    events
        .iter()
        .enumerate()
        .filter(|(_, event)| predicate(event))
        .map(|(index, _)| index)
        .collect()
}

fn only(indexes: Vec<usize>, what: &str) -> usize {
    assert_eq!(indexes.len(), 1, "expected exactly one {what}");
    indexes[0]
}

/// Dependency order, and that independent beams are not accidentally
/// serialized. The `jobs` bound itself is pinned by
/// [`parallelism_never_exceeds_the_jobs_limit`]: this diamond can never
/// exceed two concurrent beams by its own shape, so its `running_peak`
/// assertion would survive the semaphore being removed.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn respects_dependency_order_and_runs_independent_beams_concurrently() {
    const SOURCE: &str = r#"
beam a { run "step a" }
beam b { needs [a] run "step b" }
beam c { needs [a] run "step c" }
beam d { needs [b, c] run "step d" }
"#;

    let executor = Arc::new(FakeExecutor::new().on("step", behavior(0, 50)));
    let outcome = run_target(SOURCE, "d", options(2, false), executor.clone()).await;

    let commands = commands(&executor);
    assert_eq!(
        commands.len(),
        4,
        "every beam runs exactly once: {commands:?}"
    );
    let (a, b) = (position(&commands, "step a"), position(&commands, "step b"));
    let (c, d) = (position(&commands, "step c"), position(&commands, "step d"));
    assert!(
        a < b && a < c,
        "`a` runs before its dependents: {commands:?}"
    );
    assert!(b < d && c < d, "`d` runs last: {commands:?}");

    assert_eq!(
        executor.running_peak(),
        2,
        "`b` and `c` are independent and must overlap under jobs=2"
    );

    let summary = outcome.summary();
    assert_eq!(ids(&summary.succeeded), ["a", "b", "c", "d"]);
    assert_eq!(summary.exit_code(), 0);
}

/// `--jobs N` really bounds how many beams run at once. Four mutually
/// independent beams would all overlap if nothing held them back, so the
/// peak is the bound itself and not an artifact of the graph's shape.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn parallelism_never_exceeds_the_jobs_limit() {
    const SOURCE: &str = r#"
beam one { run "step one" }
beam two { run "step two" }
beam three { run "step three" }
beam four { run "step four" }
beam all { needs [one, two, three, four] run "step all" }
"#;

    let executor = Arc::new(FakeExecutor::new().on("step", behavior(0, 50)));
    let outcome = run_target(SOURCE, "all", options(2, false), executor.clone()).await;
    assert_eq!(outcome.summary().cancelled.len(), 0);
    assert_eq!(
        executor.running_peak(),
        2,
        "four independent beams under jobs=2 must never exceed two at once"
    );

    let executor = Arc::new(FakeExecutor::new().on("step", behavior(0, 20)));
    let outcome = run_target(SOURCE, "all", options(1, false), executor.clone()).await;
    assert_eq!(outcome.summary().succeeded.len(), 5);
    assert_eq!(
        executor.running_peak(),
        1,
        "jobs=1 must serialize the run completely"
    );
}

/// Fail-fast cancels every beam that has not started yet, but a beam that
/// already holds a permit keeps running to completion. Note the
/// counter-intuitive consequence the rule implies and this test pins:
/// `d` needs only `c`, which succeeds, yet `d` is still cancelled — `b`'s
/// failure cancels *all* not-yet-started beams, not just its own
/// dependents.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn fail_fast_cancels_pending_but_lets_running_finish() {
    const SOURCE: &str = r#"
beam b { run "fail b" }
beam c { run "slow c" }
beam d { needs [c] run "step d" }
beam all { needs [b, d] run "step all" }
"#;

    let executor = Arc::new(
        FakeExecutor::new()
            .on("fail b", behavior(1, 50))
            .on("slow c", behavior(0, 300)),
    );
    let outcome = run_target(SOURCE, "all", options(2, false), executor.clone()).await;

    let summary = outcome.summary();
    assert_eq!(ids(&summary.failed), ["b"]);
    assert_eq!(
        ids(&summary.succeeded),
        ["c"],
        "`c` was already running when `b` failed and must finish"
    );
    assert_eq!(ids(&summary.cancelled), ["d", "all"]);
    assert_eq!(summary.exit_code(), 1);

    // `b` and `c` start concurrently, so the order they were recorded in
    // is not meaningful here — only that nothing else ever started.
    let mut commands = commands(&executor);
    commands.sort();
    assert_eq!(
        commands,
        ["fail b", "slow c"],
        "no beam starts after the failure"
    );
}

/// With `keep_going`, a failure no longer cancels the beams that have not
/// started: `d` is independent of `b` and still runs. What `keep_going`
/// does *not* change is dependency propagation — `e` needs the failed `b`
/// and `all` needs the cancelled `e`, so both are cancelled anyway.
///
/// Runs on the current-thread runtime with `jobs = 1` on purpose: tasks are
/// polled in spawn order (declaration order) and the semaphore is fair, so
/// `b` provably takes the only permit first and `d` is still pending when
/// `b` fails — which is exactly the situation `keep_going` governs.
#[tokio::test]
async fn keep_going_runs_independent_beams() {
    const SOURCE: &str = r#"
beam b { run "fail b" }
beam d { run "step d" }
beam e { needs [b] run "step e" }
beam all { needs [d, e] run "step all" }
"#;

    let executor = Arc::new(
        FakeExecutor::new()
            .on("fail b", behavior(1, 10))
            .on("step d", behavior(0, 10)),
    );
    let outcome = run_target(SOURCE, "all", options(1, true), executor.clone()).await;

    let summary = outcome.summary();
    assert_eq!(ids(&summary.failed), ["b"]);
    assert_eq!(ids(&summary.succeeded), ["d"]);
    assert_eq!(ids(&summary.cancelled), ["e", "all"]);
    assert_eq!(summary.exit_code(), 1);

    assert_eq!(
        commands(&executor),
        ["fail b", "step d"],
        "`d` was still pending when `b` failed, and ran anyway"
    );
}

#[tokio::test]
async fn allow_failure_does_not_stop_the_run() {
    const SOURCE: &str = r#"
beam lint { allow_failure true run "run lint" }
beam build { needs [lint] run "run build" }
"#;

    let executor = Arc::new(FakeExecutor::new().on("run lint", behavior(1, 0)));
    let outcome = run_target(SOURCE, "build", options(2, false), executor.clone()).await;

    let summary = outcome.summary();
    assert_eq!(ids(&summary.failed_allowed), ["lint"]);
    assert_eq!(ids(&summary.succeeded), ["build"]);
    assert!(summary.failed.is_empty());
    assert!(summary.cancelled.is_empty());
    assert_eq!(
        summary.exit_code(),
        0,
        "an allowed failure does not affect the exit code"
    );

    assert_eq!(commands(&executor), ["run lint", "run build"]);
}

/// A beam's commands run one after another and stop at the first failure:
/// a beam declaring `["build", "deploy"]` must not deploy what did not
/// build.
#[tokio::test]
async fn a_failing_command_skips_the_rest_of_its_beam() {
    const SOURCE: &str = r#"
beam release { run ["fail first", "never second"] }
"#;

    let executor = Arc::new(FakeExecutor::new().on("fail first", behavior(3, 0)));
    let outcome = run_target(SOURCE, "release", options(2, false), executor.clone()).await;

    assert_eq!(
        commands(&executor),
        ["fail first"],
        "the second command must never run"
    );
    let summary = outcome.summary();
    assert_eq!(ids(&summary.failed), ["release"]);
    assert_eq!(summary.exit_code(), 1);
}

/// The same rule under `allow_failure`: the beam's outcome is forgiving,
/// its remaining commands are not still run.
#[tokio::test]
async fn a_failing_command_skips_the_rest_even_when_failure_is_allowed() {
    const SOURCE: &str = r#"
beam lint { allow_failure true run ["fail first", "never second"] }
"#;

    let executor = Arc::new(FakeExecutor::new().on("fail first", behavior(1, 0)));
    let outcome = run_target(SOURCE, "lint", options(2, false), executor.clone()).await;

    assert_eq!(commands(&executor), ["fail first"]);
    assert_eq!(ids(&outcome.summary().failed_allowed), ["lint"]);
}

/// A `run` template that cannot be rendered at schedule time — here an
/// environment variable that is deliberately deferred to that moment and
/// turns out to be unset — is that beam's failure, not the run's. It gets
/// the same treatment as a command that cannot be spawned: a started
/// event, the message as stderr output, a failed status, and a run that
/// keeps going for everything that does not depend on it.
#[tokio::test]
async fn a_template_that_cannot_render_fails_only_its_own_beam() {
    const SOURCE: &str = r#"
beam broken { run "echo {env('ALBA_MISSING_VAR_XYZ')}" }
beam other { run "step other" }
beam all { needs [broken, other] run "step all" }
"#;

    let executor = Arc::new(FakeExecutor::new());
    let outcome = run_target(SOURCE, "all", options(2, true), executor.clone()).await;

    let summary = outcome.summary();
    assert_eq!(ids(&summary.failed), ["broken"]);
    assert_eq!(
        ids(&summary.succeeded),
        ["other"],
        "a beam whose template fails must not stop the rest of the run"
    );
    assert_eq!(ids(&summary.cancelled), ["all"]);
    assert_eq!(summary.exit_code(), 1);

    let events = &outcome.events;
    let started = only(
        positions(
            events,
            |event| matches!(event, RunEvent::BeamStarted { id } if id.0 == "broken"),
        ),
        "`broken` started event",
    );
    let reported = only(
        positions(events, |event| {
            matches!(event, RunEvent::BeamOutput { id, line, .. }
                if id.0 == "broken"
                    && line.stream == Stream::Stderr
                    && line.text.contains("ALBA_MISSING_VAR_XYZ"))
        }),
        "render failure reported as stderr output",
    );
    let finished = only(
        positions(
            events,
            |event| matches!(event, RunEvent::BeamFinished { id, .. } if id.0 == "broken"),
        ),
        "`broken` finished event",
    );
    assert!(started < reported && reported < finished);

    let last = events.last().expect("the run must emit events");
    assert!(
        matches!(last, RunEvent::RunFinished { .. }),
        "the run must still end with its summary, got {last:?}"
    );
}

/// The forgiving half of the same rule: a beam that declares
/// `allow_failure` and whose template cannot render lands in
/// `failed_allowed`, so its dependents still run.
#[tokio::test]
async fn a_template_that_cannot_render_honours_allow_failure() {
    const SOURCE: &str = r#"
beam lint { allow_failure true run "echo {env('ALBA_MISSING_VAR_XYZ')}" }
beam build { needs [lint] run "step build" }
"#;

    let executor = Arc::new(FakeExecutor::new());
    let outcome = run_target(SOURCE, "build", options(2, false), executor.clone()).await;

    let summary = outcome.summary();
    assert_eq!(ids(&summary.failed_allowed), ["lint"]);
    assert_eq!(ids(&summary.succeeded), ["build"]);
    assert_eq!(summary.exit_code(), 0);
}

/// Without `--keep-going`, a template failure stops the beams that have
/// not started, exactly like any other failure — rather than abandoning
/// the run and discarding what already succeeded.
#[tokio::test]
async fn a_template_that_cannot_render_triggers_fail_fast() {
    const SOURCE: &str = r#"
beam broken { run "echo {env('ALBA_MISSING_VAR_XYZ')}" }
beam after { needs [broken] run "step after" }
"#;

    let executor = Arc::new(FakeExecutor::new());
    let outcome = run_target(SOURCE, "after", options(1, false), executor.clone()).await;

    let summary = outcome.summary();
    assert_eq!(ids(&summary.failed), ["broken"]);
    assert_eq!(ids(&summary.cancelled), ["after"]);
    assert!(commands(&executor).is_empty());
}

/// An `Executor` that cannot spawn a command at all — a missing shell, a
/// `cwd` that does not exist. Scriptable behaviour `FakeExecutor` does not
/// offer, so this test drives the case directly through the trait.
struct SpawnFailureExecutor;

#[async_trait::async_trait]
impl Executor for SpawnFailureExecutor {
    async fn open(&self, _beam: BeamContext) -> Result<Box<dyn ExecSession>, ExecError> {
        Ok(Box::new(SpawnFailureSession))
    }
}

struct SpawnFailureSession;

#[async_trait::async_trait]
impl ExecSession for SpawnFailureSession {
    async fn execute(
        &mut self,
        cmd: CommandSpec,
        _ctx: ExecContext,
    ) -> Result<ExecResult, ExecError> {
        if cmd.command.contains("broken") {
            return Err(ExecError {
                message: format!("failed to spawn `{}`", cmd.command),
            });
        }
        Ok(ExecResult { exit_code: 0 })
    }

    async fn close(self: Box<Self>) -> Result<(), ExecError> {
        Ok(())
    }

    async fn kill(self: Box<Self>) {}
}

/// A command that never produced an exit code is that beam's failure, not
/// the run's: it is reported like any other failure, its message reaches
/// the user as stderr output, and `--keep-going` still runs what does not
/// depend on it.
#[tokio::test]
async fn a_command_that_cannot_be_spawned_fails_only_its_own_beam() {
    const SOURCE: &str = r#"
beam broken { run "broken command" }
beam other { run "step other" }
beam all { needs [broken, other] run "step all" }
"#;

    let outcome = run_target(
        SOURCE,
        "all",
        options(2, true),
        Arc::new(SpawnFailureExecutor),
    )
    .await;

    let summary = outcome.summary();
    assert_eq!(ids(&summary.failed), ["broken"]);
    assert_eq!(
        ids(&summary.succeeded),
        ["other"],
        "a beam that cannot spawn must not stop the rest of the run"
    );
    assert_eq!(ids(&summary.cancelled), ["all"]);
    assert_eq!(summary.exit_code(), 1);

    let events = &outcome.events;
    let reported = only(
        positions(events, |event| {
            matches!(event, RunEvent::BeamOutput { id, line, .. }
                if id.0 == "broken"
                    && line.stream == Stream::Stderr
                    && line.text.contains("failed to spawn `broken command`"))
        }),
        "spawn failure reported as stderr output",
    );
    let finished = only(
        positions(
            events,
            |event| matches!(event, RunEvent::BeamFinished { id, .. } if id.0 == "broken"),
        ),
        "`broken` finished event",
    );
    assert!(
        reported < finished,
        "the failure message must reach the user before the beam finishes"
    );
}

/// Commands run in their Beamfile's directory, which `cwd` overrides —
/// relatively to that directory, or absolutely.
#[tokio::test]
async fn commands_run_in_the_beams_directory_unless_cwd_overrides_it() {
    // `load_str` has no file on disk, so every beam's directory is `.`.
    #[cfg(windows)]
    const ABSOLUTE: &str = "C:/opt/alba";
    #[cfg(not(windows))]
    const ABSOLUTE: &str = "/opt/alba";

    let source = format!(
        r#"
beam plain {{ run "step plain" }}
beam relative {{ cwd "sub/dir" run "step relative" }}
beam absolute {{ cwd "{ABSOLUTE}" run "step absolute" }}
beam all {{ needs [plain, relative, absolute] run "step all" }}
"#
    );

    let executor = Arc::new(FakeExecutor::new());
    let outcome = run_target(&source, "all", options(4, false), executor.clone()).await;
    assert_eq!(outcome.summary().succeeded.len(), 4);

    let cwd_of = |command: &str| -> PathBuf {
        executor
            .calls()
            .into_iter()
            .find(|call| call.command == command)
            .unwrap_or_else(|| panic!("`{command}` was never executed"))
            .cwd
    };
    assert_eq!(cwd_of("step plain"), Path::new("."));
    assert_eq!(cwd_of("step relative"), Path::new(".").join("sub/dir"));
    assert_eq!(
        cwd_of("step absolute"),
        Path::new(ABSOLUTE),
        "an absolute `cwd` replaces the Beamfile's directory"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cancellation_token_cancels_everything() {
    const SOURCE: &str = r#"
beam a { run "slow a" }
beam b { needs [a] run "slow b" }
"#;

    let executor = Arc::new(FakeExecutor::new().on("slow", behavior(0, 30_000)));
    let cancel = CancellationToken::new();

    let trigger = cancel.clone();
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(10)).await;
        trigger.cancel();
    });

    let started = Instant::now();
    let outcome =
        run_target_with_cancel(SOURCE, "b", options(2, false), executor.clone(), cancel).await;
    let elapsed = started.elapsed();

    assert!(
        elapsed < Duration::from_secs(10),
        "a cancelled run must return promptly, took {elapsed:?}"
    );

    let summary = outcome.summary();
    assert_eq!(ids(&summary.cancelled), ["a", "b"]);
    assert!(summary.failed.is_empty(), "cancellation is not a failure");
}

/// Per beam, `BeamStarted` precedes every `BeamOutput`, which precedes
/// `BeamFinished`; `RunFinished` is the very last event. `a` runs two
/// commands to pin the rule that a multi-command `run` still emits exactly
/// one started/finished pair.
#[tokio::test]
async fn events_are_emitted_in_order_per_beam() {
    const SOURCE: &str = r#"
beam a { run ["step a1", "step a2"] }
beam b { needs [a] run "step b" }
"#;

    let executor = Arc::new(
        FakeExecutor::new()
            .on("a1", {
                let mut behavior = behavior(0, 0);
                behavior.output_lines = vec![stdout("a1 out")];
                behavior
            })
            .on("a2", {
                let mut behavior = behavior(0, 0);
                behavior.output_lines = vec![stdout("a2 out")];
                behavior
            })
            .on("step b", {
                let mut behavior = behavior(0, 0);
                behavior.output_lines = vec![stdout("b out")];
                behavior
            }),
    );
    let outcome = run_target(SOURCE, "b", options(2, false), executor.clone()).await;
    let events = &outcome.events;

    for beam in ["a", "b"] {
        let started = only(
            positions(
                events,
                |e| matches!(e, RunEvent::BeamStarted { id } if id.0 == beam),
            ),
            &format!("`{beam}` started event"),
        );
        let finished = positions(
            events,
            |e| matches!(e, RunEvent::BeamFinished { id, status, .. } if id.0 == beam && *status == BeamStatus::Succeeded),
        );
        let finished = only(finished, &format!("`{beam}` finished event"));
        let output = positions(
            events,
            |e| matches!(e, RunEvent::BeamOutput { id, .. } if id.0 == beam),
        );
        assert!(!output.is_empty(), "`{beam}` emitted no output");
        for index in output {
            assert!(
                started < index && index < finished,
                "`{beam}`: started {started}, output {index}, finished {finished}"
            );
        }
    }

    let a_finished = only(
        positions(
            events,
            |e| matches!(e, RunEvent::BeamFinished { id, .. } if id.0 == "a"),
        ),
        "`a` finished event",
    );
    let b_started = only(
        positions(
            events,
            |e| matches!(e, RunEvent::BeamStarted { id } if id.0 == "b"),
        ),
        "`b` started event",
    );
    assert!(a_finished < b_started, "`b` starts after `a` finished");

    let run_finished = only(
        positions(events, |e| matches!(e, RunEvent::RunFinished { .. })),
        "run finished event",
    );
    assert_eq!(
        run_finished,
        events.len() - 1,
        "`RunFinished` must be the last event"
    );
    let RunEvent::RunFinished { summary } = &events[run_finished] else {
        unreachable!()
    };
    assert_eq!(ids(&summary.succeeded), ["a", "b"]);
    assert_eq!(ids(&outcome.summary().succeeded), ["a", "b"]);
}

/// A docker beam dispatches to the `docker` executor slot, never to
/// `embedded` or `system`, and its rendered configuration reaches the
/// session as `BeamContext.options`.
#[tokio::test]
async fn a_docker_beam_dispatches_to_the_docker_slot_with_its_options() {
    const SOURCE: &str = r#"
beam ship {
  executor docker { image "alpine:3" workdir "/w" }
  run "deploy"
}
"#;

    let fake_a = Arc::new(FakeExecutor::new());
    let fake_b = Arc::new(FakeExecutor::new());
    let outcome = run_target_with_executors(
        SOURCE,
        "ship",
        options(2, false),
        Executors {
            embedded: fake_a.clone(),
            system: fake_a.clone(),
            docker: fake_b.clone(),
        },
        CancellationToken::new(),
    )
    .await;

    assert_eq!(ids(&outcome.summary().succeeded), ["ship"]);
    assert_eq!(
        fake_b.events(),
        vec![
            FakeEvent::Opened {
                beam: "ship".to_string(),
                options: serde_json::json!({"image": "alpine:3", "volumes": [], "workdir": "/w"}),
            },
            FakeEvent::Executed {
                command: "deploy".to_string(),
            },
            FakeEvent::Closed {
                beam: "ship".to_string(),
            },
        ]
    );
    assert!(
        fake_a.events().is_empty(),
        "neither the embedded nor the system slot must see this beam"
    );
}

#[tokio::test]
async fn target_parameters_are_bound_into_commands_and_env() {
    const SOURCE: &str = r#"
beam deploy(target) {
  env { DEPLOY_TARGET = target }
  run "deploy {target}"
}
"#;

    let executor = Arc::new(FakeExecutor::new());
    let options = RunOptions {
        jobs: 2,
        keep_going: false,
        params: vec!["staging".to_string()],
        cache: None,
        extra_env: Vec::new(),
    };
    let outcome = run_target(SOURCE, "deploy", options, executor.clone()).await;

    assert_eq!(ids(&outcome.summary().succeeded), ["deploy"]);
    let calls = executor.calls();
    assert_eq!(calls[0].command, "deploy staging");
    assert_eq!(
        calls[0].env,
        [("DEPLOY_TARGET".to_string(), "staging".to_string())]
    );
}

#[tokio::test]
async fn parameter_count_must_match_the_target() {
    const NO_PARAMS: &str = r#"beam build { run "step build" }"#;
    const ONE_PARAM: &str = r#"beam deploy(target) { run "deploy {target}" }"#;

    let too_many = run_target(
        NO_PARAMS,
        "build",
        RunOptions {
            jobs: 1,
            keep_going: false,
            params: vec!["extra".to_string()],
            cache: None,
            extra_env: Vec::new(),
        },
        Arc::new(FakeExecutor::new()),
    )
    .await;
    assert_eq!(
        too_many.error().to_string(),
        "beam `build` takes no parameters, got 1"
    );

    let missing = run_target(
        ONE_PARAM,
        "deploy",
        options(1, false),
        Arc::new(FakeExecutor::new()),
    )
    .await;
    assert_eq!(
        missing.error().to_string(),
        "beam `deploy` expects 1 parameter (`target`), got 0"
    );
}

/// A beam declaring `executor system_shell` dispatches to `Executors::system`
/// while every other beam keeps using `Executors::embedded` — the two slots
/// are genuinely independent, not just two names for one executor.
#[tokio::test]
async fn a_system_shell_beam_uses_the_system_executor() {
    const SOURCE: &str = r#"
beam a { run "step a" }
beam b { executor system_shell run "step b" }
beam all { needs [a, b] run "step all" }
"#;

    let embedded = Arc::new(FakeExecutor::new());
    let system = Arc::new(FakeExecutor::new());
    let executors = Executors {
        embedded: embedded.clone(),
        system: system.clone(),
        docker: Arc::new(FakeExecutor::new()),
    };

    let outcome = run_target_with_executors(
        SOURCE,
        "all",
        options(2, false),
        executors,
        CancellationToken::new(),
    )
    .await;

    assert_eq!(ids(&outcome.summary().succeeded), ["a", "b", "all"]);
    assert_eq!(
        commands(&embedded),
        ["step a", "step all"],
        "the default-executor beams must run on the embedded slot only"
    );
    assert_eq!(
        commands(&system),
        ["step b"],
        "the system-shell beam must run on the system slot only"
    );
}

/// `executor system_shell` is a real, implemented executor kind, exactly
/// like `executor docker` — neither trips the unschedulability gate `run`
/// applies before starting any beam, which only ever rejects an
/// unresolvable plugin executor or a parameter mismatch (see the module
/// doc comment's "Shape" section).
#[tokio::test]
async fn a_system_shell_beam_passes_validation() {
    const SOURCE: &str = r#"
beam build { executor system_shell run "step build" }
"#;

    let executor = Arc::new(FakeExecutor::new());
    let outcome = run_target(SOURCE, "build", options(2, false), executor.clone()).await;

    assert_eq!(ids(&outcome.summary().succeeded), ["build"]);
    assert_eq!(commands(&executor), ["step build"]);
}

/// Every run announces its plan before any beam starts: the TUI's tree
/// and graph are built from this snapshot, and a reloadable Beamfile
/// means the graph is not known once and for all.
#[tokio::test]
async fn a_run_announces_its_plan_before_any_beam_starts() {
    const SOURCE: &str = r#"
beam codegen { run "step codegen" }
beam build { needs [codegen] run "step build" }
"#;

    let executor = Arc::new(FakeExecutor::new());
    let outcome = run_target(SOURCE, "build", options(2, false), executor.clone()).await;
    assert!(outcome.summary().failed.is_empty(), "precondition");

    let first = outcome.events.first().expect("at least one event");
    match first {
        RunEvent::RunStarted {
            targets,
            affected_by,
            beams,
            edges,
        } => {
            assert_eq!(ids(targets), ["build"]);
            assert!(affected_by.is_none());
            assert_eq!(ids(beams), ["codegen", "build"]);
            assert_eq!(edges.len(), 1);
            assert_eq!(
                (edges[0].0.0.as_str(), edges[0].1.0.as_str()),
                ("build", "codegen")
            );
        }
        other => panic!("first event must be RunStarted, got {other:?}"),
    }
}

/// The helper for a run with several targets: the union of their
/// subgraphs, each beam scheduled once.
async fn run_targets(source: &str, targets: &[&str], executor: Arc<dyn Executor>) -> Outcome {
    let project = load_str(source).unwrap();
    let targets = Targets {
        beams: targets.iter().map(|t| BeamId(t.to_string())).collect(),
        affected_by: Some("main".to_string()),
    };
    let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
    let result = run(
        &project,
        &targets,
        options(4, false),
        Executors::uniform(executor),
        sender,
        CancellationToken::new(),
    )
    .await;
    let mut events = Vec::new();
    while let Ok(event) = receiver.try_recv() {
        events.push(event);
    }
    Outcome { result, events }
}

#[tokio::test]
async fn several_targets_schedule_the_union_of_their_subgraphs_once() {
    const SOURCE: &str = r#"
beam shared { run "s" }
beam a { needs [shared] run "a" }
beam b { needs [shared] run "b" }
beam c { run "c" }
"#;
    let executor = Arc::new(FakeExecutor::new());
    let outcome = run_targets(SOURCE, &["a", "b"], executor).await;
    let summary = outcome.summary();
    let mut ran = ids(&summary.succeeded);
    ran.sort();
    assert_eq!(ran, ["a", "b", "shared"]);
    match &outcome.events[0] {
        RunEvent::RunStarted {
            targets,
            affected_by,
            ..
        } => {
            assert_eq!(ids(targets), ["a", "b"]);
            assert_eq!(affected_by.as_deref(), Some("main"));
        }
        other => panic!("first event must be RunStarted, got {other:?}"),
    }
}

#[tokio::test]
async fn no_targets_is_an_empty_run() {
    let executor = Arc::new(FakeExecutor::new());
    let outcome = run_targets("beam a { run \"a\" }", &[], executor).await;
    let summary = outcome.summary();
    assert!(summary.succeeded.is_empty() && summary.failed.is_empty());
    assert!(matches!(&outcome.events[0], RunEvent::RunStarted { beams, .. } if beams.is_empty()));
    assert!(matches!(
        outcome.events.last(),
        Some(RunEvent::RunFinished { .. })
    ));
}

#[tokio::test]
async fn positional_arguments_need_exactly_one_target() {
    const SOURCE: &str = r#"
beam a(x) { run "a {x}" }
beam b { run "b" }
"#;
    let project = load_str(SOURCE).unwrap();
    let targets = Targets {
        beams: vec![BeamId("a".into()), BeamId("b".into())],
        affected_by: None,
    };
    let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
    let mut options = options(1, false);
    options.params = vec!["v".to_string()];
    let err = run(
        &project,
        &targets,
        options,
        Executors::uniform(Arc::new(FakeExecutor::new())),
        sender,
        CancellationToken::new(),
    )
    .await
    .unwrap_err();
    assert!(err.to_string().contains("exactly one target"), "{err}");
}

#[tokio::test]
async fn only_the_target_beam_may_declare_parameters() {
    const SOURCE: &str = r#"
beam helper(name) { run "help {name}" }
beam build { needs [helper] run "step build" }
"#;

    let executor = Arc::new(FakeExecutor::new());
    let outcome = run_target(SOURCE, "build", options(2, false), executor.clone()).await;

    assert_eq!(
        outcome.error().to_string(),
        "beam `helper` declares parameters, but only a target beam can be given any"
    );
    assert!(commands(&executor).is_empty());
}

/// A plugin binary missing from the `PATH` is caught during planning, the
/// same as any other unschedulable run: nothing is spawned, and the fake
/// executor — which would stand in for every built-in kind — never sees a
/// single event.
#[tokio::test]
async fn a_run_with_an_unresolvable_plugin_fails_before_starting_anything() {
    const SOURCE: &str = r#"
beam d { executor notinstalled run "x" }
"#;

    let executor = Arc::new(FakeExecutor::new());
    let outcome = run_target(SOURCE, "d", options(2, false), executor.clone()).await;

    assert!(matches!(outcome.error(), EngineError::Unschedulable(_)));
    assert!(executor.events().is_empty());
}

/// A beam's single session brackets every one of its commands: `open` runs
/// once before the first command, `execute` runs once per command, and
/// `close` runs once after the last — never interleaved with another
/// beam's session, since this beam declares no dependents to race against.
#[tokio::test]
async fn a_beam_opens_one_session_runs_its_commands_in_it_and_closes_it() {
    const SOURCE: &str = r#"
beam build { run ["echo a", "echo b"] }
"#;

    let executor = Arc::new(FakeExecutor::new());
    let outcome = run_target(SOURCE, "build", options(2, false), executor.clone()).await;

    assert_eq!(ids(&outcome.summary().succeeded), ["build"]);
    assert_eq!(
        executor.events(),
        vec![
            FakeEvent::Opened {
                beam: "build".to_string(),
                options: serde_json::Value::Null,
            },
            FakeEvent::Executed {
                command: "echo a".to_string(),
            },
            FakeEvent::Executed {
                command: "echo b".to_string(),
            },
            FakeEvent::Closed {
                beam: "build".to_string(),
            },
        ]
    );
}

/// `close` is guaranteed cleanup, not a happy-path courtesy: a beam whose
/// first command fails must still close its session, and the commands after
/// the failing one must never reach it.
#[tokio::test]
async fn the_session_closes_even_when_a_command_fails() {
    const SOURCE: &str = r#"
beam build { run ["boom", "echo never"] }
"#;

    let executor = Arc::new(FakeExecutor::new().on("boom", behavior(1, 0)));
    let outcome = run_target(SOURCE, "build", options(2, false), executor.clone()).await;

    assert_eq!(ids(&outcome.summary().failed), ["build"]);
    let events = executor.events();
    assert!(
        matches!(events.last(), Some(FakeEvent::Closed { beam }) if beam == "build"),
        "the session must still close after the failing command: {events:?}"
    );
    assert!(
        !events.iter().any(
            |event| matches!(event, FakeEvent::Executed { command } if command == "echo never")
        ),
        "the command after the failing one must never run: {events:?}"
    );
}

/// The same guarantee under cancellation: a session opened for a beam whose
/// only command is cancelled mid-flight must still close.
#[tokio::test]
async fn the_session_closes_when_the_run_is_cancelled_mid_command() {
    const SOURCE: &str = r#"
beam build { run "slow" }
"#;

    let executor = Arc::new(FakeExecutor::new().on("slow", behavior(0, 30_000)));
    let cancel = CancellationToken::new();

    let trigger = cancel.clone();
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(50)).await;
        trigger.cancel();
    });

    let outcome =
        run_target_with_cancel(SOURCE, "build", options(2, false), executor.clone(), cancel).await;

    assert_eq!(ids(&outcome.summary().cancelled), ["build"]);
    let events = executor.events();
    assert!(
        events
            .iter()
            .any(|event| matches!(event, FakeEvent::Closed { beam } if beam == "build")),
        "the session must still close after cancellation: {events:?}"
    );
}

/// A minimal `Executor`, distinct from `FakeExecutor`, whose `execute`
/// ignores cancellation entirely and always reports exit code 0 —
/// modelling a plugin that answers a cancelled command with
/// `{"type":"exit","code":0}`, a convention `docs/plugin-protocol.md`
/// explicitly leaves open (only the reference plugin's own `-1` is
/// documented, not required). `FakeSession` cannot model this: its own
/// cancellation race always reports `-1`, which is exactly the code this
/// scenario must NOT depend on.
struct IgnoresCancelExecutor {
    ran: Arc<Mutex<Vec<String>>>,
}

#[async_trait::async_trait]
impl Executor for IgnoresCancelExecutor {
    async fn open(&self, _beam: BeamContext) -> Result<Box<dyn ExecSession>, ExecError> {
        Ok(Box::new(IgnoresCancelSession {
            ran: Arc::clone(&self.ran),
        }))
    }
}

struct IgnoresCancelSession {
    ran: Arc<Mutex<Vec<String>>>,
}

#[async_trait::async_trait]
impl ExecSession for IgnoresCancelSession {
    async fn execute(
        &mut self,
        cmd: CommandSpec,
        _ctx: ExecContext,
    ) -> Result<ExecResult, ExecError> {
        self.ran.lock().unwrap().push(cmd.command);
        // Long enough that the test's cancellation trigger (10ms) fires
        // while this is still running, short enough to keep the test
        // fast — and never raced against `_ctx.cancel`, which is the
        // whole point: this command finishes and reports success
        // regardless of cancellation.
        tokio::time::sleep(Duration::from_millis(30)).await;
        Ok(ExecResult { exit_code: 0 })
    }

    async fn close(self: Box<Self>) -> Result<(), ExecError> {
        Ok(())
    }

    async fn kill(self: Box<Self>) {}
}

/// The engine must not read a cancelled command's exit code as the only
/// signal for whether to keep going: a beam whose first command answers
/// `0` despite the run having been cancelled mid-flight must still stop
/// there, not dispatch its remaining commands one after another. Pins
/// Important finding 6 from the whole-branch review — the scheduler's
/// command loop only ever broke on a non-zero exit code, so a conformant
/// plugin using any convention besides the reference plugin's `-1` could
/// make a cancelled beam run every remaining command anyway, each one
/// opening only to be cancelled again.
#[tokio::test]
async fn cancellation_is_rechecked_between_commands_even_when_the_exit_code_is_zero() {
    const SOURCE: &str = r#"
beam build { run ["one", "two"] }
"#;

    let ran = Arc::new(Mutex::new(Vec::new()));
    let executor = Arc::new(IgnoresCancelExecutor {
        ran: Arc::clone(&ran),
    });

    let cancel = CancellationToken::new();
    let trigger = cancel.clone();
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(10)).await;
        trigger.cancel();
    });

    let outcome =
        run_target_with_cancel(SOURCE, "build", options(2, false), executor, cancel).await;

    assert_eq!(ids(&outcome.summary().cancelled), ["build"]);
    assert_eq!(
        *ran.lock().unwrap(),
        vec!["one"],
        "cancellation fired while `one` was still running, so `two` must \
         never run even though `one` reported exit code 0"
    );
}

/// An executor that cannot even open a session for this beam — no runtime,
/// a missing image — is that beam's own failure, reported exactly like a
/// command that could not be spawned: a stderr line carrying the message,
/// then `Failed { exit_code: -1 }`. No command ever reaches a session that
/// was never created, and there is nothing to close.
#[tokio::test]
async fn an_open_failure_is_the_beams_failure_with_the_message_on_stderr() {
    const SOURCE: &str = r#"
beam build { run "echo a" }
"#;

    let executor = Arc::new(FakeExecutor::new().fail_open("build", "no runtime here"));
    let outcome = run_target(SOURCE, "build", options(2, false), executor.clone()).await;

    let summary = outcome.summary();
    assert_eq!(ids(&summary.failed), ["build"]);
    assert_eq!(summary.exit_code(), 1);

    let events = &outcome.events;
    let finished = events
        .iter()
        .find_map(|event| match event {
            RunEvent::BeamFinished { id, status, .. } if id.0 == "build" => Some(status.clone()),
            _ => None,
        })
        .expect("`build` must finish");
    assert_eq!(finished, BeamStatus::Failed { exit_code: -1 });

    assert!(
        events.iter().any(
            |event| matches!(event, RunEvent::BeamOutput { id, line, .. }
            if id.0 == "build"
                && line.stream == Stream::Stderr
                && line.text.contains("no runtime here"))
        ),
        "the open failure's message must reach the user as stderr output: {events:?}"
    );

    let fake_events = executor.events();
    assert!(
        !fake_events
            .iter()
            .any(|event| matches!(event, FakeEvent::Executed { .. })),
        "no command may reach a session that was never opened: {fake_events:?}"
    );
    assert!(
        !fake_events
            .iter()
            .any(|event| matches!(event, FakeEvent::Closed { .. })),
        "there is no session to close: {fake_events:?}"
    );
}

/// `extra_env` is the session's ambient environment (the TUI forcing
/// colour): it reaches every command, and a beam's own `env` wins on a
/// name both set.
#[tokio::test]
async fn extra_env_reaches_the_executor_and_the_beam_env_wins() {
    let source = r#"
beam plain {
  run "echo plain"
}

beam own {
  env { FORCE_COLOR = "0" }
  needs [plain]
  run "echo own"
}
"#;
    let executor = Arc::new(FakeExecutor::new());
    let mut options = options(1, false);
    options.extra_env = vec![
        ("FORCE_COLOR".to_string(), "1".to_string()),
        ("CLICOLOR_FORCE".to_string(), "1".to_string()),
    ];
    let outcome = run_target(source, "own", options, executor.clone()).await;
    outcome.summary();

    let calls = executor.calls();
    let env_of = |command: &str| -> Vec<(String, String)> {
        calls
            .iter()
            .find(|call| call.command == command)
            .unwrap_or_else(|| panic!("no call for {command}"))
            .env
            .clone()
    };
    assert_eq!(
        env_of("echo plain"),
        vec![
            ("FORCE_COLOR".to_string(), "1".to_string()),
            ("CLICOLOR_FORCE".to_string(), "1".to_string()),
        ]
    );
    let own = env_of("echo own");
    assert_eq!(
        own.iter().filter(|(name, _)| name == "FORCE_COLOR").count(),
        1,
        "one entry per name, never a duplicate the executor would have to arbitrate: {own:?}"
    );
    assert!(
        own.contains(&("FORCE_COLOR".to_string(), "0".to_string())),
        "the beam's own value wins: {own:?}"
    );
    assert!(own.contains(&("CLICOLOR_FORCE".to_string(), "1".to_string())));
}