sloop-daemon 0.5.0

mod support;

use std::fs;
use std::os::unix::fs::PermissionsExt;
use std::os::unix::process::CommandExt;
use std::path::PathBuf;
use std::process::{Child, Command};
use std::time::Duration;

use support::{World, process_alive, revert_trigger_rename, wait_until, wait_until_slow};

/// Writes a scripted fake agent and a repository config pointing at it, with
/// an optional flow test command. The agent script is committed before
/// the daemon starts so worktrees branch from a clean default branch.
fn configure(world: &World, agent_body: &str, test_cmd: Option<&str>) {
    let script = world.root().join("fake-agent.sh");
    fs::write(&script, format!("#!/bin/sh\n{agent_body}")).expect("write fake agent script");
    fs::create_dir_all(world.root().join(".agents/sloop/flows")).expect("create flow directory");
    fs::write(
        world.root().join(".agents/sloop/flows/default.yaml"),
        "stages:\n  - { name: build, action: agent }\n  - { name: merge, action: { builtin: merge } }\n",
    )
    .expect("write default test flow");

    let flow_test = match test_cmd {
        Some(cmd) => format!("flow:\n  test_cmd: [\"sh\", \"-c\", \"{cmd}\"]\n"),
        None => String::new(),
    };
    fs::write(
        world.root().join(".agents/sloop/config.yaml"),
        format!(
            "version: 1\nscheduler:\n  max_parallel_tasks: 1\nagent:\n  default_target: fake\n  targets:\n    fake:\n      cmd: [\"sh\", \"{}\", \"{{prompt}}\"]\n{flow_test}",
            script.display()
        ),
    )
    .expect("write agent config");
}

/// An agent that commits one file, as a well-behaved worker would.
const COMMITTING_AGENT: &str = concat!(
    "echo done > work.txt\n",
    "git add work.txt\n",
    "git -c user.name=agent -c user.email=agent@example.invalid commit --quiet -m 'agent work'\n",
    "exit 0\n"
);

fn post_and_run(world: &World, name: &str) -> String {
    let id = post_manual(world, name);
    assert!(world.sloop(&["run", &id]).status.success());
    id
}

fn post_manual(world: &World, name: &str) -> String {
    let ticket = world.write_ticket(name, "# Work\n");
    let output = world.sloop(&["post", ticket.to_str().unwrap(), "--manual"]);
    assert!(output.status.success());
    World::json_stdout(&output)["data"]["ticket"]["id"]
        .as_str()
        .expect("ticket id")
        .to_owned()
}

/// Failure envelopes are written to stderr; success envelopes to stdout.
fn json_stderr(output: &std::process::Output) -> serde_json::Value {
    serde_json::from_slice(&output.stderr).expect("stderr is JSON")
}

fn tickets(world: &World) -> serde_json::Value {
    let output = world.sloop(&["status"]);
    assert!(output.status.success());
    World::json_stdout(&output)["data"]["tickets"].clone()
}

fn write_flow(world: &World, contents: &str) {
    fs::write(
        world.root().join(".agents/sloop/flows/default.yaml"),
        contents,
    )
    .expect("write test flow");
}

/// The resolved stage verdicts a run recorded, in log order. Rows carrying no
/// verdict are an execution's action, answered for by the check row behind it.
fn stage_runs(world: &World, position: usize) -> Vec<(i64, String, String, String)> {
    let run_id = world.run_id(position);
    let connection = rusqlite::Connection::open(world.db_path()).expect("open state database");
    let mut statement = connection
        .prepare(
            "SELECT stage_index, stage, state, evidence_json
             FROM stage_runs
             WHERE run_id = ?1 AND state IS NOT NULL ORDER BY seq",
        )
        .expect("prepare stage query");
    statement
        .query_map([&run_id], |row| {
            let evidence: String = row.get(3)?;
            let evidence: serde_json::Value = serde_json::from_str(&evidence).unwrap();
            Ok((
                row.get(0)?,
                row.get(1)?,
                row.get(2)?,
                evidence["output"].as_str().unwrap_or_default().to_owned(),
            ))
        })
        .expect("query stages")
        .collect::<Result<Vec<_>, _>>()
        .expect("read stages")
}

/// The run's whole stage-evidence log as stored, keyed the way the walk reads
/// it: `(seq, stage_index, stage, attempt, phase, state)`.
fn stage_log(
    world: &World,
    position: usize,
) -> Vec<(i64, i64, String, i64, String, Option<String>)> {
    let run_id = world.run_id(position);
    let connection = rusqlite::Connection::open(world.db_path()).expect("open state database");
    let mut statement = connection
        .prepare(
            "SELECT seq, stage_index, stage, attempt, phase, state
             FROM stage_runs WHERE run_id = ?1 ORDER BY seq",
        )
        .expect("prepare stage log query");
    statement
        .query_map([&run_id], |row| {
            Ok((
                row.get(0)?,
                row.get(1)?,
                row.get(2)?,
                row.get(3)?,
                row.get(4)?,
                row.get(5)?,
            ))
        })
        .expect("query stage log")
        .collect::<Result<Vec<_>, _>>()
        .expect("read stage log")
}

fn stage_evidence(world: &World, position: usize, stage: &str) -> serde_json::Value {
    let run_id = world.run_id(position);
    let connection = rusqlite::Connection::open(world.db_path()).expect("open state database");
    let evidence: String = connection
        .query_row(
            "SELECT evidence_json FROM stage_runs
             WHERE run_id = ?1 AND stage = ?2 AND state IS NOT NULL
             ORDER BY seq DESC LIMIT 1",
            [run_id.as_str(), stage],
            |row| row.get(0),
        )
        .expect("read stage evidence");
    serde_json::from_str(&evidence).expect("stage evidence is JSON")
}

fn default_branch_has(world: &World, file: &str) -> bool {
    // The default branch is what the root checkout points at; a completed
    // fast-forward merge makes the agent's file visible there.
    world.root().join(file).is_file()
}

fn read_process_ids(path: PathBuf) -> Vec<u32> {
    fs::read_to_string(path)
        .expect("read stage process IDs")
        .split_whitespace()
        .map(|pid| pid.parse().expect("process ID is an integer"))
        .collect()
}

fn wait_for_processes_to_exit(process_ids: Vec<u32>) {
    for pid in process_ids {
        wait_until(&format!("stage process {pid} exits"), || {
            !process_alive(pid)
        });
    }
}

fn spawn_unrelated_process_group() -> Child {
    Command::new("sh")
        .args(["-c", "exec sleep 1000"])
        .process_group(0)
        .spawn()
        .expect("spawn unrelated process group")
}

struct InterruptedMergeFixture {
    started: PathBuf,
    invocations: PathBuf,
}

fn configure_blocking_merge(world: &World) -> InterruptedMergeFixture {
    configure(
        world,
        concat!(
            "echo agent > work.txt\n",
            "git add work.txt\n",
            "git -c user.name=agent -c user.email=agent@example.invalid commit --quiet -m agent\n",
            "root=\"$(dirname \"$0\")\"\n",
            "echo main > \"$root/main.txt\"\n",
            "git -C \"$root\" add main.txt\n",
            "git -C \"$root\" -c user.name=operator -c user.email=operator@example.invalid commit --quiet -m main\n",
        ),
        None,
    );
    let hook = world.root().join(".git/hooks/pre-merge-commit");
    let fixture = InterruptedMergeFixture {
        started: world.root().join("merge-hook-started"),
        invocations: world.root().join("merge-hook-invocations"),
    };
    let release = world.root().join("merge-hook-release");
    fs::write(
        &hook,
        format!(
            "#!/bin/sh\nprintf x >> {invocations}\ntouch {started}\nwhile [ ! -e {release} ]; do sleep 0.01; done\n",
            invocations = fixture.invocations.display(),
            started = fixture.started.display(),
            release = release.display(),
        ),
    )
    .unwrap();
    fs::set_permissions(&hook, fs::Permissions::from_mode(0o755)).unwrap();
    world.commit_all("initial");
    fixture
}

#[test]
fn committed_work_with_passing_tests_is_merged() {
    let world = World::configured();
    configure(&world, COMMITTING_AGENT, Some("echo tests passed"));
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "merge-me.md");

    wait_until("the ticket reaches merged", || {
        tickets(&world)["merged"] == 1
    });
    assert!(default_branch_has(&world, "work.txt"));

    // The implicit test stage's output is captured evidence.
    let output = world.sloop(&["logs", &world.run_id(1)]);
    assert!(output.status.success());
    let entries = World::json_stdout(&output)["data"]["entries"].clone();
    assert!(
        entries
            .as_array()
            .unwrap()
            .iter()
            .any(|entry| entry["source"] == "stage" && entry["stage"] == "test"),
        "no stage records in {entries}"
    );
}

#[test]
fn built_in_default_without_test_command_merges_committed_work() {
    let world = World::configured();
    let invocations = world.root().join("agent-invocations");
    let agent = format!(
        "printf x >> {}\n{}",
        invocations.display(),
        COMMITTING_AGENT
    );
    configure(&world, &agent, None);
    fs::remove_file(world.root().join(".agents/sloop/flows/default.yaml")).unwrap();
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "built-in-default.md");

    wait_until("the built-in default merges", || {
        tickets(&world)["merged"] == 1
    });
    assert_eq!(fs::read_to_string(invocations).unwrap(), "x");
    assert!(default_branch_has(&world, "work.txt"));
}

#[test]
fn flow_executes_in_order_and_records_one_row_per_stage() {
    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    let check = world.root().join("check-stage.sh");
    fs::write(
        &check,
        "#!/bin/sh\nset -eu\ntest -z \"${SLOOP_SOCKET:-}\"\ntest -z \"${SLOOP_TOKEN:-}\"\necho checked\n",
    )
    .unwrap();
    write_flow(
        &world,
        &format!(
            "stages:\n  - {{ name: build, action: agent }}\n  - name: check\n    action: {{ exec: [\"sh\", {}] }}\n  - {{ name: merge, action: {{ builtin: merge }} }}\n",
            serde_json::to_string(&check.to_string_lossy()).unwrap(),
        ),
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "flow-pass.md");

    wait_until("the flow merges", || tickets(&world)["merged"] == 1);
    assert!(default_branch_has(&world, "work.txt"));
    let stages = stage_runs(&world, 1);
    assert_eq!(
        stages
            .iter()
            .map(|(index, name, state, _)| (*index, name.as_str(), state.as_str()))
            .collect::<Vec<_>>(),
        [
            (0, "build", "passed"),
            (1, "check", "passed"),
            (2, "merge", "passed")
        ]
    );
    assert!(
        stages
            .iter()
            .all(|(_, _, _, output)| *output == format!("runs/{}/output.ndjson", world.run_id(1)))
    );
    // Each stage is judged without a second process, so each appends exactly
    // one action row, and `seq` numbers them in the order they executed.
    assert_eq!(
        stage_log(&world, 1),
        [
            (
                1,
                0,
                "build".into(),
                1,
                "action".into(),
                Some("passed".into())
            ),
            (
                2,
                1,
                "check".into(),
                1,
                "action".into(),
                Some("passed".into())
            ),
            (
                3,
                2,
                "merge".into(),
                1,
                "action".into(),
                Some("passed".into())
            ),
        ]
    );
    let logs = world.sloop(&["logs", &world.run_id(1)]);
    let entries = World::json_stdout(&logs)["data"]["entries"]
        .as_array()
        .unwrap()
        .clone();
    assert!(
        entries
            .iter()
            .any(|entry| entry["source"] == "stage" && entry["stage"] == "check")
    );
}

#[test]
fn failed_exec_halts_before_merge_and_preserves_commits_for_review() {
    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    write_flow(
        &world,
        "stages:\n  - { name: build, action: agent }\n  - { name: reject, action: { exec: ['false'] } }\n  - { name: merge, action: { builtin: merge } }\n",
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "flow-fail.md");

    wait_until("the flow needs review", || {
        tickets(&world)["needs_review"] == 1
    });
    assert!(!default_branch_has(&world, "work.txt"));
    assert_eq!(
        stage_runs(&world, 1)
            .into_iter()
            .map(|(_, name, state, _)| (name, state))
            .collect::<Vec<_>>(),
        [
            ("build".into(), "passed".into()),
            ("reject".into(), "failed".into())
        ]
    );
}

#[test]
fn check_verdict_uses_the_check_commands_exit() {
    let passing = World::configured();
    configure(&passing, COMMITTING_AGENT, None);
    write_flow(
        &passing,
        "stages:\n  - { name: build, action: agent }\n  - { name: verify, action: { exec: ['true'] }, result_check: { exec: ['true'] } }\n  - { name: merge, action: { builtin: merge } }\n",
    );
    passing.commit_all("initial");
    passing.start_daemon();
    post_and_run(&passing, "check-pass.md");
    wait_until("the passing check merges", || {
        tickets(&passing)["merged"] == 1
    });

    let failing = World::configured();
    configure(&failing, COMMITTING_AGENT, None);
    write_flow(
        &failing,
        "stages:\n  - { name: build, action: agent }\n  - { name: verify, action: { exec: ['true'] }, result_check: { exec: ['false'] } }\n  - { name: merge, action: { builtin: merge } }\n",
    );
    failing.commit_all("initial");
    failing.start_daemon();
    post_and_run(&failing, "check-fail.md");
    wait_until("the failing check needs review", || {
        tickets(&failing)["needs_review"] == 1
    });
    assert_eq!(
        stage_runs(&failing, 1)
            .into_iter()
            .map(|(_, name, state, _)| (name, state))
            .collect::<Vec<_>>(),
        [
            ("build".into(), "passed".into()),
            ("verify".into(), "failed".into())
        ]
    );
    // The action exited clean and the check is what rejected it, so the two
    // readings sit on separate rows and only the check's is the verdict.
    assert_eq!(
        stage_log(&failing, 1)
            .into_iter()
            .map(|(_, _, stage, _, phase, state)| (stage, phase, state))
            .collect::<Vec<_>>(),
        [
            ("build".into(), "action".into(), Some("passed".into())),
            ("verify".into(), "action".into(), None),
            ("verify".into(), "check".into(), Some("failed".into())),
        ]
    );
}

/// A stage judged by an independent actor has two things to account for, and
/// records both: the action's own exit, and the check's. The resolved verdict
/// rides on the check, because that is where the execution ended.
#[test]
fn a_check_actor_records_its_own_row_beside_the_action_it_judged() {
    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    write_flow(
        &world,
        "stages:\n  - { name: build, action: agent }\n  - { name: verify, action: { exec: ['sh', '-c', 'exit 0'] }, result_check: { exec: ['true'] } }\n  - { name: merge, action: { builtin: merge } }\n",
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "check-phases.md");
    wait_until("the checked flow merges", || tickets(&world)["merged"] == 1);

    assert_eq!(
        stage_log(&world, 1),
        [
            (
                1,
                0,
                "build".into(),
                1,
                "action".into(),
                Some("passed".into())
            ),
            (2, 1, "verify".into(), 1, "action".into(), None),
            (
                3,
                1,
                "verify".into(),
                1,
                "check".into(),
                Some("passed".into())
            ),
            (
                4,
                2,
                "merge".into(),
                1,
                "action".into(),
                Some("passed".into())
            ),
        ]
    );
    // Two rows for one stage, and `show` still reports the three stages the
    // flow declares — one line each, judged by the row the verdict rode on.
    let shown = world.show_snapshot(&world.run_id(1));
    assert_eq!(
        shown["stages"]
            .as_array()
            .expect("stages array")
            .iter()
            .map(|stage| {
                (
                    stage["stage"].as_str().unwrap_or_default().to_owned(),
                    stage["state"].as_str().unwrap_or_default().to_owned(),
                )
            })
            .collect::<Vec<_>>(),
        [
            ("build".to_owned(), "passed".to_owned()),
            ("verify".to_owned(), "passed".to_owned()),
            ("merge".to_owned(), "passed".to_owned()),
        ]
    );
}

#[test]
fn reported_stage_without_a_report_fails_with_a_reason() {
    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    write_flow(
        &world,
        "stages:\n  - { name: build, action: agent }\n  - { name: review, action: { exec: ['true'] }, result_check: reported }\n  - { name: merge, action: { builtin: merge } }\n",
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "missing-report.md");

    wait_until("the unreported stage needs review", || {
        tickets(&world)["needs_review"] == 1
    });
    let evidence = stage_evidence(&world, 1, "review");
    assert_eq!(evidence["verdict_source"], "reported");
    assert_eq!(evidence["reason"], "no verdict reported");
}

/// The shipped default flow's shape: a reported review stage decides the run.
/// A reported `fail` must hold the work back even though the reviewer command
/// itself exits 0, and a reported `pass` must let the merge through.
#[test]
fn a_reported_review_gates_the_merge_in_both_directions() {
    fn review_flow(world: &World, verdict: &str) {
        let reviewer = world.root().join("reviewer.sh");
        fs::write(
            &reviewer,
            format!(
                "#!/bin/sh\n\"{sloop}\" verdict {verdict} --reason 'scripted {verdict}'\nexit 0\n",
                sloop = env!("CARGO_BIN_EXE_sloop"),
            ),
        )
        .expect("write reviewer script");
        write_flow(
            world,
            &format!(
                "stages:\n  - {{ name: build, action: agent }}\n  - {{ name: review, action: {{ exec: ['sh', '{}'] }}, result_check: reported }}\n  - {{ name: merge, action: {{ builtin: merge }} }}\n",
                reviewer.display()
            ),
        );
    }

    let failing = World::configured();
    configure(&failing, COMMITTING_AGENT, None);
    review_flow(&failing, "fail");
    failing.commit_all("initial");
    failing.start_daemon();
    post_and_run(&failing, "review-fail.md");

    wait_until("the failed review needs review", || {
        tickets(&failing)["needs_review"] == 1
    });
    assert_eq!(tickets(&failing)["merged"], 0);
    assert!(!default_branch_has(&failing, "work.txt"));
    let evidence = stage_evidence(&failing, 1, "review");
    assert_eq!(evidence["verdict_source"], "reported");
    assert_eq!(evidence["reason"], "scripted fail");
    assert_eq!(
        stage_runs(&failing, 1)
            .into_iter()
            .map(|(_, name, state, _)| (name, state))
            .collect::<Vec<_>>(),
        [
            ("build".into(), "passed".into()),
            ("review".into(), "failed".into())
        ]
    );

    let passing = World::configured();
    configure(&passing, COMMITTING_AGENT, None);
    review_flow(&passing, "pass");
    passing.commit_all("initial");
    passing.start_daemon();
    post_and_run(&passing, "review-pass.md");

    wait_until("the passed review merges", || {
        tickets(&passing)["merged"] == 1
    });
    assert!(default_branch_has(&passing, "work.txt"));
    assert_eq!(
        stage_evidence(&passing, 1, "review")["reason"],
        "scripted pass"
    );
}

#[test]
fn unknown_bound_flow_never_falls_back_to_default() {
    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    world.commit_all("initial");
    world.start_daemon();
    let ticket = post_manual(&world, "missing-flow.md");
    let connection = rusqlite::Connection::open(world.db_path()).unwrap();
    connection
        .execute(
            "UPDATE tickets SET flow = 'missing' WHERE id = ?1",
            [&ticket],
        )
        .unwrap();

    assert!(world.sloop(&["run", &ticket]).status.success());
    wait_until("the invalid flow is rejected before claim", || {
        fs::read_to_string(world.daemon_log())
            .is_ok_and(|log| log.contains("bound_flow_resolution_failed"))
    });
    assert!(!world.run_worktree(1).exists());
    let runs: i64 = connection
        .query_row("SELECT COUNT(*) FROM runs", [], |row| row.get(0))
        .unwrap();
    assert_eq!(runs, 0);
    assert_eq!(tickets(&world)["ready"], 1);
}

#[test]
fn stage_evidence_write_failure_does_not_strand_active_accounting() {
    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    world.commit_all("initial");
    world.start_daemon();
    let ticket = post_manual(&world, "stage-write-failure.md");
    let connection = rusqlite::Connection::open(world.db_path()).unwrap();
    connection
        .execute_batch(
            "CREATE TRIGGER reject_stage_row
             BEFORE INSERT ON stage_runs
             BEGIN SELECT RAISE(FAIL, 'stage write denied'); END;",
        )
        .unwrap();

    assert!(world.sloop(&["run", &ticket]).status.success());
    wait_until("the evidence failure settles for review", || {
        tickets(&world)["needs_review"] == 1
    });
    let status = world.sloop(&["status"]);
    assert_eq!(
        World::json_stdout(&status)["data"]["gate"]["active_agents"],
        0
    );
}

#[test]
fn merge_success_survives_merge_stage_evidence_failure() {
    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    world.commit_all("initial");
    world.start_daemon();
    let ticket = post_manual(&world, "merge-evidence-failure.md");
    let connection = rusqlite::Connection::open(world.db_path()).unwrap();
    connection
        .execute_batch(
            "CREATE TRIGGER reject_merge_stage
             BEFORE INSERT ON stage_runs
             WHEN NEW.stage = 'merge'
             BEGIN SELECT RAISE(FAIL, 'merge stage write denied'); END;",
        )
        .unwrap();

    assert!(world.sloop(&["run", &ticket]).status.success());
    wait_until(
        "the successful merge settles despite evidence failure",
        || tickets(&world)["merged"] == 1,
    );
    assert!(default_branch_has(&world, "work.txt"));
    assert_eq!(
        stage_runs(&world, 1)
            .into_iter()
            .map(|(_, stage, _, _)| stage)
            .collect::<Vec<_>>(),
        ["build"]
    );
}

#[test]
fn incomplete_commit_observation_keeps_a_failed_later_stage_for_review() {
    let world = World::configured();
    configure(
        &world,
        "root=\"$(dirname \"$0\")\"\ngit reflog expire --expire=now --all\nexit 0\n",
        None,
    );
    write_flow(
        &world,
        "stages:\n  - { name: build, action: agent }\n  - { name: reject, action: { exec: ['false'] } }\n  - { name: merge, action: { builtin: merge } }\n",
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "unknown-commits.md");

    wait_until("unknown commit evidence is retained for review", || {
        tickets(&world)["needs_review"] == 1
    });
    assert_eq!(
        world
            .run_evidence(&world.run_id(1), "commits_observed")
            .unwrap()["complete"],
        false
    );
}

#[test]
fn exec_stage_order_comes_from_the_flow_file() {
    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    let order = world.root().join("stage-order");
    write_flow(
        &world,
        &format!(
            "stages:\n  - {{ name: build, action: agent }}\n  - {{ name: first, action: {{ exec: [\"sh\", \"-c\", \"printf first, >> {}\"] }} }}\n  - {{ name: second, action: {{ exec: [\"sh\", \"-c\", \"printf second >> {}\"] }} }}\n  - {{ name: merge, action: {{ builtin: merge }} }}\n",
            order.display(),
            order.display(),
        ),
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "ordered.md");

    wait_until("the ordered flow merges", || tickets(&world)["merged"] == 1);
    assert_eq!(fs::read_to_string(order).unwrap(), "first,second");
}

#[test]
fn restart_between_exec_stages_skips_the_completed_stage() {
    const HOOK: &str = "after-stage-first";

    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    world.arm_test_hook(HOOK);
    let invocations = world.root().join("flow-invocations");
    fs::write(
        world.root().join(".agents/sloop/flows/resume.yaml"),
        format!(
            "stages:\n  - {{ name: build, action: agent }}\n  - {{ name: first, action: {{ exec: [\"sh\", \"-c\", \"printf 1 >> {}\"] }} }}\n  - {{ name: second, action: {{ exec: [\"sh\", \"-c\", \"printf 2 >> {}\"] }} }}\n  - {{ name: merge, action: {{ builtin: merge }} }}\n",
            invocations.display(),
            invocations.display(),
        ),
    )
    .expect("write named recovery flow");
    world.commit_all("initial");
    let daemon_pid = world.start_daemon()["data"]["pid"].as_u64().unwrap() as u32;
    let ticket_path = world.write_ticket("resume-flow.md", "---\nflow: resume\n---\n# Work\n");
    let output = world.sloop(&["post", ticket_path.to_str().unwrap(), "--manual"]);
    assert!(output.status.success());
    let ticket = World::json_stdout(&output)["data"]["ticket"]["id"]
        .as_str()
        .unwrap()
        .to_owned();
    assert!(world.sloop(&["run", &ticket]).status.success());
    wait_until("the first stage is durably complete", || {
        world.test_hook_reached(HOOK)
            && fs::read_to_string(&invocations).is_ok_and(|value| value == "1")
    });

    world.kill_daemon(daemon_pid);
    fs::remove_file(world.root().join(".agents/sloop/flows/resume.yaml"))
        .expect("remove admitted flow before restart");
    world.start_daemon();
    wait_until("recovery completes the remaining flow", || {
        tickets(&world)["merged"] == 1
    });
    assert_eq!(fs::read_to_string(invocations).unwrap(), "12");
    assert_eq!(stage_runs(&world, 1).len(), 4);
    // Recovery folded the log and stood where it left off: four executions in
    // the order they ran, no stage repeated.
    assert_eq!(
        stage_log(&world, 1)
            .into_iter()
            .map(|(seq, index, stage, _, _, _)| (seq, index, stage))
            .collect::<Vec<_>>(),
        [
            (1, 0, "build".into()),
            (2, 1, "first".into()),
            (3, 2, "second".into()),
            (4, 3, "merge".into()),
        ]
    );
}

/// Rewrites the run into the shape it had three schema versions ago: stage rows
/// with no `seq` and no `phase` in a table named after the regime that wrote
/// them, a run state named after that regime, its process checkpoint under the
/// old evidence kind, and triggers still spelled `activations`. Putting the
/// schema version back makes the next open a genuine pre-migration database.
fn plant_old_stage_shape(world: &World) {
    revert_trigger_rename(world);
    let connection = rusqlite::Connection::open(world.db_path()).expect("open state database");
    connection
        .execute_batch(
            "CREATE TABLE aftercare_stages (
                 run_id          TEXT NOT NULL REFERENCES runs(id),
                 stage_index     INTEGER NOT NULL,
                 stage           TEXT NOT NULL,
                 state           TEXT NOT NULL,
                 attempt         INTEGER NOT NULL DEFAULT 1,
                 started_at_ms   INTEGER,
                 finished_at_ms  INTEGER,
                 exit_code       INTEGER,
                 evidence_json   TEXT,
                 PRIMARY KEY (run_id, stage_index, attempt)
             );
             INSERT INTO aftercare_stages
                 (run_id, stage_index, stage, state, attempt, started_at_ms,
                  finished_at_ms, exit_code, evidence_json)
             SELECT run_id, stage_index, stage, state, attempt, started_at_ms,
                    finished_at_ms, exit_code, evidence_json
             FROM stage_runs WHERE state IS NOT NULL;
             DROP TABLE stage_runs;
             UPDATE runs SET state = 'aftercare' WHERE state = 'driving';
             UPDATE run_evidence
                SET kind = 'aftercare_process',
                    dedupe_key = 'settlement:' || run_id || ':aftercare_process'
              WHERE kind = 'stage_process';
             PRAGMA user_version = 13;",
        )
        .expect("plant pre-log stage rows");
}

/// A run queued before the stage table became a log must recover to the same
/// `Step` afterwards. The migration numbers the rows it finds in
/// `(stage_index, attempt)` order — the order the linear walk wrote them — so
/// the fold that replays them stands exactly where it stood before.
#[test]
fn a_run_planted_in_the_old_stage_shape_migrates_and_resumes_where_it_stood() {
    const HOOK: &str = "after-stage-first";

    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    world.arm_test_hook(HOOK);
    let invocations = world.root().join("flow-invocations");
    fs::write(
        world.root().join(".agents/sloop/flows/resume.yaml"),
        format!(
            "stages:\n  - {{ name: build, action: agent }}\n  - {{ name: first, action: {{ exec: [\"sh\", \"-c\", \"printf 1 >> {}\"] }} }}\n  - {{ name: second, action: {{ exec: [\"sh\", \"-c\", \"printf 2 >> {}\"] }} }}\n  - {{ name: merge, action: {{ builtin: merge }} }}\n",
            invocations.display(),
            invocations.display(),
        ),
    )
    .expect("write named recovery flow");
    world.commit_all("initial");
    let daemon_pid = world.start_daemon()["data"]["pid"].as_u64().unwrap() as u32;
    let ticket_path = world.write_ticket("old-shape-flow.md", "---\nflow: resume\n---\n# Work\n");
    let output = world.sloop(&["post", ticket_path.to_str().unwrap(), "--manual"]);
    assert!(output.status.success());
    let ticket = World::json_stdout(&output)["data"]["ticket"]["id"]
        .as_str()
        .unwrap()
        .to_owned();
    assert!(world.sloop(&["run", &ticket]).status.success());
    wait_until("the first stage is durably complete", || {
        world.test_hook_reached(HOOK)
            && fs::read_to_string(&invocations).is_ok_and(|value| value == "1")
    });

    world.kill_daemon(daemon_pid);
    plant_old_stage_shape(&world);
    world.start_daemon();
    wait_until("the migrated run completes the remaining flow", || {
        tickets(&world)["merged"] == 1
    });

    // `first` is not re-run: the fold read the backfilled rows and resumed at
    // `second`, exactly as it would have without the migration.
    assert_eq!(fs::read_to_string(invocations).unwrap(), "12");
    assert_eq!(
        stage_log(&world, 1),
        [
            (
                1,
                0,
                "build".into(),
                1,
                "action".into(),
                Some("passed".into())
            ),
            (
                2,
                1,
                "first".into(),
                1,
                "action".into(),
                Some("passed".into())
            ),
            (
                3,
                2,
                "second".into(),
                1,
                "action".into(),
                Some("passed".into())
            ),
            (
                4,
                3,
                "merge".into(),
                1,
                "action".into(),
                Some("passed".into())
            ),
        ]
    );
}

#[test]
fn cancel_kills_a_custom_exec_process_group_and_preserves_the_worktree() {
    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    let process_ids = world.root().join("custom-stage-processes");
    write_flow(
        &world,
        &format!(
            "stages:\n  - {{ name: build, action: agent }}\n  - {{ name: wait, action: {{ exec: [\"sh\", \"-c\", \"sleep 1000 & printf '%s %s' $$ $! > {}; wait\"] }} }}\n  - {{ name: merge, action: {{ builtin: merge }} }}\n",
            process_ids.display(),
        ),
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "cancel-flow.md");
    wait_until("the custom stage is checkpointed", || {
        process_ids.is_file()
            && world
                .run_evidence(&world.run_id(1), "stage_process")
                .is_some()
    });
    let process_ids = read_process_ids(process_ids);

    assert!(world.sloop(&["cancel", &world.run_id(1)]).status.success());
    wait_until("the cancelled flow releases its ticket", || {
        let counts = tickets(&world);
        counts["ready"] == 1 && counts["claimed"] == 0
    });
    wait_for_processes_to_exit(process_ids);
    assert!(world.run_worktree(1).is_dir());
}

#[test]
fn cancel_never_signals_a_recycled_stage_process_group() {
    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    let process_ids = world.root().join("recycled-stage-processes");
    write_flow(
        &world,
        &format!(
            "stages:\n  - {{ name: build, action: agent }}\n  - {{ name: wait, action: {{ exec: [\"sh\", \"-c\", \"sleep 1000 & printf '%s %s' $$ $! > {}; wait\"] }} }}\n  - {{ name: merge, action: {{ builtin: merge }} }}\n",
            process_ids.display(),
        ),
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "recycled-stage-pgid.md");
    wait_until("the custom stage is checkpointed", || {
        process_ids.is_file()
            && world
                .run_evidence(&world.run_id(1), "stage_process")
                .is_some()
    });
    let original_processes = read_process_ids(process_ids);
    let mut unrelated = spawn_unrelated_process_group();
    let unrelated_pid = unrelated.id();
    let connection = rusqlite::Connection::open(world.db_path()).unwrap();
    connection
        .execute(
            "UPDATE run_evidence SET data_json = ?1
             WHERE run_id = ?2 AND kind = 'stage_process'",
            [
                serde_json::json!({
                    "stage": "wait",
                    "pid": unrelated_pid,
                    "pid_start_time": 0,
                    "process_group_id": unrelated_pid,
                })
                .to_string(),
                world.run_id(1),
            ],
        )
        .unwrap();

    assert!(world.sloop(&["cancel", &world.run_id(1)]).status.success());
    assert!(
        unrelated.try_wait().unwrap().is_none(),
        "a recycled process group must not be signalled"
    );

    world.kill_process_group(original_processes[0]);
    wait_until("the cancelled flow releases its ticket", || {
        let counts = tickets(&world);
        counts["ready"] == 1 && counts["claimed"] == 0
    });
    unsafe {
        libc::kill(-(unrelated_pid as libc::pid_t), libc::SIGKILL);
    }
    unrelated.wait().unwrap();
}

#[test]
fn exec_stage_exit_kills_pipe_holding_stragglers() {
    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    let process_id = world.root().join("exec-straggler");
    write_flow(
        &world,
        &format!(
            "stages:\n  - {{ name: build, action: agent }}\n  - {{ name: check, action: {{ exec: [\"sh\", \"-c\", \"sleep 600 & echo $! > {}; exit 0\"] }} }}\n  - {{ name: merge, action: {{ builtin: merge }} }}\n",
            process_id.display(),
        ),
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "exec-straggler.md");

    wait_until("the flow settles despite inherited pipes", || {
        tickets(&world)["merged"] == 1
    });
    wait_for_processes_to_exit(read_process_ids(process_id));
}

#[test]
fn recovery_does_not_signal_a_group_after_the_recorded_leader_exits() {
    const HOOK: &str = "after-stage-process-checkpoint-wait";

    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    world.arm_test_hook(HOOK);
    let process_ids = world.root().join("orphaned-exec-processes");
    write_flow(
        &world,
        &format!(
            "stages:\n  - {{ name: build, action: agent }}\n  - {{ name: wait, action: {{ exec: [\"sh\", \"-c\", \"sleep 1000 & printf '%s %s\\n' $$ $! >> {}; exit 0\"] }} }}\n  - {{ name: merge, action: {{ builtin: merge }} }}\n",
            process_ids.display(),
        ),
    );
    world.commit_all("initial");
    let daemon_pid = world.start_daemon()["data"]["pid"].as_u64().unwrap() as u32;
    post_and_run(&world, "orphaned-exec-child.md");
    wait_until("the exec checkpoint is durable", || {
        world.test_hook_reached(HOOK)
            && world
                .run_evidence(&world.run_id(1), "stage_process")
                .is_some()
            && process_ids.is_file()
    });
    let first_processes = read_process_ids(process_ids.clone());

    world.kill_daemon(daemon_pid);
    wait_until("the exec leader exits while its child survives", || {
        !process_alive(first_processes[0]) && process_alive(first_processes[1])
    });
    world.release_test_hook(HOOK);
    world.start_daemon();
    wait_until(
        "recovery completes without signalling the stale group",
        || tickets(&world)["merged"] == 1,
    );
    assert!(
        process_alive(first_processes[1]),
        "the unverifiable leaderless group must not be signalled"
    );
    assert!(
        fs::read_to_string(world.daemon_log())
            .unwrap()
            .contains("stale_stage_group_not_signalled")
    );
    unsafe {
        libc::kill(first_processes[1] as libc::pid_t, libc::SIGKILL);
    }
    wait_until("the stale descendant is cleaned up by the test", || {
        !process_alive(first_processes[1])
    });
}

#[test]
fn recovery_preserves_partial_merge_state_and_fails_for_review() {
    let world = World::configured();
    let fixture = configure_blocking_merge(&world);
    let daemon_pid = world.start_daemon()["data"]["pid"].as_u64().unwrap() as u32;
    post_and_run(&world, "interrupted-merge.md");
    wait_until("the first merge reaches its hook", || {
        fixture.started.is_file()
            && world
                .run_evidence(&world.run_id(1), "stage_process")
                .is_some()
    });

    world.kill_daemon(daemon_pid);
    let status_before = Command::new("git")
        .args(["status", "--porcelain=v1", "-z"])
        .current_dir(world.root())
        .output()
        .unwrap()
        .stdout;
    let index_before = Command::new("git")
        .args(["ls-files", "--stage", "-z"])
        .current_dir(world.root())
        .output()
        .unwrap()
        .stdout;
    let work_before = fs::read(world.root().join("work.txt")).ok();
    world.start_daemon();
    wait_until("the partial merge fails safely for review", || {
        tickets(&world)["needs_review"] == 1
    });
    let status_after = Command::new("git")
        .args(["status", "--porcelain=v1", "-z"])
        .current_dir(world.root())
        .output()
        .unwrap()
        .stdout;
    let index_after = Command::new("git")
        .args(["ls-files", "--stage", "-z"])
        .current_dir(world.root())
        .output()
        .unwrap()
        .stdout;
    assert_eq!(status_after, status_before);
    assert_eq!(index_after, index_before);
    assert_eq!(fs::read(world.root().join("work.txt")).ok(), work_before);
    assert_eq!(fs::read_to_string(fixture.invocations).unwrap(), "x");
}

#[test]
fn recovery_settles_a_completed_uncheckpointed_merge_idempotently() {
    const HOOK: &str = "after-successful-merge-process-exit";

    let world = World::configured();
    world.arm_test_hook(HOOK);
    configure(&world, COMMITTING_AGENT, None);
    world.commit_all("initial");
    let daemon_pid = world.start_daemon()["data"]["pid"].as_u64().unwrap() as u32;
    post_and_run(&world, "completed-before-checkpoint.md");
    wait_until("the merge completes before its stage checkpoint", || {
        world.test_hook_reached(HOOK)
            && default_branch_has(&world, "work.txt")
            && world
                .run_evidence(&world.run_id(1), "stage_process")
                .is_some()
    });

    world.kill_daemon(daemon_pid);
    world.release_test_hook(HOOK);
    world.start_daemon();
    wait_until("recovery settles the completed merge", || {
        tickets(&world)["merged"] == 1
    });
    assert!(default_branch_has(&world, "work.txt"));
    assert_eq!(
        stage_runs(&world, 1)
            .into_iter()
            .map(|(_, stage, state, _)| (stage, state))
            .collect::<Vec<_>>(),
        [
            ("build".into(), "passed".into()),
            ("merge".into(), "passed".into())
        ]
    );
}

#[test]
fn recovery_does_not_reapply_a_completed_merge_after_target_reset() {
    const HOOK: &str = "after-successful-merge-process-exit";

    let world = World::configured();
    world.arm_test_hook(HOOK);
    configure(&world, COMMITTING_AGENT, None);
    world.commit_all("initial");
    let initial_head = Command::new("git")
        .args(["rev-parse", "HEAD"])
        .current_dir(world.root())
        .output()
        .unwrap();
    let initial_head = String::from_utf8(initial_head.stdout).unwrap();
    let daemon_pid = world.start_daemon()["data"]["pid"].as_u64().unwrap() as u32;
    post_and_run(&world, "reset-after-merge.md");
    wait_until("the merge completes before target reset", || {
        world.test_hook_reached(HOOK)
            && world
                .run_evidence(&world.run_id(1), "stage_process")
                .is_some_and(|evidence| evidence["merge"]["completed_target"].is_string())
    });

    world.kill_daemon(daemon_pid);
    let reset = Command::new("git")
        .args(["reset", "--hard", initial_head.trim()])
        .current_dir(world.root())
        .status()
        .unwrap();
    assert!(reset.success());
    world.release_test_hook(HOOK);
    world.start_daemon();
    wait_until("the reset merge is left for review", || {
        tickets(&world)["needs_review"] == 1
    });
    let head = Command::new("git")
        .args(["rev-parse", "HEAD"])
        .current_dir(world.root())
        .output()
        .unwrap();
    assert_eq!(head.stdout, initial_head.as_bytes());
    assert!(!default_branch_has(&world, "work.txt"));
}

#[test]
fn recovery_does_not_reapply_a_completed_merge_after_target_moves() {
    const HOOK: &str = "after-successful-merge-process-exit";

    let world = World::configured();
    world.arm_test_hook(HOOK);
    configure(&world, COMMITTING_AGENT, None);
    world.commit_all("initial");
    let initial_head = Command::new("git")
        .args(["rev-parse", "HEAD"])
        .current_dir(world.root())
        .output()
        .unwrap();
    let initial_head = String::from_utf8(initial_head.stdout).unwrap();
    let daemon_pid = world.start_daemon()["data"]["pid"].as_u64().unwrap() as u32;
    post_and_run(&world, "move-after-merge.md");
    wait_until("the merge completes before target move", || {
        world.test_hook_reached(HOOK)
            && world
                .run_evidence(&world.run_id(1), "stage_process")
                .is_some_and(|evidence| evidence["merge"]["completed_target"].is_string())
    });

    world.kill_daemon(daemon_pid);
    assert!(
        Command::new("git")
            .args(["reset", "--hard", initial_head.trim()])
            .current_dir(world.root())
            .status()
            .unwrap()
            .success()
    );
    fs::write(world.root().join("operator.txt"), "operator\n").unwrap();
    assert!(
        Command::new("git")
            .args(["add", "operator.txt"])
            .current_dir(world.root())
            .status()
            .unwrap()
            .success()
    );
    assert!(
        Command::new("git")
            .args([
                "-c",
                "user.name=operator",
                "-c",
                "user.email=operator@example.invalid",
                "commit",
                "--quiet",
                "-m",
                "operator moved target",
            ])
            .current_dir(world.root())
            .status()
            .unwrap()
            .success()
    );
    let moved_head = Command::new("git")
        .args(["rev-parse", "HEAD"])
        .current_dir(world.root())
        .output()
        .unwrap()
        .stdout;
    world.release_test_hook(HOOK);
    world.start_daemon();
    wait_until("the moved target is left for review", || {
        tickets(&world)["needs_review"] == 1
    });
    assert_eq!(
        Command::new("git")
            .args(["rev-parse", "HEAD"])
            .current_dir(world.root())
            .output()
            .unwrap()
            .stdout,
        moved_head
    );
    assert_eq!(
        fs::read_to_string(world.root().join("operator.txt")).unwrap(),
        "operator\n"
    );
    assert!(!default_branch_has(&world, "work.txt"));
}

#[test]
fn recovery_does_not_merge_an_advanced_run_branch() {
    const HOOK: &str = "after-stage-process-checkpoint-merge";

    let world = World::configured();
    world.arm_test_hook(HOOK);
    configure(&world, COMMITTING_AGENT, None);
    world.commit_all("initial");
    let initial_head = Command::new("git")
        .args(["rev-parse", "HEAD"])
        .current_dir(world.root())
        .output()
        .unwrap()
        .stdout;
    let daemon_pid = world.start_daemon()["data"]["pid"].as_u64().unwrap() as u32;
    post_and_run(&world, "advanced-run-branch.md");
    wait_until("the blocked merge has its baseline checkpoint", || {
        world.test_hook_reached(HOOK)
            && world
                .run_evidence(&world.run_id(1), "stage_process")
                .is_some_and(|evidence| evidence["merge"]["branch_tip"].is_string())
    });

    world.kill_daemon(daemon_pid);
    let worktree = world.run_worktree(1);
    fs::write(worktree.join("advanced.txt"), "advanced\n").unwrap();
    assert!(
        Command::new("git")
            .args(["add", "advanced.txt"])
            .current_dir(&worktree)
            .status()
            .unwrap()
            .success()
    );
    assert!(
        Command::new("git")
            .args([
                "-c",
                "user.name=operator",
                "-c",
                "user.email=operator@example.invalid",
                "commit",
                "--quiet",
                "-m",
                "advanced run branch",
            ])
            .current_dir(&worktree)
            .status()
            .unwrap()
            .success()
    );
    world.release_test_hook(HOOK);
    world.start_daemon();
    wait_until("the advanced run branch is left for review", || {
        tickets(&world)["needs_review"] == 1
    });
    assert_eq!(
        Command::new("git")
            .args(["rev-parse", "HEAD"])
            .current_dir(world.root())
            .output()
            .unwrap()
            .stdout,
        initial_head
    );
    assert!(!default_branch_has(&world, "work.txt"));
    assert!(!default_branch_has(&world, "advanced.txt"));
}

#[test]
fn recovery_preserves_conflicted_and_unrelated_operator_edits_after_merge_exit() {
    const HOOK: &str = "after-failed-merge-process-exit";

    let world = World::configured();
    world.arm_test_hook(HOOK);
    configure(
        &world,
        concat!(
            "echo agent-version > work.txt\n",
            "git add work.txt\n",
            "git -c user.name=agent -c user.email=agent@example.invalid commit --quiet -m agent\n",
            "root=\"$(dirname \"$0\")\"\n",
            "echo main-version > \"$root/work.txt\"\n",
            "git -C \"$root\" add work.txt\n",
            "git -C \"$root\" -c user.name=operator -c user.email=operator@example.invalid commit --quiet -m main\n",
        ),
        None,
    );
    world.commit_all("initial");
    let daemon_pid = world.start_daemon()["data"]["pid"].as_u64().unwrap() as u32;
    post_and_run(&world, "exited-conflict.md");
    wait_until("the failed merge exits with owned conflict state", || {
        world.test_hook_reached(HOOK)
            && world
                .run_evidence(&world.run_id(1), "stage_process")
                .is_some()
            && world.root().join(".git/MERGE_HEAD").is_file()
    });
    let merge_pid = world
        .run_evidence(&world.run_id(1), "stage_process")
        .unwrap()["pid"]
        .as_u64()
        .unwrap() as u32;
    assert!(!process_alive(merge_pid));

    let conflict_edit = b"operator conflict edit\n";
    let unrelated_edit = b"operator unrelated edit\n";
    fs::write(world.root().join("work.txt"), conflict_edit).unwrap();
    fs::write(world.root().join("operator.txt"), unrelated_edit).unwrap();
    let added = Command::new("git")
        .args(["add", "operator.txt"])
        .current_dir(world.root())
        .status()
        .unwrap();
    assert!(added.success());
    let status_before = Command::new("git")
        .args(["status", "--porcelain=v1", "-z"])
        .current_dir(world.root())
        .output()
        .unwrap()
        .stdout;
    let index_before = Command::new("git")
        .args(["ls-files", "--stage", "-z"])
        .current_dir(world.root())
        .output()
        .unwrap()
        .stdout;
    let merge_head_before = fs::read(world.root().join(".git/MERGE_HEAD")).unwrap();
    let merge_message_before = fs::read(world.root().join(".git/MERGE_MSG")).unwrap();

    world.kill_daemon(daemon_pid);
    world.release_test_hook(HOOK);
    world.start_daemon();
    wait_until("recovery leaves the edited conflict for review", || {
        tickets(&world)["needs_review"] == 1
    });
    assert_eq!(
        fs::read(world.root().join("work.txt")).unwrap(),
        conflict_edit
    );
    assert_eq!(
        fs::read(world.root().join("operator.txt")).unwrap(),
        unrelated_edit
    );
    assert_eq!(
        fs::read(world.root().join(".git/MERGE_HEAD")).unwrap(),
        merge_head_before
    );
    assert_eq!(
        fs::read(world.root().join(".git/MERGE_MSG")).unwrap(),
        merge_message_before
    );
    assert_eq!(
        Command::new("git")
            .args(["status", "--porcelain=v1", "-z"])
            .current_dir(world.root())
            .output()
            .unwrap()
            .stdout,
        status_before
    );
    assert_eq!(
        Command::new("git")
            .args(["ls-files", "--stage", "-z"])
            .current_dir(world.root())
            .output()
            .unwrap()
            .stdout,
        index_before
    );
}

#[test]
fn recovery_preserves_operator_changes_made_during_an_interrupted_merge() {
    let world = World::configured();
    let fixture = configure_blocking_merge(&world);
    let daemon_pid = world.start_daemon()["data"]["pid"].as_u64().unwrap() as u32;
    post_and_run(&world, "operator-changed-merge.md");
    wait_until("the merge reaches its hook", || {
        fixture.started.is_file()
            && world
                .run_evidence(&world.run_id(1), "stage_process")
                .is_some()
    });
    let merge_pid = world
        .run_evidence(&world.run_id(1), "stage_process")
        .unwrap()["pid"]
        .as_u64()
        .unwrap() as u32;

    world.kill_daemon(daemon_pid);
    fs::write(world.root().join("operator.txt"), "operator\n").unwrap();
    let added = Command::new("git")
        .args(["add", "operator.txt"])
        .current_dir(world.root())
        .status()
        .unwrap();
    assert!(added.success());
    let staged_before = Command::new("git")
        .args(["diff", "--cached", "--name-only"])
        .current_dir(world.root())
        .output()
        .unwrap()
        .stdout;
    world.start_daemon();
    wait_until("recovery leaves the modified merge for review", || {
        !process_alive(merge_pid) && tickets(&world)["needs_review"] == 1
    });

    let staged = Command::new("git")
        .args(["diff", "--cached", "--name-only"])
        .current_dir(world.root())
        .output()
        .unwrap();
    assert!(
        String::from_utf8_lossy(&staged.stdout)
            .lines()
            .any(|path| path == "operator.txt"),
        "operator.txt must remain staged: {}",
        String::from_utf8_lossy(&staged.stdout)
    );
    assert_eq!(staged.stdout, staged_before);
    assert_eq!(fs::read_to_string(fixture.invocations).unwrap(), "x");
}

#[test]
fn recovery_preserves_unrelated_merge_state() {
    let world = World::configured();
    let fixture = configure_blocking_merge(&world);
    let daemon_pid = world.start_daemon()["data"]["pid"].as_u64().unwrap() as u32;
    post_and_run(&world, "unrelated-merge.md");
    wait_until("the merge reaches its hook", || {
        fixture.started.is_file()
            && world
                .run_evidence(&world.run_id(1), "stage_process")
                .is_some()
    });
    let merge_pid = world
        .run_evidence(&world.run_id(1), "stage_process")
        .unwrap()["pid"]
        .as_u64()
        .unwrap() as u32;
    world.kill_daemon(daemon_pid);
    let unrelated_head = Command::new("git")
        .args(["rev-parse", "HEAD"])
        .current_dir(world.root())
        .output()
        .unwrap();
    let unrelated_head = String::from_utf8(unrelated_head.stdout).unwrap();
    fs::write(world.root().join(".git/MERGE_HEAD"), &unrelated_head).unwrap();
    let status_before = Command::new("git")
        .args(["status", "--porcelain=v1", "-z"])
        .current_dir(world.root())
        .output()
        .unwrap()
        .stdout;
    let index_before = Command::new("git")
        .args(["ls-files", "--stage", "-z"])
        .current_dir(world.root())
        .output()
        .unwrap()
        .stdout;

    world.start_daemon();
    wait_until("recovery refuses the unrelated merge", || {
        !process_alive(merge_pid) && tickets(&world)["needs_review"] == 1
    });
    assert_eq!(
        fs::read_to_string(world.root().join(".git/MERGE_HEAD")).unwrap(),
        unrelated_head
    );
    assert_eq!(
        Command::new("git")
            .args(["status", "--porcelain=v1", "-z"])
            .current_dir(world.root())
            .output()
            .unwrap()
            .stdout,
        status_before
    );
    assert_eq!(
        Command::new("git")
            .args(["ls-files", "--stage", "-z"])
            .current_dir(world.root())
            .output()
            .unwrap()
            .stdout,
        index_before
    );
    assert_eq!(fs::read_to_string(fixture.invocations).unwrap(), "x");
}

#[test]
fn recovery_never_removes_an_unowned_index_lock() {
    let world = World::configured();
    let fixture = configure_blocking_merge(&world);
    let daemon_pid = world.start_daemon()["data"]["pid"].as_u64().unwrap() as u32;
    post_and_run(&world, "unowned-index-lock.md");
    wait_until("the merge reaches its hook", || {
        fixture.started.is_file()
            && world
                .run_evidence(&world.run_id(1), "stage_process")
                .is_some()
    });
    let merge_pid = world
        .run_evidence(&world.run_id(1), "stage_process")
        .unwrap()["pid"]
        .as_u64()
        .unwrap() as u32;
    world.kill_daemon(daemon_pid);
    let index_lock = world.root().join(".git/index.lock");
    fs::write(&index_lock, "operator lock\n").unwrap();
    let status_before = Command::new("git")
        .args(["status", "--porcelain=v1", "-z"])
        .current_dir(world.root())
        .output()
        .unwrap()
        .stdout;
    let index_before = fs::read(world.root().join(".git/index")).unwrap();

    world.start_daemon();
    wait_until("recovery refuses the unowned index lock", || {
        !process_alive(merge_pid) && tickets(&world)["needs_review"] == 1
    });
    assert_eq!(fs::read_to_string(index_lock).unwrap(), "operator lock\n");
    assert_eq!(
        fs::read(world.root().join(".git/index")).unwrap(),
        index_before
    );
    assert_eq!(
        Command::new("git")
            .args(["status", "--porcelain=v1", "-z"])
            .current_dir(world.root())
            .output()
            .unwrap()
            .stdout,
        status_before
    );
    assert_eq!(fs::read_to_string(fixture.invocations).unwrap(), "x");
}

#[test]
fn cancellation_at_arbitrary_exec_startup_kills_the_stage_group() {
    const HOOK: &str = "before-stage-process-checkpoint-wait";

    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    world.arm_test_hook(HOOK);
    let process_ids = world.root().join("racing-exec-processes");
    write_flow(
        &world,
        &format!(
            "stages:\n  - {{ name: build, action: agent }}\n  - {{ name: wait, action: {{ exec: [\"sh\", \"-c\", \"sleep 1000 & printf '%s %s' $$ $! > {}; wait\"] }} }}\n  - {{ name: merge, action: {{ builtin: merge }} }}\n",
            process_ids.display(),
        ),
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "cancel-racing-exec.md");
    wait_until("the arbitrary exec reaches its startup gate", || {
        world.test_hook_reached(HOOK) && process_ids.is_file()
    });
    assert!(
        world
            .run_evidence(&world.run_id(1), "stage_process")
            .is_none()
    );

    assert!(world.sloop(&["cancel", &world.run_id(1)]).status.success());
    let process_ids = read_process_ids(process_ids);
    world.release_test_hook(HOOK);
    wait_until("the startup-racing exec is cancelled", || {
        let counts = tickets(&world);
        counts["ready"] == 1 && counts["claimed"] == 0
    });
    wait_for_processes_to_exit(process_ids);
}

#[test]
fn restart_reruns_an_interrupted_stage_and_then_merges() {
    let world = World::configured();
    let started = world.root().join("test-started");
    let release = world.root().join("test-release");
    let finished = world.root().join("test-finished");
    let invocations = world.root().join("test-invocations");
    let test_cmd = format!(
        "printf x >> {invocations}; touch {started}; while [ ! -e {release} ]; do sleep 0.01; done; touch {finished}",
        invocations = invocations.display(),
        started = started.display(),
        release = release.display(),
        finished = finished.display(),
    );
    configure(&world, COMMITTING_AGENT, Some(&test_cmd));
    world.commit_all("initial");
    let daemon_pid = world.start_daemon()["data"]["pid"]
        .as_u64()
        .expect("daemon pid") as u32;
    post_and_run(&world, "recover-stage.md");
    wait_until("the first test invocation starts", || started.is_file());

    world.kill_daemon(daemon_pid);
    world.start_daemon();
    wait_until("recovery replaces the interrupted test", || {
        fs::read_to_string(&invocations).is_ok_and(|contents| contents == "xx")
    });
    fs::write(&release, "").expect("release interrupted test");

    wait_until("the resumed driver merges the work", || {
        tickets(&world)["merged"] == 1
    });
    assert!(finished.is_file());
    assert_eq!(
        fs::read_to_string(invocations).expect("read invocation record"),
        "xx"
    );
    assert!(default_branch_has(&world, "work.txt"));
}

#[test]
fn cancel_stops_an_active_stage_process_and_releases_the_ticket() {
    let world = World::configured();
    let process_ids = world.root().join("cancel-test-processes");
    let test_cmd = format!(
        "sleep 1000 & printf '%s %s' $$ $! > {process_ids}; wait",
        process_ids = process_ids.display(),
    );
    configure(&world, COMMITTING_AGENT, Some(&test_cmd));
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "cancel-stage.md");
    wait_until("the stage process checkpoint is durable", || {
        process_ids.is_file()
            && world
                .run_evidence(&world.run_id(1), "stage_process")
                .is_some()
    });
    let process_ids = read_process_ids(process_ids);

    let cancelled = world.sloop(&["cancel", &world.run_id(1)]);
    assert!(cancelled.status.success());
    wait_until("the cancelled run settles", || {
        let counts = tickets(&world);
        counts["ready"] == 1 && counts["claimed"] == 0
    });
    assert_eq!(
        world.wait_snapshot(&world.run_id(1))["data"]["state"],
        "cancelled"
    );
    wait_for_processes_to_exit(process_ids);
}

#[test]
fn cancellation_before_the_test_process_checkpoint_stops_the_walk() {
    const HOOK: &str = "before-test-process-checkpoint";

    let world = World::configured();
    world.arm_test_hook(HOOK);
    let process_ids = world.root().join("racing-test-processes");
    let test_cmd = format!(
        "sleep 1000 & printf '%s %s' $$ $! > {process_ids}; wait",
        process_ids = process_ids.display(),
    );
    configure(&world, COMMITTING_AGENT, Some(&test_cmd));
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "cancel-before-checkpoint.md");
    wait_until("the test process reaches its checkpoint gate", || {
        world.test_hook_reached(HOOK) && process_ids.is_file()
    });
    assert!(
        world
            .run_evidence(&world.run_id(1), "stage_process")
            .is_none()
    );

    let cancelled = world.sloop(&["cancel", &world.run_id(1)]);
    assert!(cancelled.status.success());
    assert!(
        world
            .run_evidence(&world.run_id(1), "cancel_requested")
            .is_some()
    );
    assert!(
        world
            .run_evidence(&world.run_id(1), "stage_process")
            .is_none()
    );
    let process_ids = read_process_ids(process_ids);
    world.release_test_hook(HOOK);

    wait_until("the cancelled run settles", || {
        let counts = tickets(&world);
        counts["ready"] == 1 && counts["claimed"] == 0
    });
    assert_eq!(
        world.wait_snapshot(&world.run_id(1))["data"]["state"],
        "cancelled"
    );
    assert!(
        world
            .run_evidence(&world.run_id(1), "stage_process")
            .is_none()
    );
    wait_for_processes_to_exit(process_ids);
}

#[test]
fn a_moved_default_branch_without_conflicts_still_merges() {
    let world = World::configured();
    // The agent commits its work, then moves the default branch with an
    // unrelated commit before exiting — deterministically simulating an
    // operator landing something mid-run. `$0` is the script's absolute
    // path in the repository root.
    configure(
        &world,
        concat!(
            "echo done > work.txt\n",
            "git add work.txt\n",
            "git -c user.name=agent -c user.email=agent@example.invalid commit --quiet -m 'agent work'\n",
            "root=\"$(dirname \"$0\")\"\n",
            "echo moved > \"$root/moved.txt\"\n",
            "git -C \"$root\" add moved.txt\n",
            "git -C \"$root\" -c user.name=op -c user.email=op@example.invalid commit --quiet -m 'main moved'\n",
            "exit 0\n",
        ),
        None,
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "diverge-me.md");

    wait_until("the ticket reaches merged", || {
        tickets(&world)["merged"] == 1
    });
    // Both histories are present: the merge produced a merge commit rather
    // than parking the run for a human.
    assert!(default_branch_has(&world, "work.txt"));
    assert!(default_branch_has(&world, "moved.txt"));
}

#[test]
fn a_conflicting_merge_parks_the_ticket_and_preserves_the_conflict() {
    let world = World::configured();
    // Same interleaving, but both sides edit the same file: the merge state is
    // evidence for the human and must not be automatically aborted.
    configure(
        &world,
        concat!(
            "echo agent-version > work.txt\n",
            "git add work.txt\n",
            "git -c user.name=agent -c user.email=agent@example.invalid commit --quiet -m 'agent work'\n",
            "root=\"$(dirname \"$0\")\"\n",
            "echo main-version > \"$root/work.txt\"\n",
            "git -C \"$root\" add work.txt\n",
            "git -C \"$root\" -c user.name=op -c user.email=op@example.invalid commit --quiet -m 'main moved'\n",
            "exit 0\n",
        ),
        None,
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "conflict-me.md");

    wait_until("the ticket reaches needs_review", || {
        tickets(&world)["needs_review"] == 1
    });
    assert!(world.root().join(".git/MERGE_HEAD").is_file());
    let status = Command::new("git")
        .args(["status", "--porcelain", "--untracked-files=no"])
        .current_dir(world.root())
        .output()
        .unwrap();
    assert!(
        String::from_utf8_lossy(&status.stdout).contains("AA work.txt"),
        "conflict must remain visible: {}",
        String::from_utf8_lossy(&status.stdout)
    );
    let conflict = fs::read_to_string(world.root().join("work.txt")).unwrap();
    assert!(conflict.contains("agent-version"));
    assert!(conflict.contains("main-version"));
}

#[test]
fn exit_zero_without_commits_needs_review_before_tests() {
    let world = World::configured();
    configure(&world, "exit 0\n", Some("echo tested > tested.txt"));
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "empty.md");

    wait_until("the ticket reaches needs review", || {
        tickets(&world)["needs_review"] == 1
    });
    assert!(!world.run_worktree(1).join("tested.txt").exists());
}

#[test]
fn rewriting_the_default_branch_does_not_invent_run_commits() {
    let world = World::configured();
    configure(
        &world,
        concat!(
            "root=\"$(dirname \"$0\")\"\n",
            "echo rewritten > \"$root/rewritten.txt\"\n",
            "git -C \"$root\" add rewritten.txt\n",
            "git -C \"$root\" -c user.name=op -c user.email=op@example.invalid commit --quiet --amend --no-edit\n",
            "exit 0\n",
        ),
        None,
    );
    world.commit_all("initial");
    world.start_daemon();
    let ticket = post_and_run(&world, "rewrite.md");

    wait_until("the no-op ticket reaches needs review", || {
        tickets(&world)["needs_review"] == 1
    });
    assert!(default_branch_has(&world, "rewritten.txt"));

    let shown = world.sloop(&["show", "default"]);
    assert!(shown.status.success());
    let response = World::json_stdout(&shown);
    let activity = response["data"]["value"]["tickets"]
        .as_array()
        .unwrap()
        .iter()
        .find(|entry| entry["id"] == ticket)
        .expect("ticket activity");
    assert_eq!(activity["commits"], serde_json::json!([]));
}

#[test]
fn a_nonzero_exit_fails_even_when_the_agent_committed() {
    let world = World::configured();
    configure(
        &world,
        &COMMITTING_AGENT.replace("exit 0", "exit 1"),
        Some("echo should-not-run > tested.txt"),
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "failed.md");

    wait_until("the ticket reaches failed", || {
        tickets(&world)["failed"] == 1
    });
    assert!(!world.run_worktree(1).join("tested.txt").exists());
}

#[test]
fn committed_work_with_failing_tests_is_left_for_review() {
    let world = World::configured();
    configure(&world, COMMITTING_AGENT, Some("exit 1"));
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "review-me.md");

    wait_until("the ticket reaches needs_review", || {
        tickets(&world)["needs_review"] == 1
    });
    assert!(
        !default_branch_has(&world, "work.txt"),
        "failing tests must never merge"
    );

    // The branch survives so a human can salvage the commits.
    let branches = Command::new("git")
        .args(["branch", "--list", "sloop/*"])
        .current_dir(world.root())
        .output()
        .unwrap();
    assert!(!String::from_utf8_lossy(&branches.stdout).trim().is_empty());
}

fn pid_alive(pid: &str) -> bool {
    Command::new("kill")
        .args(["-0", pid])
        .status()
        .map(|status| status.success())
        .unwrap_or(false)
}

#[test]
fn cancel_kills_the_whole_process_group_and_frees_the_ticket() {
    let world = World::configured();
    // The agent backgrounds a grandchild, records its PID, then blocks.
    configure(
        &world,
        concat!(
            "echo started\n",
            "sh -c 'sleep 30' &\n",
            "echo $! > grandchild.pid\n",
            "tries=0\n",
            "while [ \"$tries\" -lt 400 ]; do sleep 0.05; tries=$((tries + 1)); done\n",
        ),
        None,
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "cancel-me.md");

    let pid_file: PathBuf = world.run_worktree(1).join("grandchild.pid");
    wait_until("the agent starts and records its grandchild", || {
        pid_file.is_file()
    });
    let grandchild = fs::read_to_string(&pid_file).unwrap().trim().to_owned();
    assert!(pid_alive(&grandchild));

    let output = world.sloop(&["cancel", &world.run_id(1)]);
    assert!(
        output.status.success(),
        "cancel failed: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    let data = World::json_stdout(&output)["data"].clone();
    assert_eq!(data["id"], world.run_id(1));
    assert_eq!(data["state"], "cancelling");
    assert_eq!(data["preserved"], true);

    wait_until("the grandchild dies with the group", || {
        !pid_alive(&grandchild)
    });
    wait_until("the ticket returns to ready", || {
        tickets(&world)["ready"] == 1
    });

    // Worktree, branch, and captured logs survive cancellation as evidence.
    assert!(world.run_worktree(1).is_dir());
    let output = world.sloop(&["logs", &world.run_id(1)]);
    assert!(output.status.success());
    assert!(
        !World::json_stdout(&output)["data"]["entries"]
            .as_array()
            .unwrap()
            .is_empty()
    );

    // Cancelling twice while the exit is already resolved cannot double-free.
    let repeat = world.sloop(&["cancel", &world.run_id(1)]);
    let response = json_stderr(&repeat);
    assert_eq!(response["ok"], false);
    assert_eq!(response["error"]["code"], "conflict");
}

#[test]
fn a_finished_run_cannot_be_cancelled() {
    let world = World::configured();
    configure(&world, "exit 0\n", None);
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "done.md");

    wait_until("the run finishes", || tickets(&world)["needs_review"] == 1);

    let output = world.sloop(&["cancel", &world.run_id(1)]);
    let response = json_stderr(&output);
    assert_eq!(response["ok"], false);
    assert_eq!(response["error"]["code"], "conflict");

    let missing = world.sloop(&["cancel", "R99"]);
    let response = json_stderr(&missing);
    assert_eq!(response["error"]["code"], "not_found");
}

#[test]
fn straggler_group_members_are_killed_and_the_run_still_settles() {
    let world = World::configured();
    let straggler_pid = world.root().join("straggler-pid");
    let agent = format!(
        "sleep 600 &\necho $! > {pid_file}\n{committing}",
        pid_file = straggler_pid.display(),
        committing = COMMITTING_AGENT,
    );
    configure(&world, &agent, Some("echo tests passed"));
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "straggler.md");

    // Without the straggler kill, the background sleep keeps the agent's
    // stdout pipe open and the run never leaves `running`.
    wait_until("the ticket reaches merged", || {
        tickets(&world)["merged"] == 1
    });
    wait_for_processes_to_exit(read_process_ids(straggler_pid));
    wait_until("no agents remain active", || {
        let output = world.sloop(&["status"]);
        World::json_stdout(&output)["data"]["gate"]["active_agents"] == 0
    });
}

#[test]
fn authentication_and_configuration_rejections_fail_without_later_stages() {
    let cases = [
        (
            "auth.md",
            "printf '%s\n' 'status 401 Missing bearer or basic authentication in header' >&2\nexit 1\n",
            "authentication_required",
            "codex.authentication.missing-header",
        ),
        (
            "config.md",
            "printf '%s\n' 'status 400 model is not supported when using Codex with a ChatGPT account' >&2\nexit 1\n",
            "invalid_configuration",
            "codex.configuration.unsupported-chatgpt-model",
        ),
    ];

    for (ticket_name, body, class, rule_id) in cases {
        let world = World::configured();
        let stage_marker = world.root().join("later-stage-ran");
        configure(
            &world,
            body,
            Some(&format!("touch {}", stage_marker.display())),
        );
        world.commit_all("initial");
        world.start_daemon();
        let ticket = post_and_run(&world, ticket_name);

        wait_until("the rejected run fails", || tickets(&world)["failed"] == 1);
        assert!(!stage_marker.exists());
        let evidence = world
            .run_evidence(&world.run_id(1), "vendor_error_classified")
            .expect("vendor classification evidence");
        assert_eq!(evidence["class"], class);
        assert_eq!(evidence["vendor"], "codex");
        assert_eq!(evidence["rule_id"], rule_id);
        let encoded = evidence.to_string();
        assert!(
            !encoded.contains("Missing bearer"),
            "unsafe evidence: {encoded}"
        );
        assert!(
            !encoded.contains("ChatGPT account"),
            "unsafe evidence: {encoded}"
        );

        let shown = world.sloop(&["show", &ticket]);
        assert!(shown.status.success());
        assert_eq!(
            World::json_stdout(&shown)["data"]["value"]["classification"]["class"],
            class
        );
        assert!(world.sloop(&["retry", &ticket]).status.success());
        let shown = world.sloop(&["show", &ticket]);
        assert!(World::json_stdout(&shown)["data"]["value"]["classification"].is_null());
    }
}

#[test]
fn retryable_and_unknown_rejections_release_under_a_target_cooldown() {
    let cases = [
        (
            "rate.md",
            "printf \"You've hit your limit\\n\" >&2\nexit 1\n",
            "rate_limited",
        ),
        (
            "unknown.md",
            "printf '%s\n' 'Unexpected server error. Check server logs for details.' >&2\nexit 1\n",
            "unknown_rejection",
        ),
    ];

    for (ticket_name, body, class) in cases {
        let world = World::configured();
        configure(&world, body, None);
        world.commit_all("initial");
        world.start_daemon();
        post_and_run(&world, ticket_name);

        wait_until("the rejected ticket is released", || {
            let counts = tickets(&world);
            counts["ready"] == 1 && counts["claimed"] == 0
        });
        let response = world.wait_snapshot(&world.run_id(1));
        assert_eq!(response["data"]["state"], "rate_limited");
        assert_eq!(response["data"]["classification"]["class"], class);

        let status = world.sloop(&["status"]);
        assert_eq!(
            World::json_stdout(&status)["data"]["gate"]["cooldowns"][0]["target"],
            "fake"
        );

        let listed = world.sloop(&["list"]);
        let reason = World::json_stdout(&listed)["data"]["tickets"][0]["reason"]
            .as_str()
            .unwrap()
            .to_owned();
        assert!(reason.contains("cooling down"), "{reason}");
    }
}

/// Verbatim stdout line captured from a `claude` run that hit a session
/// limit. The target runs with `--output-format stream-json`, so the vendor's
/// refusal arrives as a synthetic assistant message on stdout instead of as
/// stderr text.
const CLAUDE_SESSION_LIMIT_STDOUT: &str = r#"{"type":"assistant","message":{"id":"462199bc-1a76-4c09-aaff-df681c45b92c","model":"<synthetic>","role":"assistant","stop_reason":"stop_sequence","stop_sequence":"","type":"message","content":[{"type":"text","text":"You've hit your session limit · resets 12:50pm (Australia/Sydney)"}]},"session_id":"7f4420d2-0557-4535-849c-975a1adda3bc","error":"rate_limit","is_api_error_message":true}"#;

/// The rejection a real run hit: recognised from the JSON payload on stdout,
/// it has to cool the target down rather than spend the ticket's attempts.
#[test]
fn a_claude_session_limit_on_stdout_cools_the_target_down() {
    let world = World::configured();
    configure(
        &world,
        &format!("cat <<'PAYLOAD'\n{CLAUDE_SESSION_LIMIT_STDOUT}\nPAYLOAD\nexit 1\n"),
        None,
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "session-limit.md");

    wait_until("the session-limited ticket is released", || {
        let counts = tickets(&world);
        counts["ready"] == 1 && counts["failed"] == 0
    });
    let snapshot = world.wait_snapshot(&world.run_id(1));
    assert_eq!(snapshot["data"]["state"], "rate_limited");
    assert_eq!(snapshot["data"]["classification"]["class"], "rate_limited");
    assert_eq!(
        snapshot["data"]["classification"]["rule_id"],
        "claude.rate-limit.usage-limit"
    );

    let evidence = world
        .run_evidence(&world.run_id(1), "vendor_error_classified")
        .expect("vendor classification evidence");
    assert!(evidence["cooldown_until_ms"].as_i64().is_some());
    let encoded = evidence.to_string();
    assert!(!encoded.contains("Australia/Sydney"), "unsafe: {encoded}");

    let connection = rusqlite::Connection::open(world.db_path()).unwrap();
    let cooldowns: i64 = connection
        .query_row("SELECT COUNT(*) FROM cooldowns", [], |row| row.get(0))
        .unwrap();
    assert_eq!(cooldowns, 1, "the rejection must record a target cooldown");
    let status = world.sloop(&["status"]);
    assert_eq!(
        World::json_stdout(&status)["data"]["gate"]["cooldowns"][0]["target"],
        "fake"
    );
}

#[test]
fn cooldown_and_automatic_retry_survive_a_daemon_restart() {
    let world = World::configured();
    let first_run = world.root().join("first-run-finished");
    let body = format!(
        "if [ ! -e {marker} ]; then touch {marker}; printf \"You've hit your limit\\n\" >&2; exit 1; fi\n\
         echo recovered > work.txt\n\
         git add work.txt\n\
         git -c user.name=agent -c user.email=agent@example.invalid commit --quiet -m recovered\n",
        marker = first_run.display(),
    );
    configure(&world, &body, None);
    world.commit_all("initial");
    let daemon = world.start_daemon();
    let ticket = post_and_run(&world, "restart-cooldown.md");
    wait_until("the first run enters cooldown", || {
        tickets(&world)["ready"] == 1
            && world
                .run_evidence(&world.run_id(1), "vendor_error_classified")
                .is_some()
    });

    let pid = daemon["data"]["pid"].as_u64().unwrap() as u32;
    world.kill_daemon(pid);
    world.start_daemon();
    let connection = rusqlite::Connection::open(world.db_path()).unwrap();
    let run_count: i64 = connection
        .query_row("SELECT COUNT(*) FROM runs", [], |row| row.get(0))
        .unwrap();
    assert_eq!(run_count, 1, "the cooldown must gate restart dispatch");
    drop(connection);

    let listed = world.sloop(&["list"]);
    let rows = World::json_stdout(&listed);
    let row = rows["data"]["tickets"]
        .as_array()
        .unwrap()
        .iter()
        .find(|row| row["id"] == ticket)
        .unwrap();
    assert!(row["reason"].as_str().unwrap().contains("cooling down"));

    world.tick(Duration::from_secs(301));
    wait_until("the released trigger retries after cooldown", || {
        tickets(&world)["merged"] == 1
    });
    assert!(world.root().join("work.txt").is_file());
}

#[test]
fn a_recognized_rejection_with_commits_never_tests_or_merges_the_work() {
    let world = World::configured();
    let stage_marker = world.root().join("rejected-later-stage");
    configure(
        &world,
        concat!(
            "echo preserved > rejected-work.txt\n",
            "git add rejected-work.txt\n",
            "git -c user.name=agent -c user.email=agent@example.invalid commit --quiet -m preserved\n",
            "printf '%s\n' 'Unexpected server error. Check server logs for details.' >&2\n",
            "exit 0\n",
        ),
        Some(&format!("touch {}", stage_marker.display())),
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "preserve-rejected.md");

    wait_until("the rejected ticket is released under a cooldown", || {
        tickets(&world)["ready"] == 1
    });
    assert!(!stage_marker.exists());
    assert!(!world.root().join("rejected-work.txt").exists());
    assert_eq!(
        world
            .run_evidence(&world.run_id(1), "vendor_error_classified")
            .unwrap()["class"],
        "unknown_rejection"
    );
    assert_eq!(
        stage_runs(&world, 1)
            .into_iter()
            .map(|(_, stage, state, _)| (stage, state))
            .collect::<Vec<_>>(),
        [("build".into(), "failed".into())]
    );
    let shown = world.sloop_plain(&["show", &world.run_id(1)]);
    let text = String::from_utf8_lossy(&shown.stdout);
    assert!(text.contains("OpenCode rejected the request"), "{text}");
}

#[test]
fn a_target_cooldown_does_not_block_other_agent_targets() {
    let world = World::configured();
    fs::create_dir_all(world.root().join(".agents/sloop/flows")).unwrap();
    write_flow(
        &world,
        "stages:\n  - { name: build, action: agent }\n  - { name: merge, action: { builtin: merge } }\n",
    );
    let limited = world.root().join("limited.sh");
    let healthy = world.root().join("healthy.sh");
    fs::write(
        &limited,
        "#!/bin/sh\nprintf \"You've hit your limit\\n\" >&2\nexit 1\n",
    )
    .unwrap();
    fs::write(
        &healthy,
        concat!(
            "#!/bin/sh\n",
            "echo healthy > healthy.txt\n",
            "git add healthy.txt\n",
            "git -c user.name=agent -c user.email=agent@example.invalid commit --quiet -m healthy\n",
        ),
    )
    .unwrap();
    fs::write(
        world.root().join(".agents/sloop/config.yaml"),
        format!(
            "version: 1\nscheduler:\n  max_parallel_tasks: 1\nagent:\n  default_target: limited\n  targets:\n    limited:\n      cmd: [\"sh\", \"{}\", \"{{prompt}}\"]\n    healthy:\n      cmd: [\"sh\", \"{}\", \"{{prompt}}\"]\n",
            limited.display(),
            healthy.display(),
        ),
    )
    .unwrap();
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "limited.md");
    wait_until("the limited target cools down", || {
        tickets(&world)["ready"] == 1
    });

    let healthy_ticket = world.write_ticket(
        "healthy.md",
        "---\ntarget: healthy\n---\n# Healthy target\n",
    );
    let posted = world.sloop(&["post", healthy_ticket.to_str().unwrap(), "--manual"]);
    assert!(posted.status.success());
    let id = World::json_stdout(&posted)["data"]["ticket"]["id"]
        .as_str()
        .unwrap()
        .to_owned();
    assert!(world.sloop(&["run", &id]).status.success());

    wait_until("the healthy target runs during the other cooldown", || {
        tickets(&world)["merged"] == 1
    });
    assert!(world.root().join("healthy.txt").is_file());
}

#[test]
fn cancellation_after_rejection_checkpoint_wins_without_a_cooldown() {
    const HOOK: &str = "after-agent-exit-checkpoint";

    let world = World::configured();
    world.arm_test_hook(HOOK);
    configure(
        &world,
        "printf \"You've hit your limit\\n\" >&2\nexit 1\n",
        None,
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "cancel-rejection.md");
    wait_until("the rejection checkpoint is durable", || {
        world.test_hook_reached(HOOK)
            && world
                .run_evidence(&world.run_id(1), "vendor_error_classified")
                .is_some()
    });

    assert!(world.sloop(&["cancel", &world.run_id(1)]).status.success());
    world.release_test_hook(HOOK);
    wait_until("cancellation settles", || tickets(&world)["ready"] == 1);
    assert_eq!(
        world.wait_snapshot(&world.run_id(1))["data"]["state"],
        "cancelled"
    );
    let connection = rusqlite::Connection::open(world.db_path()).unwrap();
    let cooldowns: i64 = connection
        .query_row("SELECT COUNT(*) FROM cooldowns", [], |row| row.get(0))
        .unwrap();
    assert_eq!(cooldowns, 0);
}

#[test]
fn recovery_preserves_the_checkpointed_cooldown_deadline() {
    const HOOK: &str = "after-agent-exit-checkpoint";

    let world = World::configured();
    world.arm_test_hook(HOOK);
    configure(
        &world,
        "printf \"You've hit your limit\\n\" >&2\nexit 1\n",
        None,
    );
    world.commit_all("initial");
    let daemon = world.start_daemon();
    post_and_run(&world, "recover-deadline.md");
    wait_until("the rejection deadline is checkpointed", || {
        world.test_hook_reached(HOOK)
            && world
                .run_evidence(&world.run_id(1), "vendor_error_classified")
                .is_some()
    });
    let deadline = world
        .run_evidence(&world.run_id(1), "vendor_error_classified")
        .unwrap()["cooldown_until_ms"]
        .as_i64()
        .unwrap();

    world.kill_daemon(daemon["data"]["pid"].as_u64().unwrap() as u32);
    world.tick(Duration::from_secs(240));
    world.start_daemon();
    wait_until("recovery settles the rejected run", || {
        tickets(&world)["ready"] == 1
    });

    let connection = rusqlite::Connection::open(world.db_path()).unwrap();
    let persisted: i64 = connection
        .query_row("SELECT until_ms FROM cooldowns", [], |row| row.get(0))
        .unwrap();
    assert_eq!(persisted, deadline);
    let run_count: i64 = connection
        .query_row("SELECT COUNT(*) FROM runs", [], |row| row.get(0))
        .unwrap();
    assert_eq!(
        run_count, 1,
        "recovery must not dispatch before the deadline"
    );
}

/// An agent that commits on its first invocation and reports a verdict on
/// every later one, recording the worker token it was handed each time. Both
/// agent stages of a flow run this same script, so the files it leaves behind
/// say how many times it ran and under which credentials.
const REVIEWING_AGENT: &str = concat!(
    "root=\"$(dirname \"$0\")\"\n",
    "count=$(cat \"$root/agent-invocations\" 2>/dev/null || echo 0)\n",
    "count=$((count + 1))\n",
    "printf '%s' \"$count\" > \"$root/agent-invocations\"\n",
    "printf '%s\\n' \"$SLOOP_TOKEN\" >> \"$root/agent-tokens\"\n",
    "if [ \"$count\" = 1 ]; then\n",
    "  echo done > work.txt\n",
    "  git add work.txt\n",
    "  git -c user.name=agent -c user.email=agent@example.invalid commit --quiet -m 'agent work'\n",
    "else\n",
    "  echo reviewing\n",
    "  \"$SLOOP_BIN\" --json verdict pass --reason reviewed > /dev/null\n",
    "fi\n",
    "exit 0\n",
);

fn lines_of(path: PathBuf) -> Vec<String> {
    fs::read_to_string(path)
        .expect("read agent record")
        .lines()
        .map(str::to_owned)
        .collect()
}

/// Two agent stages in one flow: the driver runs each as its own supervised
/// process, and each gets credentials of its own. The review stage's reported
/// verdict is proof that its token authenticated for *its* stage — the first
/// agent's token could not have carried it, because the token it was issued is
/// no longer the run's.
#[test]
fn a_two_agent_flow_supervises_both_stages_with_separate_worker_tokens() {
    let world = World::configured();
    configure(&world, REVIEWING_AGENT, None);
    write_flow(
        &world,
        "stages:\n  - { name: build, action: agent }\n  - { name: review, action: agent, result_check: reported }\n  - { name: merge, action: { builtin: merge } }\n",
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "two-agents.md");

    // Two agent stages spawn a process each before the merge runs, so this waits
    // on more real work than the 10-second probe is meant for.
    wait_until_slow("the two-agent flow merges", || {
        tickets(&world)["merged"] == 1
    });
    assert!(default_branch_has(&world, "work.txt"));
    assert_eq!(
        fs::read_to_string(world.root().join("agent-invocations")).unwrap(),
        "2",
        "both agent stages must run"
    );

    let tokens = lines_of(world.root().join("agent-tokens"));
    assert_eq!(tokens.len(), 2, "one token per agent stage: {tokens:?}");
    assert!(tokens.iter().all(|token| !token.is_empty()));
    assert_ne!(
        tokens[0], tokens[1],
        "each agent stage is scoped to its own worker token"
    );

    assert_eq!(
        stage_runs(&world, 1)
            .into_iter()
            .map(|(_, name, state, _)| (name, state))
            .collect::<Vec<_>>(),
        [
            ("build".into(), "passed".into()),
            ("review".into(), "passed".into()),
            ("merge".into(), "passed".into()),
        ]
    );
    let review = stage_evidence(&world, 1, "review");
    assert_eq!(review["verdict_source"], "reported");
    assert_eq!(review["reason"], "reviewed");

    // The review agent's output is captured under its own stage, which only a
    // supervised process produces.
    let logs = world.sloop(&["logs", &world.run_id(1)]);
    let entries = World::json_stdout(&logs)["data"]["entries"]
        .as_array()
        .unwrap()
        .clone();
    assert!(
        entries
            .iter()
            .any(|entry| entry["source"] == "agent" && entry["stage"] == "review"),
        "no supervised review-stage output in {entries:?}"
    );
}

/// Like [`REVIEWING_AGENT`], but the reviewer saves the daemon's reply so the
/// test can wait on the report having *landed* rather than on the process
/// having started.
const RECORDING_REVIEWER: &str = concat!(
    "root=\"$(dirname \"$0\")\"\n",
    "count=$(cat \"$root/agent-invocations\" 2>/dev/null || echo 0)\n",
    "count=$((count + 1))\n",
    "printf '%s' \"$count\" > \"$root/agent-invocations\"\n",
    "if [ \"$count\" = 1 ]; then\n",
    "  echo done > work.txt\n",
    "  git add work.txt\n",
    "  git -c user.name=agent -c user.email=agent@example.invalid commit --quiet -m 'agent work'\n",
    "else\n",
    "  \"$SLOOP_BIN\" --json verdict pass --reason reviewed > \"$root/verdict.out\" 2>&1 || true\n",
    "fi\n",
    "exit 0\n",
);

/// A worker's verdict is credited to the stage its credential names, even when
/// it reports before the driver has checkpointed that stage's process.
///
/// The driver clears the `stage_process` row between stages and writes the next
/// one only after that stage's child is spawned, so for the width of the launch
/// window there is no row at all. A reviewer that reported inside it used to be
/// refused with `conflict` — and because a reviewer's report is fire-and-forget,
/// nothing surfaced the refusal: the `reported` check went on waiting for a
/// report that had already been thrown away, and the run sat unsettled until
/// its lease expired. The window is real but narrow, so this holds it open
/// rather than racing it — on Linux the daemon wins by default and the bug is
/// invisible, which is how it reached a release.
#[test]
fn a_verdict_reported_before_the_stage_checkpoint_lands_on_the_reporting_stage() {
    let world = World::configured();
    configure(&world, RECORDING_REVIEWER, None);
    write_flow(
        &world,
        "stages:\n  - { name: build, action: agent }\n  - { name: review, action: agent, result_check: reported }\n  - { name: merge, action: { builtin: merge } }\n",
    );
    world.commit_all("initial");
    // Fires after the review child is spawned and before its `stage_process`
    // row is written, so the reviewer reports while the row still says `build`.
    world.arm_test_hook("before-stage-process-checkpoint-review");
    world.start_daemon();
    post_and_run(&world, "checkpoint-window.md");

    wait_until_slow("the reviewer reports inside the launch window", || {
        fs::read_to_string(world.root().join("verdict.out"))
            .ok()
            .is_some_and(|text| serde_json::from_str::<serde_json::Value>(&text).is_ok())
    });
    let reply: serde_json::Value =
        serde_json::from_str(&fs::read_to_string(world.root().join("verdict.out")).unwrap())
            .unwrap();
    assert_eq!(
        reply["ok"], true,
        "the reviewer's report was refused: {reply}"
    );

    world.release_test_hook("before-stage-process-checkpoint-review");

    wait_until_slow("the flow merges after the delayed checkpoint", || {
        tickets(&world)["merged"] == 1
    });
    let review = stage_evidence(&world, 1, "review");
    assert_eq!(
        review["verdict_source"], "reported",
        "the report must be credited to `review`, not to the stage the \
         checkpoint still named"
    );
    assert_eq!(review["reason"], "reviewed");
}

/// An agent stage need not open a flow. One that follows an exec stage is
/// launched by the same driver loop, in the same worktree, with a worker token
/// that authenticates — which the note it records proves.
#[test]
fn an_agent_stage_after_an_exec_stage_runs_supervised_with_its_own_token() {
    let world = World::configured();
    let prepared = world.root().join("prepared");
    let agent = format!(
        "test -f {prepared}\n\"$SLOOP_BIN\" --json note 'agent reporting' > /dev/null\n{COMMITTING_AGENT}",
        prepared = prepared.display(),
    );
    configure(&world, &agent, None);
    write_flow(
        &world,
        &format!(
            "stages:\n  - {{ name: prepare, action: {{ exec: [\"sh\", \"-c\", \"printf ready > {}\"] }} }}\n  - {{ name: build, action: agent }}\n  - {{ name: merge, action: {{ builtin: merge }} }}\n",
            prepared.display(),
        ),
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "agent-second.md");

    wait_until("the flow with a trailing agent merges", || {
        tickets(&world)["merged"] == 1
    });
    assert!(default_branch_has(&world, "work.txt"));
    assert_eq!(
        stage_runs(&world, 1)
            .into_iter()
            .map(|(index, name, state, _)| (index, name, state))
            .collect::<Vec<_>>(),
        [
            (0, "prepare".into(), "passed".into()),
            (1, "build".into(), "passed".into()),
            (2, "merge".into(), "passed".into()),
        ]
    );
    // The note only lands if the agent's token authenticated on its run's
    // worker socket.
    assert_eq!(world.run_note_count(&world.run_id(1)), 1);
}

/// Resumption does not depend on where the walk stood. A daemon killed while
/// the *first* stage of a flow was executing — a position that used to belong
/// to the dispatcher rather than to any resumable regime — comes back, replays
/// the log, and finishes the flow.
#[test]
fn restart_during_the_first_stage_resumes_the_flow() {
    const HOOK: &str = "after-stage-process-checkpoint-open";

    let world = World::configured();
    configure(&world, COMMITTING_AGENT, None);
    world.arm_test_hook(HOOK);
    let invocations = world.root().join("open-invocations");
    write_flow(
        &world,
        &format!(
            "stages:\n  - {{ name: open, action: {{ exec: [\"sh\", \"-c\", \"printf x >> {}\"] }} }}\n  - {{ name: build, action: agent }}\n  - {{ name: merge, action: {{ builtin: merge }} }}\n",
            invocations.display(),
        ),
    );
    world.commit_all("initial");
    let daemon_pid = world.start_daemon()["data"]["pid"].as_u64().unwrap() as u32;
    post_and_run(&world, "restart-first-stage.md");
    wait_until("the first stage reaches its checkpoint gate", || {
        world.test_hook_reached(HOOK)
    });
    assert_eq!(world.run_state(&world.run_id(1)), "driving");

    world.kill_daemon(daemon_pid);
    world.release_test_hook(HOOK);
    world.start_daemon();

    wait_until("the resumed driver finishes the flow", || {
        tickets(&world)["merged"] == 1
    });
    assert!(default_branch_has(&world, "work.txt"));
    assert_eq!(
        stage_runs(&world, 1)
            .into_iter()
            .map(|(_, name, state, _)| (name, state))
            .collect::<Vec<_>>(),
        [
            ("open".into(), "passed".into()),
            ("build".into(), "passed".into()),
            ("merge".into(), "passed".into()),
        ]
    );
}

/// Cancellation kills whatever stage is executing, wherever it sits. Here the
/// agent is the *second* stage, so nothing about the kill can lean on the agent
/// being the run's opening move.
#[test]
fn cancel_kills_an_agent_stage_process_group_in_a_later_position() {
    let world = World::configured();
    let process_ids = world.root().join("late-agent-processes");
    let agent = format!(
        "sleep 1000 & printf '%s %s' $$ $! > {process_ids}; wait\n",
        process_ids = process_ids.display(),
    );
    configure(&world, &agent, None);
    write_flow(
        &world,
        "stages:\n  - { name: prepare, action: { exec: [\"true\"] } }\n  - { name: build, action: agent }\n  - { name: merge, action: { builtin: merge } }\n",
    );
    world.commit_all("initial");
    world.start_daemon();
    post_and_run(&world, "cancel-late-agent.md");

    wait_until("the later agent stage is executing", || {
        process_ids.is_file()
            && world
                .run_evidence(&world.run_id(1), "stage_process")
                .is_some_and(|evidence| evidence["stage"] == "build")
    });
    let process_ids = read_process_ids(process_ids);

    let cancelled = world.sloop(&["cancel", &world.run_id(1)]);
    assert!(
        cancelled.status.success(),
        "cancel failed: {}",
        String::from_utf8_lossy(&cancelled.stderr)
    );
    wait_until("the cancelled run settles", || {
        let counts = tickets(&world);
        counts["ready"] == 1 && counts["claimed"] == 0
    });
    assert_eq!(
        world.wait_snapshot(&world.run_id(1))["data"]["state"],
        "cancelled"
    );
    wait_for_processes_to_exit(process_ids);
}