//! The two engines composed — Slice 4.
//!
//! Engine Z (the ZeroClaw graph in `cm_orchestrator`) owns durability and
//! heterogeneity: deterministic planners, per-step checkpoint/resume, a stale-run
//! sweep, cancellation, and a different model per node. Engine C (Claude Code in
//! a microVM) owns shared context, self-sizing and cheap fan-out. Neither has the
//! other's asset, which is why keeping both is a composition rather than a
//! compromise.
//!
//! This module is the join: a [`TurnExecutor`] whose "turn" is a whole
//! Claude-Code-in-a-VM session. Because `topology_worker` already dispatches by
//! tier, implementing the existing trait inherits the planners, checkpointing,
//! reaper, cancellation, `close_finished_phases`, evaluation, capture and
//! delivery unchanged. `recursive_exec::SubTopologyExecutor` is the precedent: a
//! `run_turn` may be arbitrarily heavy.
//!
//! # The file-handoff trap
//!
//! A VM is inject-tar → run → collect-tar → destroy. A graph of per-node VMs with
//! **text-only** handoff would silently lose every file an earlier node wrote:
//! node 2 would boot from the original checkout, see none of node 1's work, and
//! still report success — the exact silent-success shape this project keeps
//! paying for.
//!
//! The answer here is that the mission's **host checkout is the medium**. Every
//! node injects from `repo` and collects back over `repo`, so the tree carries
//! forward node to node and the last node's tree is what delivery diffs. Two
//! properties make that safe rather than lucky:
//!
//! - `execute_resumable` runs steps strictly **sequentially**, so two VMs are
//!   never writing the same host directory at once;
//! - the vm id is deterministic per (phase, iteration, step), so a resumed step
//!   whose VM is somehow still alive is refused by the node ("vm already exists")
//!   instead of quietly producing a second writer.
//!
//! `a_later_node_sees_an_earlier_nodes_files` proves the handoff, and
//! `text_only_handoff_loses_the_earlier_nodes_work` is its negative control.
//!
//! # Keeping a long turn alive
//!
//! `requeue_stale` requeues a `running` job that has not touched `updated_at` in
//! 180 seconds, and one node here can run for an hour. `SubTopologyExecutor`
//! keeps its parent alive from each *leaf step*, which it has and this does not:
//! there is nothing between the start and end of a VM turn. So the turn holds a
//! ticker that touches `updated_at` every [`KEEPALIVE_SECS`] and is aborted on
//! drop. Without it a healthy composed run is requeued mid-node, claimed again,
//! and boots a second VM against the same checkout.

use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use cm_domain::NodeId;
use cm_orchestrator::{OrchestratorError, TurnExecutor, TurnOutcome, TurnRequest};
use sqlx::PgPool;
use uuid::Uuid;

use crate::microvm_executor::{PhaseVm, VmPhase};

/// How often a running VM turn touches its run's `updated_at`.
///
/// Comfortably inside the 180s stale window, and cheap: one UPDATE per node per
/// half minute against a row nothing else is writing.
const KEEPALIVE_SECS: u64 = 30;

/// A [`TurnExecutor`] that runs each graph node as a full Claude-Code session
/// inside its own microVM, against the mission's shared host checkout.
pub struct MicroVmTurnExecutor<V: PhaseVm> {
    vms: V,
    pool: PgPool,
    /// The durable outer run. Touched for keepalive; its status gates the turn.
    run_id: Uuid,
    mission_id: Uuid,
    phase_id: Uuid,
    iteration: i32,
    /// The mission's host checkout — injected into every node's VM and collected
    /// back over, which is how file work survives a node boundary.
    repo: PathBuf,
    /// Whether the mission has a repository. Carried so every graph node gets
    /// the same workspace treatment as a solo phase — see `VmPhase::has_repo`.
    has_repo: bool,
    /// `missions.target_node_id`: the fleet node a mission was placed on. A node
    /// may override it with `attrs["node_id"]`.
    default_fleet_node: Option<Uuid>,
    /// `missions.backend`: which rootfs image. A node may override it with
    /// `attrs["backend"]`, which is what makes a graph heterogeneous — a
    /// `validator` node on a different provider's image is then a first-class
    /// graph node rather than a bolt-on.
    default_backend: Option<String>,
    /// `missions.team_engine`, passed through so a composed node can itself ask
    /// for Claude Code fan-out inside its VM.
    team_engine: Option<String>,
    /// The phase's completion gate, enforced inside every node's VM.
    gate: Option<crate::vm_stop_gate::StopGate>,
    /// Which step is next. `execute_resumable` is sequential and gives the
    /// executor no index, so the executor counts — and the count starts from the
    /// checkpoint on resume, or two VMs would share an id across a restart.
    step: std::sync::atomic::AtomicU32,
}

/// Everything a composed run needs that is not the graph itself.
pub struct ComposedRun {
    pub run_id: Uuid,
    pub mission_id: Uuid,
    pub phase_id: Uuid,
    pub iteration: i32,
    pub repo: PathBuf,
    pub has_repo: bool,
    pub target_node_id: Option<Uuid>,
    pub backend: Option<String>,
    pub team_engine: Option<String>,
    /// What must hold before a node's agent may stop. See [`crate::vm_stop_gate`].
    pub gate: Option<crate::vm_stop_gate::StopGate>,
    /// Steps already completed, from the durable checkpoint. Nonzero on resume.
    pub completed_steps: u32,
}

impl<V: PhaseVm> MicroVmTurnExecutor<V> {
    pub fn new(vms: V, pool: PgPool, r: ComposedRun) -> Self {
        Self {
            vms,
            pool,
            run_id: r.run_id,
            mission_id: r.mission_id,
            phase_id: r.phase_id,
            iteration: r.iteration,
            repo: r.repo,
            has_repo: r.has_repo,
            default_fleet_node: r.target_node_id,
            default_backend: r.backend,
            team_engine: r.team_engine,
            gate: r.gate,
            step: std::sync::atomic::AtomicU32::new(r.completed_steps),
        }
    }

    /// Which fleet node this graph node runs on.
    ///
    /// Fail-closed on a malformed override: placing a node on the mission's node
    /// because its own `node_id` did not parse would run the work somewhere the
    /// graph did not ask for and say nothing.
    fn fleet_node(&self, req: &TurnRequest) -> Result<NodeId, OrchestratorError> {
        let id = match req.attrs.get("node_id") {
            Some(raw) => Uuid::parse_str(raw.trim()).map_err(|_| {
                OrchestratorError::Executor(format!(
                    "node {} has an invalid node_id attr: {raw}",
                    req.node_id
                ))
            })?,
            None => self.default_fleet_node.ok_or_else(|| {
                OrchestratorError::Executor(format!(
                    "node {} has no node_id attr and the mission has no \
                     target_node_id — a microVM node cannot run on the gateway, \
                     which has no /dev/kvm",
                    req.node_id
                ))
            })?,
        };
        Ok(NodeId::from(id))
    }
}

impl<V: PhaseVm> TurnExecutor for MicroVmTurnExecutor<V> {
    async fn run_turn(&self, req: TurnRequest) -> Result<TurnOutcome, OrchestratorError> {
        let fleet_node = self.fleet_node(&req)?;
        let backend = req
            .attrs
            .get("backend")
            .map(|s| s.trim().to_string())
            .filter(|s| !s.is_empty())
            .or_else(|| self.default_backend.clone());

        if !self.repo.is_dir() {
            return Err(OrchestratorError::Executor(format!(
                "mission has no checkout at {} — a composed node needs the \
                 repository, and it is also how the previous node's work reaches \
                 this one",
                self.repo.display()
            )));
        }

        let step = self
            .step
            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);

        // Held for the length of the VM turn: an hour of silence would otherwise
        // look exactly like a dead worker to `requeue_stale`.
        let _alive = Keepalive::spawn(self.pool.clone(), self.run_id);

        let task = node_task_text(&req);
        let outcome = self
            .vms
            .run(VmPhase {
                task_policy: None,
                // Every node of a composed graph streams to the same outer run,
                // which is the one the operator is watching.
                run_id: Some(self.run_id),
                node_id: fleet_node,
                mission_id: self.mission_id,
                phase_id: self.phase_id,
                iteration: self.iteration,
                task: &task,
                backend: backend.as_deref(),
                repo: &self.repo,
                has_repo: self.has_repo,
                team_engine: self.team_engine.as_deref(),
                // Each node is its own agent session, so each carries the
                // phase's gate. Threaded from the run rather than rebuilt here:
                // one source for what "done" means, whichever executor asks.
                gate: self.gate.as_ref(),
                step: Some(step),
                // Same live drain as the solo path. A composed graph node can
                // run for an hour too, and its files are the only account of
                // what it did until the next node collects.
                tap_sink: Some(crate::phase_runner::vm_tool_recorder(
                    &self.pool,
                    self.mission_id,
                    self.phase_id,
                    self.run_id,
                )),
            })
            .await
            .map_err(|e| {
                OrchestratorError::Executor(format!("node {} in a microVM: {e}", req.node_id))
            })?;

        // Recorded BEFORE the failure branches below. A node that could not be
        // collected, or whose gate capped, still touched files — and on this
        // path those touches are the only account of what it did, since the
        // work never reached a diff.
        crate::phase_runner::record_vm_tools(
            &self.pool,
            self.mission_id,
            self.phase_id,
            self.run_id,
            &outcome.tools,
            // No turn agents supplied, so nothing is attributed — the same
            // `agent_id: None` this path has always written. Resolving the
            // graph node to an agent uuid is the fix, and it cannot be tested
            // while the fleet is offline; guessing at it here would put one
            // node's actions on another node's record.
            &[],
        )
        .await;

        // A node whose work never came back must fail the run rather than hand
        // the next node a tree missing the previous one's edits. On this path an
        // uncollected turn is worse than on the solo one: the loss is silent,
        // because the next node still boots from a checkout that looks fine.
        if !outcome.collected {
            return Err(OrchestratorError::Executor(format!(
                "node {}'s work could not be collected from its VM, so the next \
                 node would not see it: {}",
                req.node_id,
                outcome.summary.chars().take(400).collect::<String>()
            )));
        }
        // Same rule as the solo path: the gate is the only thing that runs a
        // `done_when_check`, so a release at the cap must fail the run rather
        // than hand the next node a tree that does not satisfy the condition
        // every node in this graph was told to satisfy.
        if outcome.released_at_cap == Some(true) {
            return Err(OrchestratorError::Executor(format!(
                "node {}'s completion gate released it after {} refusal(s) with its check \
                 still failing: {}",
                req.node_id,
                crate::vm_stop_gate::MAX_BLOCKS,
                outcome.summary.chars().take(400).collect::<String>()
            )));
        }
        if outcome.rc != 0 {
            return Err(OrchestratorError::Executor(format!(
                "node {} exited {}: {}",
                req.node_id,
                outcome.rc,
                outcome.summary.chars().take(400).collect::<String>()
            )));
        }

        eprintln!(
            "microvm_turn_executor: run {} node {} (role {}, step {}) ok — subagents: {}",
            self.run_id,
            req.node_id,
            req.role,
            step,
            outcome
                .subagents
                .map(|n| n.to_string())
                .unwrap_or_else(|| "?".into()),
        );

        Ok(TurnOutcome {
            output: outcome.summary,
            // `claude -p` does not report token usage on stdout, and inventing a
            // number here would corrupt the run totals the harness reads. Zero is
            // the honest value for "not measured on this path".
            tokens: 0,
            gated: Vec::new(),
            spend: Default::default(),
        })
    }
}

/// What one graph node is told.
///
/// The upstream outputs are included as context, but the load-bearing sentence is
/// that the previous node's *files* are already in the tree: a node told only
/// about the text would re-do work it is standing on.
fn node_task_text(req: &TurnRequest) -> String {
    let mut s = format!(
        "You are the `{}` stage of a multi-stage mission.\n\nMISSION TASK\n{}\n",
        req.role, req.task
    );
    if !req.context.is_empty() {
        s.push_str(
            "\nWHAT CAME BEFORE\nThe earlier stages' work is ALREADY IN THIS \
             WORKING TREE — the repository you have been given is their output, \
             not a fresh checkout. Read the files before changing them, and do \
             not redo what is already done. Their closing reports:\n",
        );
        for (i, c) in req.context.iter().enumerate() {
            s.push_str(&format!("\n--- stage {} ---\n{}\n", i + 1, c));
        }
    }
    s
}

/// Touches a run's `updated_at` until dropped.
struct Keepalive(tokio::task::JoinHandle<()>);

impl Keepalive {
    fn spawn(pool: PgPool, run_id: Uuid) -> Self {
        Keepalive(tokio::spawn(async move {
            let mut ticker = tokio::time::interval(Duration::from_secs(KEEPALIVE_SECS));
            loop {
                ticker.tick().await;
                let _ = cm_db::repo::topology_runs::touch(&pool, run_id).await;
            }
        }))
    }
}

impl Drop for Keepalive {
    fn drop(&mut self) {
        self.0.abort();
    }
}

/// Build the executor the worker uses, over real VMs on the fleet.
pub fn for_fleet(
    hub: Arc<crate::fleet::NodeHub>,
    pool: PgPool,
    r: ComposedRun,
) -> MicroVmTurnExecutor<crate::microvm_executor::HubVms> {
    MicroVmTurnExecutor::new(crate::microvm_executor::HubVms::new(hub), pool, r)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::microvm_executor::VmOutcome;
    use std::collections::BTreeMap;
    use std::sync::Mutex;

    /// A VM modelled honestly: the host tree is packed in, the "agent" works on a
    /// COPY that no host path points at, and the result is unpacked back over the
    /// host tree. That is the real inject → run → collect shape, which is what
    /// makes the negative control below meaningful — remove the collect and the
    /// handoff breaks exactly as it would in production.
    struct FakeVms {
        /// Whether the guest's tree is collected back to the host.
        collect: bool,
        /// vm ids used, in order — the id is what stops two nodes colliding.
        ids: Mutex<Vec<String>>,
        /// (backend, fleet node) per call, for the heterogeneity assertions.
        placements: Mutex<Vec<(Option<String>, NodeId)>>,
    }

    impl FakeVms {
        fn new(collect: bool) -> Self {
            Self {
                collect,
                ids: Mutex::new(Vec::new()),
                placements: Mutex::new(Vec::new()),
            }
        }
    }

    impl PhaseVm for FakeVms {
        async fn run(&self, p: VmPhase<'_>) -> Result<VmOutcome, String> {
            self.ids
                .lock()
                .unwrap()
                .push(format!("{}-{:?}", p.phase_id.simple(), p.step));
            self.placements
                .lock()
                .unwrap()
                .push((p.backend.map(str::to_string), p.node_id));

            // inject: the host checkout goes in as a tar.
            let tar = crate::mission_fs::pack_dir(p.repo, "repo")?;
            let guest = tempfile::tempdir().map_err(|e| e.to_string())?;
            crate::mission_fs::unpack_into(&tar, guest.path())?;
            let guest_repo = guest.path().join("repo");

            // run: the agent records that it was here, and reports what it found
            // of the previous stages — the observation the handoff test reads.
            let seen: Vec<String> = std::fs::read_dir(&guest_repo)
                .map_err(|e| e.to_string())?
                .filter_map(|e| e.ok())
                .map(|e| e.file_name().to_string_lossy().to_string())
                .filter(|n| n.starts_with("stage-"))
                .collect();
            let mine = guest_repo.join(format!("stage-{}.txt", p.step.unwrap_or(0)));
            std::fs::write(&mine, "work").map_err(|e| e.to_string())?;

            // collect: the guest tree comes back over the same host path.
            if self.collect {
                let back = crate::mission_fs::pack_dir(&guest_repo, "repo")?;
                let parent = p.repo.parent().ok_or("no parent")?;
                crate::mission_fs::unpack_into(&back, parent)?;
            }

            Ok(VmOutcome {
                summary: format!("saw:[{}]", seen.join(",")),
                rc: 0,
                collected: true,
                subagents: Some(0),
                teammates: None,
                stop_blocks: None,
                released_at_cap: None,
                tools: Vec::new(),
            rootfs: None,
            cli_version: None,
            tool_gate: None,
            })
        }
    }

    fn req(node: &str, role: &str, context: Vec<String>) -> TurnRequest {
        TurnRequest {
            node_id: node.into(),
            role: role.into(),
            agent: None,
            attrs: BTreeMap::new(),
            task: "build the thing".into(),
            context,
        }
    }

    fn exec<V: PhaseVm>(vms: V, repo: PathBuf) -> MicroVmTurnExecutor<V> {
        // A pool that is never connected: every test here fails the turn before
        // any query, or drives one whose only DB touch is the best-effort
        // keepalive (which swallows its own errors by design).
        let pool = sqlx::postgres::PgPoolOptions::new()
            .max_connections(1)
            .connect_lazy("postgres://invalid/invalid")
            .expect("a lazy pool never dials");
        MicroVmTurnExecutor::new(
            vms,
            pool,
            ComposedRun {
                run_id: Uuid::now_v7(),
                mission_id: Uuid::now_v7(),
                phase_id: Uuid::now_v7(),
                iteration: 1,
                repo,
                has_repo: true,
                target_node_id: Some(Uuid::now_v7()),
                backend: Some("claude".into()),
                team_engine: None,
                gate: None,
                completed_steps: 0,
            },
        )
    }

    fn a_checkout() -> tempfile::TempDir {
        let d = tempfile::tempdir().unwrap();
        std::fs::create_dir_all(d.path().join("repo")).unwrap();
        std::fs::write(d.path().join("repo").join("README.md"), "hello").unwrap();
        d
    }

    /// THE trap this slice exists to solve. A per-node VM is destroyed with its
    /// filesystem, so unless the tree is carried forward, node 2 works from the
    /// original checkout and silently loses node 1's edits — while still
    /// reporting success.
    #[tokio::test]
    async fn a_later_node_sees_an_earlier_nodes_files() {
        let d = a_checkout();
        let e = exec(FakeVms::new(true), d.path().join("repo"));

        let first = e.run_turn(req("n1", "implementer", vec![])).await.unwrap();
        assert_eq!(first.output, "saw:[]", "the first node starts clean");

        let second = e
            .run_turn(req("n2", "verifier", vec![first.output.clone()]))
            .await
            .unwrap();
        assert!(
            second.output.contains("stage-0.txt"),
            "node 2 could not see node 1's file: {}",
            second.output
        );
        // And the host tree — what delivery diffs — holds both nodes' work.
        for f in ["stage-0.txt", "stage-1.txt"] {
            assert!(d.path().join("repo").join(f).exists(), "{f} missing on the host");
        }
    }

    /// The negative control, run rather than assumed: with the collect removed —
    /// i.e. a text-only handoff between nodes — the test above fails. A guard
    /// that cannot detect the bug it was written for is decoration.
    #[tokio::test]
    async fn text_only_handoff_loses_the_earlier_nodes_work() {
        let d = a_checkout();
        let e = exec(FakeVms::new(false), d.path().join("repo"));

        e.run_turn(req("n1", "implementer", vec![])).await.unwrap();
        let second = e.run_turn(req("n2", "verifier", vec![])).await.unwrap();
        assert_eq!(
            second.output, "saw:[]",
            "without a collect, node 2 must NOT see node 1's work — if it does, \
             this test is no longer controlling anything"
        );
        assert!(!d.path().join("repo").join("stage-0.txt").exists());
    }

    /// Each node gets its own vm id within one phase and iteration. Two nodes
    /// sharing an id means the second is refused by the fleet node while the
    /// first is alive, and indistinguishable from a re-run once it is not.
    #[tokio::test]
    async fn every_node_runs_in_its_own_vm() {
        let d = a_checkout();
        let vms = FakeVms::new(true);
        let e = exec(vms, d.path().join("repo"));
        for n in ["n1", "n2", "n3"] {
            e.run_turn(req(n, "worker", vec![])).await.unwrap();
        }
        let ids = e.vms.ids.lock().unwrap().clone();
        let unique: std::collections::HashSet<_> = ids.iter().collect();
        assert_eq!(unique.len(), ids.len(), "{ids:?}");
    }

    /// Resume must not re-use a completed step's vm id. The executor counts steps
    /// itself, so the count has to start where the checkpoint left off.
    #[tokio::test]
    async fn a_resumed_run_continues_the_step_numbering() {
        let d = a_checkout();
        let pool = sqlx::postgres::PgPoolOptions::new()
            .max_connections(1)
            .connect_lazy("postgres://invalid/invalid")
            .unwrap();
        let e = MicroVmTurnExecutor::new(
            FakeVms::new(true),
            pool,
            ComposedRun {
                run_id: Uuid::now_v7(),
                mission_id: Uuid::now_v7(),
                phase_id: Uuid::now_v7(),
                iteration: 1,
                repo: d.path().join("repo"),
                has_repo: true,
                target_node_id: Some(Uuid::now_v7()),
                backend: None,
                team_engine: None,
                gate: None,
                completed_steps: 2,
            },
        );
        e.run_turn(req("n3", "worker", vec![])).await.unwrap();
        let ids = e.vms.ids.lock().unwrap().clone();
        assert!(
            ids[0].ends_with("Some(2)"),
            "the first step after a resume must be step 2, not 0: {ids:?}"
        );
    }

    /// Per-node `backend` is what makes the outer graph heterogeneous — a
    /// validator node on another provider's image. It must override the
    /// mission's, and the mission's must still apply to nodes that say nothing.
    #[tokio::test]
    async fn a_node_may_pick_its_own_backend_and_fleet_node() {
        let d = a_checkout();
        let e = exec(FakeVms::new(true), d.path().join("repo"));
        let elsewhere = Uuid::now_v7();

        let mut r = req("n1", "worker", vec![]);
        r.attrs.insert("backend".into(), "glm".into());
        r.attrs.insert("node_id".into(), elsewhere.to_string());
        e.run_turn(r).await.unwrap();
        e.run_turn(req("n2", "worker", vec![])).await.unwrap();

        let p = e.vms.placements.lock().unwrap().clone();
        assert_eq!(p[0].0.as_deref(), Some("glm"));
        assert_eq!(p[0].1, NodeId::from(elsewhere));
        assert_eq!(p[1].0.as_deref(), Some("claude"), "the mission default");
        assert_ne!(p[1].1, NodeId::from(elsewhere));
    }

    /// A malformed `node_id` must fail the node, not fall back to the mission's.
    /// Silently running work somewhere the graph did not ask for is the same
    /// class of bug as an alias that serde dropped.
    #[tokio::test]
    async fn a_malformed_node_placement_fails_closed() {
        let d = a_checkout();
        let e = exec(FakeVms::new(true), d.path().join("repo"));
        let mut r = req("n1", "worker", vec![]);
        r.attrs.insert("node_id".into(), "not-a-uuid".into());
        let err = e.run_turn(r).await.unwrap_err().to_string();
        assert!(err.contains("invalid node_id"), "{err}");
    }

    /// An uncollected node is a failed run here, not a warning: the next node
    /// would boot from a tree that looks fine and is missing this node's work.
    #[tokio::test]
    async fn an_uncollected_node_fails_the_run() {
        struct Lost;
        impl PhaseVm for Lost {
            async fn run(&self, _p: VmPhase<'_>) -> Result<VmOutcome, String> {
                Ok(VmOutcome {
                    summary: "did plenty".into(),
                    rc: 0,
                    collected: false,
                    subagents: None,
                    teammates: None,
                    stop_blocks: None,
                    released_at_cap: None,
                    tools: Vec::new(),
            rootfs: None,
            cli_version: None,
            tool_gate: None,
                })
            }
        }
        let d = a_checkout();
        let e = exec(Lost, d.path().join("repo"));
        let err = e.run_turn(req("n1", "worker", vec![])).await.unwrap_err().to_string();
        assert!(err.contains("could not be collected"), "{err}");
    }

    /// A node whose gate gave up is a FAILED run, not a completed one.
    ///
    /// The gate is the only thing in the system that ever runs a
    /// `done_when_check`. If it releases the agent at the cap and this returns
    /// Ok, the check's failure is never seen again: the node reports success,
    /// the next node builds on a tree that does not satisfy the condition, and
    /// the phase completes green. `rc` is 0 and the work IS collected here on
    /// purpose — those are the two signals that used to decide this, and both
    /// say "fine".
    #[tokio::test]
    async fn a_node_whose_gate_gave_up_fails_the_run() {
        struct Capped;
        impl PhaseVm for Capped {
            async fn run(&self, _p: VmPhase<'_>) -> Result<VmOutcome, String> {
                Ok(VmOutcome {
                    summary: "I could not get the tests passing, but here is what I did".into(),
                    rc: 0,
                    collected: true,
                    subagents: None,
                    teammates: None,
                    stop_blocks: Some(crate::vm_stop_gate::MAX_BLOCKS),
                    released_at_cap: Some(true),
                    tools: Vec::new(),
            rootfs: None,
            cli_version: None,
            tool_gate: None,
                })
            }
        }
        let d = a_checkout();
        let e = exec(Capped, d.path().join("repo"));
        let err = e.run_turn(req("n1", "worker", vec![])).await.unwrap_err().to_string();
        assert!(err.contains("released it after"), "{err}");
    }

    /// The negative control: the SAME number of blocks, without the cap. An
    /// agent that was refused three times and then got it right on the fourth
    /// try has succeeded, and reports `blocks: 3` exactly like the test above.
    /// Failing on the count instead of the mark would fail this healthy run.
    #[tokio::test]
    async fn a_node_that_was_blocked_and_then_succeeded_passes() {
        struct Recovered;
        impl PhaseVm for Recovered {
            async fn run(&self, _p: VmPhase<'_>) -> Result<VmOutcome, String> {
                Ok(VmOutcome {
                    summary: "took me a few tries".into(),
                    rc: 0,
                    collected: true,
                    subagents: None,
                    teammates: None,
                    stop_blocks: Some(crate::vm_stop_gate::MAX_BLOCKS),
                    released_at_cap: Some(false),
                    tools: Vec::new(),
            rootfs: None,
            cli_version: None,
            tool_gate: None,
                })
            }
        }
        let d = a_checkout();
        let e = exec(Recovered, d.path().join("repo"));
        e.run_turn(req("n1", "worker", vec![]))
            .await
            .expect("a run that recovered inside its own turn is a success");
    }

    /// A node must be told its predecessors' files are already in the tree.
    /// Given only the text, an agent re-does work it is standing on.
    #[test]
    fn a_downstream_node_is_told_the_work_is_already_in_the_tree() {
        let solo = node_task_text(&req("n1", "implementer", vec![]));
        assert!(solo.contains("build the thing"));
        assert!(!solo.contains("WHAT CAME BEFORE"), "{solo}");

        let later = node_task_text(&req("n2", "verifier", vec!["I wrote foo.rs".into()]));
        assert!(later.contains("ALREADY IN THIS WORKING TREE"), "{later}");
        assert!(later.contains("I wrote foo.rs"), "{later}");
        assert!(later.contains("verifier"), "the node's role: {later}");
    }
}