magi-cli 0.189.0

//! Parking at a node boundary, and resuming into the node it stopped before.
//!
//! This is the "stop everything, replace the binary, resume" cycle: an
//! operator with fixes to install cannot wait out a competition, and killing
//! the process loses whatever the seats in flight had not written.
mod common;

use common::{Judges, SeatBlock, fixture, home_lock};
use magi::graph::{Pause, Runner};
use magi::run::RunStatus;

common::e2e! {
async fn a_parked_run_keeps_its_work_and_resumes_into_the_next_node() {
    let _home = home_lock().await;
    let fx = fixture(_home, Judges::Unanimous, false);
    let pause = Pause::new();
    let mut runner = Runner::start(&fx.repo, "create note.txt".to_owned(), fx.config.clone(), magi::run::Origin::operator())
        .await
        .expect("start");
    runner.on_pause(pause.clone());

    // Asked before the walk begins, so it parks at the first boundary it
    // reaches: after `prep`, with the worktrees built and nothing implemented.
    pause.park();
    runner.execute().await.expect("execute parks cleanly");

    let id = runner.state.id.clone();
    assert!(runner.state.parked, "the run records that it parked");
    assert!(
        !runner.state.status.done(),
        "a park is not a terminal status: {:?}",
        runner.state.status
    );
    assert_eq!(
        runner.state.candidates.len(),
        3,
        "prep finished, so its worktrees are kept"
    );
    assert!(
        runner.state.judgements.is_empty(),
        "and the graph stopped before judging"
    );
    assert!(
        runner
            .state
            .events
            .iter()
            .any(|e| e.node == "park" && e.message.contains("resume to carry on")),
        "the timeline says why it is sitting there: {:?}",
        runner.state.events.last()
    );

    // Resuming with nobody asking for a park carries the run to the end. The
    // candidates prep built are the ones it competes: no second competition,
    // and no repeated prep.
    let mut resumed = Runner::resume(&id).expect("resume");
    resumed.execute().await.expect("execute to a verdict");

    assert_eq!(
        resumed.state.candidates.len(),
        3,
        "the same three candidates, not a fresh set"
    );
    assert!(
        matches!(
            resumed.state.status,
            RunStatus::Ready | RunStatus::Merged | RunStatus::Blocked
        ),
        "the run reached a terminal status: {:?}",
        resumed.state.status
    );
    assert!(
        !resumed.state.parked,
        "and it is no longer parked once it has been carried on"
    );
}
}

common::e2e! {
async fn a_resumed_run_drops_a_stale_active_marker_left_by_a_killed_process() {
    let _home = home_lock().await;
    let fx = fixture(_home, Judges::Unanimous, false);
    let pause = Pause::new();
    let mut runner = Runner::start(&fx.repo, "create note.txt".to_owned(), fx.config.clone(), magi::run::Origin::operator())
        .await
        .expect("start");
    runner.on_pause(pause.clone());
    pause.park();
    runner.execute().await.expect("execute parks cleanly");

    // A daemon `SIGKILL`ed mid-wave never gets to clear the seat it was
    // asking, so its last batch is exactly what a reload would see: an entry
    // in `active` with nobody left to answer it.
    runner.state.seat_started(
        "implement",
        "impl-B",
        std::time::Duration::from_secs(3600),
        0,
    );
    runner.state.save().expect("save");
    let id = runner.state.id.clone();
    drop(runner);

    let resumed = Runner::resume(&id).expect("resume");
    assert!(
        resumed.state.active.contains_key("impl-B"),
        "loading a run reflects exactly what was on disk, stale or not"
    );

    // `execute` must not read that leftover as "impl-B is still running" and
    // must not let it survive into the run it carries on: the very next thing
    // it does is drop it, before dispatching any wave of its own.
    let mut resumed = resumed;
    resumed.execute().await.expect("execute to a verdict");
    assert!(
        !resumed.state.active.contains_key("impl-B"),
        "a resumed run must not go on claiming a leftover seat is still \
         answering: {:?}",
        resumed.state.active
    );
}
}

common::e2e! {
/// Driver ownership lasts only while `execute` is driving the graph. Both
/// parking and completing a resumed run must release it in memory and on disk,
/// even when the resume started with a dead process's stale driver record.
async fn execute_releases_its_driver_after_parking_and_resuming() {
    fn assert_driver_released(state: &magi::run::RunState) {
        let saved = magi::run::RunState::load(&state.id).expect("load saved state");
        for state in [state, &saved] {
            assert_eq!(state.driver_pid, None);
            assert_eq!(state.driver_started_at, None);
            assert!(state.driver_exited);
        }
    }

    let _home = home_lock().await;
    let fx = fixture(_home, Judges::Unanimous, false);
    let pause = Pause::new();
    let mut runner = Runner::start(&fx.repo, "create note.txt".to_owned(), fx.config.clone(), magi::run::Origin::operator())
        .await
        .expect("start");
    runner.on_pause(pause.clone());
    pause.park();
    runner.execute().await.expect("execute parks cleanly");
    assert_driver_released(&runner.state);

    // Pretend an earlier, now-dead process drove this run — exactly what a
    // resume after a kill finds on disk.
    runner.state.driver_pid = Some(999_999_999);
    runner.state.driver_exited = false;
    runner.state.save().expect("save");
    let id = runner.state.id.clone();
    drop(runner);

    let mut resumed = Runner::resume(&id).expect("resume");
    resumed.execute().await.expect("execute to a verdict");
    assert_driver_released(&resumed.state);
}
}

common::e2e! {
async fn a_park_asked_for_mid_walk_stops_at_the_boundary_after_it() {
    let _home = home_lock().await;
    let fx = fixture(_home, Judges::Unanimous, false);
    let pause = Pause::new();
    let mut runner = Runner::start(&fx.repo, "create note.txt".to_owned(), fx.config.clone(), magi::run::Origin::operator())
        .await
        .expect("start");
    runner.on_pause(pause.clone());

    // A park that arrives while the graph is walking is honoured at the next
    // boundary, never mid-node: the node in progress finishes and writes what
    // it produced, which is the whole reason parking is cheap.
    let asker = tokio::spawn({
        let pause = pause.clone();
        async move {
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            pause.park();
        }
    });
    runner.execute().await.expect("execute");
    asker.await.expect("asker");

    assert!(runner.state.parked);
    assert!(
        !runner.state.candidates.is_empty(),
        "whatever node it was in finished and recorded its work"
    );
    assert!(
        !runner.state.status.done(),
        "and it stopped short of a verdict: {:?}",
        runner.state.status
    );
}
}

common::e2e! {
/// The property both prior attempts at the interrupt-scheduling feature were
/// rejected for missing: a park requested while an agent call is genuinely
/// in flight must not cut that call short, and must only be honoured once
/// the node it belongs to actually finishes.
///
/// `impl-A`'s mock process proves it has actually started (a real file it
/// writes, not a guessed sleep) before this test asks for a park; only then
/// is the park released, and only after `execute` returns is the run checked
/// for what happened. If a park were somehow observed mid-call, `impl-A`
/// would be missing from the candidates `implement` returns, or `judge`
/// would already have started — either failure this test would catch.
async fn a_park_requested_while_a_seat_is_mid_call_does_not_cut_it_short() {
    let _home = home_lock().await;
    let mut fx = fixture(_home, Judges::Unanimous, false);
    let block = SeatBlock::install(&mut fx, "impl-A", "implement");

    let pause = Pause::new();
    let mut runner = Runner::start(&fx.repo, "create note.txt".to_owned(), fx.config.clone(), magi::run::Origin::operator())
        .await
        .expect("start");
    runner.on_pause(pause.clone());

    let mut exec = Box::pin(runner.execute());
    block.started(&mut exec, || None).await;
    // The call is now genuinely in flight. Ask it to park.
    pause.park();
    // Nothing inside the call can observe that - only `execute`'s own
    // boundary check, after `implement` returns, can - so letting it finish
    // on its own is what proves the point rather than merely being lucky
    // about timing.
    block.release();
    exec.await.expect("execute parks after implement");

    assert_eq!(
        runner.state.candidates.len(),
        3,
        "all three candidates exist, including impl-A's own"
    );
    let a = runner
        .state
        .candidates
        .iter()
        .find(|c| c.label == 'A')
        .expect("candidate A");
    assert!(
        a.commits > 0 && a.failed.is_none(),
        "impl-A's work was not thrown away by the park request: {a:?}"
    );
    assert!(
        runner.state.judgements.is_empty(),
        "the park took effect at the boundary right after `implement`, \
         before `judge` ever started"
    );
    assert!(runner.state.parked);
}
}

#[test]
fn seat_block_reports_early_run_completion_even_when_marker_exists() {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("runtime");

    // 1. Run finishes before marker ever appears.
    let err = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        rt.block_on(async {
            let mut run = Box::pin(async { Err::<(), &str>("prep exploded") });
            common::seat_block::wait_while_running(
                &mut run,
                || false,
                || None,
                "impl-A never started",
            )
            .await;
        });
    }))
    .expect_err("must panic when run finishes before marker");
    let msg = err
        .downcast_ref::<String>()
        .map(String::as_str)
        .or_else(|| err.downcast_ref::<&str>().copied())
        .unwrap_or("");
    assert!(
        msg.contains("prep exploded"),
        "panic carries the run's outcome: {msg}"
    );

    // 2. Both the marker and the `ended` condition are present on the same tick:
    // termination takes precedence over the marker.
    let err = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        rt.block_on(async {
            let mut run = Box::pin(std::future::pending::<()>());
            common::seat_block::wait_while_running(
                &mut run,
                || true,
                || Some("task settled as Held".to_owned()),
                "impl-A never started",
            )
            .await;
        });
    }))
    .expect_err("must panic when ended and ready are observed together");
    let msg = err
        .downcast_ref::<String>()
        .map(String::as_str)
        .or_else(|| err.downcast_ref::<&str>().copied())
        .unwrap_or("");
    assert!(
        msg.contains("task settled as Held"),
        "panic carries the ended diagnostic: {msg}"
    );
}

common::e2e! {
async fn seat_block_drop_without_release_fails_the_blocked_mock_seat() {
    let _home = home_lock().await;
    let mut fx = fixture(_home, Judges::Unanimous, false);
    let block = SeatBlock::install(&mut fx, "impl-A", "implement");
    let started_path = block.started_path().to_path_buf();
    let abandon_path = block.abandon_path().to_path_buf();

    let prompt_file = fx.tmp.path().join("prompt.md");
    std::fs::write(&prompt_file, "implement note.txt\n").expect("write prompt");
    let agent = &fx.config.agents[0];
    let mut cmd = tokio::process::Command::new(&agent.command[0]);
    cmd.args(&agent.command[1..])
        .current_dir(&fx.repo)
        .envs(&agent.env)
        .env("MAGI_PROMPT_FILE", &prompt_file)
        .env("MAGI_RUN", "20261023-000000-test")
        .env("MAGI_NODE", "implement")
        .env("MAGI_SEAT", "impl-A")
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped());
    let child = cmd.spawn().expect("spawn mock");
    let mut wait = Box::pin(child.wait_with_output());

    block.started(&mut wait, || None).await;
    assert!(started_path.exists());

    // Dropping without calling `release()` writes `abandon-impl-A`, which makes
    // the blocked mock exit non-zero rather than silently continuing.
    drop(block);
    assert!(abandon_path.exists());

    let out = wait.await.expect("mock output");
    assert!(
        !out.status.success(),
        "abandoned mock block must fail, not continue silently"
    );
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("mock: block on impl-A abandoned without release"),
        "unexpected stderr: {stderr}"
    );
}
}