//! Mission phase execution runner.
//!
//! Scans `mission_phases` where status='pending' AND the parent
//! mission is 'running'. Only advances a phase when all lower-order
//! phases have completed — enforces the research → coding → benchmark
//! sequence encoded in `mission_phases.order_idx`.
//!
//! For each eligible phase, enqueues one `topology_runs` row per team
//! whose (mission_id, purpose) matches the phase kind:
//!
//!   phase kind='research'      → teams with purpose='research'
//!   phase kind='coding'        → teams with purpose='coding'
//!   phase kind='benchmark'     → teams with purpose='coding' (fallback)
//!   phase kind='security_scan' → teams with purpose='security' or 'coding'
//!
//! Each run's task text combines the mission description + a phase-
//! kind-specific directive so the coordinator claw knows what to do.
//! The topology_worker (existing) picks up queued runs and drives
//! them through the ZeroClaw executor.
//!
//! Post-completion: when every topology_run bound to a phase is
//! terminal, the phase flips to 'completed' (or 'failed' if any run
//! failed). When every phase is terminal, the mission flips to
//! 'completed' (or 'failed').
//!
//! Cadence: 10s poll. Deliberately generous — every state transition
//! is idempotent and cheap.

use sqlx::PgPool;
use sqlx::Row;
use std::time::Duration;
use uuid::Uuid;

const POLL_INTERVAL: Duration = Duration::from_secs(10);

/// `runtime` is needed only by the completion evaluator; phases without a
/// `done_when` never touch it. `hub` is needed only by microVM missions, which
/// execute on a fleet node rather than in a container here.
pub fn spawn(
    pool: PgPool,
    runtime: cm_runtime::Runtime,
    hub: std::sync::Arc<crate::fleet::NodeHub>,
) {
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_secs(5)).await;
        let mut ticker = tokio::time::interval(POLL_INTERVAL);
        ticker.tick().await;
        loop {
            ticker.tick().await;
            if let Err(e) = sweep_once(&pool, &runtime, &hub).await {
                eprintln!("phase_runner: sweep failed: {e}");
            }
        }
    });
}

async fn sweep_once(
    pool: &PgPool,
    runtime: &cm_runtime::Runtime,
    hub: &std::sync::Arc<crate::fleet::NodeHub>,
) -> Result<(), String> {
    start_pending_phases(pool, hub).await?;
    close_finished_phases(pool).await?;
    // Between "all runs finished" and "phase done" sits the completion
    // evaluation, for phases that declare a condition.
    evaluate_finished_phases(pool, runtime).await?;
    // Capture before the mission closes and long before the sweeper reaps the
    // checkout. Idempotent, so a failure here is retried on the next tick
    // rather than losing the phase's work.
    capture_finished_coding_phases(pool).await?;
    // The same, for missions with no repository to diff. Without this the
    // container holding a research phase's only output is reaped unread.
    crate::mission_outputs::capture_repo_less_phases(pool).await?;
    // Benchmark phases record their baseline once the work exists to measure.
    baseline_finished_benchmark_phases(pool).await?;
    // Security phases run the scanners once the checkout exists to scan.
    scan_finished_security_phases(pool).await?;
    // Container-tier phases leave their tool calls in the container's tap;
    // nothing else comes to collect them.
    drain_finished_container_phases(pool).await?;
    // A failed phase makes every later phase unreachable, and saying so is what
    // lets the mission finish at all.
    skip_unreachable_phases(pool).await?;
    close_finished_missions(pool).await?;
    Ok(())
}

/// How many phases to capture per tick. Capture shells out to git against a
/// working tree, so a backlog should be worked through steadily rather than
/// all at once.
const CAPTURE_BATCH: i64 = 5;

/// Write out the diff for any finished phase of a mission that has a repo.
///
/// Not just coding phases. `phase_task_text` tells a *research* phase to
/// "save findings under /mission/repo/research/ using file_edit", so research
/// output is real work sitting in the checkout, and the checkout is deleted
/// thirty minutes after the mission ends. Filtering to coding kinds would have
/// quietly thrown away every research brief a repo-bearing mission produced.
///
/// One consequence to know about: `git diff HEAD` is cumulative, so in a
/// research→coding mission the coding phase's patch also contains the research
/// phase's files. That resolves itself once each phase commits — the next
/// phase then diffs against the previous phase's commit rather than the
/// original HEAD.
///
/// Deliberately not hung off `close_finished_phases` or
/// `evaluate_finished_phases`: a phase reaches `completed` through either
/// path depending on whether it declared a `done_when`, and bolting capture
/// onto one of them would silently skip the other. Driving it from the sweep
/// with a `NOT EXISTS` guard covers both and is retryable by construction.
async fn capture_finished_coding_phases(pool: &PgPool) -> Result<(), String> {
    let rows = sqlx::query(
        "SELECT mp.id, mp.mission_id, mp.kind, mp.config, m.runtime_kind
           FROM mission_phases mp
           JOIN missions m ON m.id = mp.mission_id
          -- Terminal, not successful. A phase that did the work and missed its
          -- goal condition still produced a diff, and that diff is what an
          -- operator needs in order to see WHY it missed and what the next pass
          -- can build on. Capturing only completed phases meant that the moment
          -- an unmet phase began reporting failed — correctly — its work was
          -- silently discarded. What was produced, and whether the goal was met,
          -- are different facts: the artifact records the first and mp.status
          -- the second.
          WHERE mp.status IN ('completed', 'failed')
            AND m.repo_id IS NOT NULL
            AND NOT EXISTS (
                SELECT 1 FROM mission_artifacts a
                 WHERE a.mission_id = mp.mission_id
                   AND a.phase_id = mp.id
                   AND a.kind = 'code_diff'
            )
          ORDER BY mp.completed_at DESC NULLS LAST
          LIMIT $1",
    )
    .bind(CAPTURE_BATCH)
    .fetch_all(pool)
    .await
    .map_err(|e| format!("select phases to capture: {e}"))?;

    for row in rows {
        use sqlx::Row;
        let phase_id: Uuid = row.get("id");
        let mission_id: Uuid = row.get("mission_id");
        let kind: String = row.get("kind");
        let config: serde_json::Value = row.get("config");
        let runtime_kind: String = row.get("runtime_kind");
        // Pull the agent's work back onto the host before capturing it.
        // Unpacks over the same checkout path, so capture below is unchanged.
        //
        // NOT for a microVM mission: it has no container, so this would fail to
        // connect and `continue` — skipping capture forever and delivering
        // nothing, while the phase sat there marked completed. Its work was
        // already collected out of the VM by `microvm_executor`, over the same
        // host path, before the VM was destroyed.
        if crate::mission_fs::copy_mode() && runtime_kind != "microvm" {
            let container = crate::mission_runtime::container_name(mission_id);
            if let Err(e) = crate::mission_fs::sync_out(&container, mission_id).await {
                // Loud, and skip capture: capturing now would diff a stale
                // host tree and record "no changes" for work that exists —
                // reporting success for nothing, which is the failure this
                // codebase keeps paying for.
                eprintln!(
                    "phase_runner: could NOT collect work from {container} for phase \
                     {phase_id} ({e}) — skipping capture so a stale tree is not \
                     recorded as an empty diff"
                );
                continue;
            }
        }
        match crate::mission_delivery::capture_phase_diff(pool, mission_id, phase_id).await {
            Ok(Some(c)) => {
                // Capture is the first moment the platform knows whether the
                // phase produced anything — it runs after the phase is already
                // `completed`, because capture selects on that status. So the
                // verdict is applied here rather than at completion.
                if empty_delivery_is_a_failure(
                    &kind,
                    c.files_changed,
                    c.diff_error.as_deref(),
                    &config,
                ) {
                    eprintln!(
                        "phase_runner: phase {phase_id} ({kind}) of mission {mission_id} \
                         delivered NO files — failing it. Set config.allow_empty = true if \
                         this phase is meant to verify rather than change."
                    );
                    if let Err(e) =
                        sqlx::query("UPDATE mission_phases SET status = 'failed' WHERE id = $1")
                            .bind(phase_id)
                            .execute(pool)
                            .await
                    {
                        eprintln!("phase_runner: failing empty phase {phase_id}: {e}");
                    }
                }
            }
            Ok(None) => {
                // The checkout is gone — reaped before capture reached this
                // phase. Record that, or the row stays eligible forever; and
                // because the batch is bounded, a handful of dead phases
                // occupy every slot permanently and no live mission is ever
                // captured again. That is exactly how this was found: five
                // reaped phases from earlier runs blocked the batch while a
                // freshly finished coding phase went untouched.
                if let Err(e) =
                    crate::mission_delivery::record_uncapturable(pool, mission_id, phase_id).await
                {
                    eprintln!("phase_runner: recording uncapturable phase {phase_id}: {e}");
                }
            }
            Err(e) => {
                // A real failure against a checkout that still exists; the
                // next tick retries it.
                eprintln!(
                    "phase_runner: capturing diff for mission {mission_id} phase {phase_id}: {e}"
                );
            }
        }
    }
    Ok(())
}

/// A repo-less mission must not be told its scratch directory is a git checkout.
///
/// The eight ClawHDF5 research documents were written because the preamble said
/// "/mission/repo ... is the mission's git checkout" to a mission that had none.
/// One agent recorded the contradiction verbatim — "No git repo — file is
/// written" — and wrote into a container that was then reaped unread.
#[cfg(test)]
mod skill_delivery_wiring_tests {
    /// Every tier without per-turn injection must be handed the task text that
    /// CARRIES the skills.
    ///
    /// The behavioural tests in `tests/mission_skill_delivery.rs` prove
    /// `phase_skills_text` and `compose_turn_prompt` work. They cannot prove
    /// the three `launch_*` calls pass the composed string rather than the bare
    /// one — and that substitution is a one-word edit that would silently
    /// return all three tiers to delivering no skill, with every test still
    /// green. Same reasoning as `mission_events::the_cap_is_enforced_in_one_statement`.
    /// The prompt must be recorded by the tier that SENDS it.
    ///
    /// Recording at the dispatch fork wrote a phase prompt for container
    /// missions too, and the container tier does not send that text — it sends
    /// the bare task and appends skills per turn. Observed on a live mission:
    /// three prompt.composed rows, one of which was never given to anything.
    #[test]
    fn every_solo_tier_records_the_prompt_it_actually_sends() {
        let src = include_str!("phase_runner.rs");
        let anchor = format!("async fn {}(", "record_phase_prompt");
        assert!(
            src.contains(&anchor),
            "the per-tier recorder is gone; a fork-level record would log \
             prompts that were never sent"
        );
        for launcher in [
            "launch_composed_microvm_phase",
            "launch_microvm_phase",
            "launch_direct_session",
        ] {
            let body = src
                .split(&format!("async fn {launcher}("))
                .nth(1)
                .and_then(|s| s.split("\nasync fn ").next())
                .unwrap_or_else(|| panic!("{launcher} not found"));
            assert!(
                body.contains("record_phase_prompt("),
                "{launcher} runs a prompt it never records — that phase becomes \
                 unexplainable after the fact"
            );
        }
    }

    #[test]
    fn the_three_solo_tiers_are_handed_the_skill_bearing_task() {
        let src = include_str!("phase_runner.rs");
        // Only the dispatch body, so the helper definitions and these tests do
        // not satisfy the assertion by accident.
        // Built at runtime, not written as one literal: a literal anchor
        // appears in THIS test's own source too, and the split then matches
        // itself instead of the code under test.
        let anchor = format!("let task_with_skills = match {}(", "phase_skills_text");
        let body = src
            .split(&anchor)
            .nth(1)
            .and_then(|s| s.split("\nasync fn ").next())
            .expect("dispatch body");

        for (call, tier) in [
            ("launch_composed_microvm_phase(", "composed microVM"),
            ("launch_microvm_phase(", "solo microVM"),
            ("launch_direct_session(", "direct session"),
        ] {
            let args = body
                .split(call)
                .nth(1)
                .and_then(|s| s.split(')').next())
                .unwrap_or_else(|| panic!("{tier}: call site not found"));
            assert!(
                args.contains("&task_with_skills"),
                "{tier} is passed the bare task — that tier has no per-turn \
                 injection, so its agents would receive no skill at all"
            );
            assert!(
                !args.contains("&task,"),
                "{tier} is passed `&task`, the pre-skills string"
            );
        }
    }
}

#[cfg(test)]
mod attribution_tests {
    use super::attribute_sessions;
    use crate::vm_tool_tap::Observed;
    use uuid::Uuid;

    fn call(session: Option<&str>) -> Observed {
        Observed {
            tool: "Bash".into(),
            path: None,
            session: session.map(str::to_string),
            subagent: None,
            subagent_id: None,
            input: serde_json::json!({"command": "ls"}),
            response: serde_json::Value::Null,
        }
    }

    /// The ordinary case: three turns, three sessions, in order.
    #[test]
    fn each_session_lands_on_the_turn_that_ran_it() {
        let (a, b, c) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7());
        let tools = [
            call(Some("s1")),
            call(Some("s1")),
            call(Some("s2")),
            call(Some("s3")),
            call(Some("s2")),
        ];
        assert_eq!(
            attribute_sessions(&tools, &[a, b, c]),
            vec![Some(a), Some(a), Some(b), Some(c), Some(b)],
            "sessions are ordered by FIRST appearance, so a later call from an \
             earlier session still belongs to that earlier turn"
        );
    }

    /// More sessions than turns — something happened this model does not
    /// describe, so it must not produce a confident answer.
    #[test]
    fn a_count_mismatch_attributes_nothing() {
        let a = Uuid::now_v7();
        let tools = [call(Some("s1")), call(Some("s2"))];
        assert_eq!(
            attribute_sessions(&tools, &[a]),
            vec![None, None],
            "a plausible-looking wrong attribution puts one agent's actions on \
             another agent's record, and a person later reasons from it"
        );
        // And the other direction.
        assert_eq!(
            attribute_sessions(&[call(Some("s1"))], &[a, Uuid::now_v7()]),
            vec![None]
        );
    }

    /// One call with no session id poisons the ORDER, not just itself.
    #[test]
    fn a_single_missing_session_refuses_the_whole_batch() {
        let (a, b) = (Uuid::now_v7(), Uuid::now_v7());
        let tools = [call(Some("s1")), call(None), call(Some("s2"))];
        assert_eq!(
            attribute_sessions(&tools, &[a, b]),
            vec![None, None, None],
            "a hole shifts every later session onto the wrong turn"
        );
    }

    /// The pre-session tap, and the microVM path that supplies no turns.
    #[test]
    fn no_turns_and_no_sessions_stay_unattributed() {
        assert_eq!(attribute_sessions(&[call(Some("s1"))], &[]), vec![None]);
        assert_eq!(attribute_sessions(&[call(None)], &[]), vec![None]);
        assert!(attribute_sessions(&[], &[]).is_empty());
    }
}

#[cfg(test)]
mod repo_less_text_tests {
    use super::*;

    #[test]
    fn a_repo_less_phase_is_not_told_it_has_a_checkout() {
        let with = phase_task_text("research", "T", None, None, true);
        let without = phase_task_text("research", "T", None, None, false);

        assert!(with.contains("mission's git checkout"), "{with}");
        assert!(
            !without.contains("git checkout"),
            "a mission with no repo must not be promised one: {without}"
        );
        // And it must say what DOES happen to the files, or an agent told only
        // "there is no repo" has no reason to write them to disk at all.
        assert!(
            without.contains("published as a mission artifact"),
            "{without}"
        );
        assert!(without.contains("NO git repository"), "{without}");
    }

    /// Both variants must keep the instruction that outputs are real files.
    /// That line is what stops an agent pasting its work into the reply, and it
    /// is load-bearing for the capture path either way.
    #[test]
    fn both_variants_still_demand_files_on_disk() {
        for has_repo in [true, false] {
            let t = phase_task_text("research", "T", None, None, has_repo);
            assert!(
                t.contains("REAL files with Write/Edit"),
                "has_repo={has_repo}: {t}"
            );
        }
    }

    /// The prompt must advertise the tools the agent actually has.
    ///
    /// Every executor ends in `claude -p`, so the names are Claude Code's.
    /// Advertising ZeroClaw's names instead (`file_edit`, `content_search`) —
    /// and denying Bash — is what made five agents stop and ask what environment
    /// they were in rather than do the work.
    #[test]
    fn the_prompt_names_the_tools_the_agent_actually_has() {
        let t = phase_task_text("coding", "T", None, None, true);
        for real in ["Read", "Edit", "Write", "Bash", "Glob", "Grep"] {
            assert!(t.contains(real), "missing {real}: {t}");
        }
        for absent in [
            "file_edit",
            "content_search",
            "glob_search",
            "git_operations",
        ] {
            assert!(
                !t.contains(absent),
                "{absent} does not exist under claude_cli — advertising it is the bug: {t}"
            );
        }
    }

    /// A repo-less agent must be told the workspace EXISTS. It is created by
    /// `mission_fs::sync_in`; saying so is what stops the agent concluding the
    /// environment is broken and refusing.
    #[test]
    fn a_repo_less_workspace_is_promised_to_exist() {
        let t = phase_task_text("research", "T", None, None, false);
        assert!(t.contains("EXISTS and is writable"), "{t}");
    }
}

/// Record the baseline for benchmark phases that have finished and have none.
///
/// `benchmark_runner` has existed since Slice 7 and was reachable only from
/// `POST /api/missions/{id}/benchmark` — an operator button. So a `benchmark`
/// mission, whose recipe declares `mode = "author_and_baseline"`, authored
/// benchmarks and then recorded nothing: `benchmark_snapshots` stayed empty and
/// the canvas had nothing to render. The measuring half of "author + baseline"
/// simply never ran.
///
/// Spawned, never awaited in the sweep: `BENCH_TIMEOUT` is 30 minutes, and this
/// loop also starts phases, closes them, evaluates and captures. Blocking it on
/// a benchmark would stall every mission on the platform behind one `cargo
/// bench`.
///
/// The `NOT EXISTS` guard is what makes that safe to fire per tick: a phase with
/// a snapshot is never selected again, so a spawned run cannot be started twice
/// while the first is still going.
async fn baseline_finished_benchmark_phases(pool: &PgPool) -> Result<(), String> {
    let rows = sqlx::query(
        "SELECT mp.id, mp.mission_id
           FROM mission_phases mp
          WHERE mp.kind = 'benchmark'
            AND mp.status IN ('completed', 'failed')
            AND NOT EXISTS (
                SELECT 1 FROM benchmark_snapshots b
                 WHERE b.phase_id = mp.id AND b.iteration = 0
            )
          ORDER BY mp.completed_at DESC NULLS LAST
          LIMIT 2",
    )
    .fetch_all(pool)
    .await
    .map_err(|e| format!("select benchmark phases: {e}"))?;

    for row in rows {
        let phase_id: Uuid = row.get("id");
        let mission_id: Uuid = row.get("mission_id");
        let pool = pool.clone();
        tokio::spawn(async move {
            match crate::benchmark_runner::baseline(&pool, mission_id, phase_id).await {
                Ok(()) => eprintln!(
                    "phase_runner: recorded benchmark baseline for phase {phase_id} \
                     of mission {mission_id}"
                ),
                // Not a phase failure. A repo with no bench harness is a normal
                // outcome, and failing the phase for it would punish a mission
                // that did exactly what it was asked. Said out loud, though —
                // "no baseline" must not be indistinguishable from "not tried".
                Err(e) => eprintln!(
                    "phase_runner: no benchmark baseline for phase {phase_id} \
                     of mission {mission_id}: {e}"
                ),
            }
        });
    }
    Ok(())
}

/// Run the scanners for `security_scan` phases that have finished without one.
///
/// The same defect the benchmark sweep above was written for, in the other
/// half of the platform. `security_scan::run` — which reads `phase.config.tools`
/// to pick scanners and upserts each finding as a `mission_task` — was
/// reachable only from `POST /api/missions/{id}/security-scan`, an operator
/// button. So `security_hardening.toml`, a workflow whose entire first phase is
/// a scan, ran an agent that was never told to scan and then never fired the
/// scanner either. The phase reported `completed` having scanned nothing, and
/// the two facts that would have exposed it — zero findings, and a green
/// phase — are exactly what a genuinely clean repository also looks like.
///
/// Guarded on the marker row rather than on the presence of findings, because
/// a clean scan writes no findings: without it, a clean phase would be
/// rescanned on every tick for as long as the mission existed.
///
/// Spawned, not awaited, for the same reason as the benchmark baseline: four
/// scanners against a large tree take minutes, and this loop also starts,
/// closes and evaluates every phase on the platform.
/// The selection half of the sweep, split out so the marker guard is testable
/// without a container: the scan itself needs Docker, the guard is the part
/// that decides whether it runs twice or never.
pub async fn unscanned_security_phases(pool: &PgPool) -> Result<Vec<(Uuid, Uuid)>, String> {
    let rows = sqlx::query(
        "SELECT mp.id, mp.mission_id
           FROM mission_phases mp
          WHERE mp.kind = 'security_scan'
            AND mp.status IN ('completed', 'failed')
            AND NOT EXISTS (
                SELECT 1 FROM mission_tasks t
                 WHERE t.phase_id = mp.id AND t.external_id = $1
            )
          ORDER BY mp.completed_at DESC NULLS LAST
          LIMIT 2",
    )
    .bind(crate::security_scan::SCAN_MARKER)
    .fetch_all(pool)
    .await
    .map_err(|e| format!("select security_scan phases: {e}"))?;
    Ok(rows
        .into_iter()
        .map(|r| (r.get("id"), r.get("mission_id")))
        .collect())
}

async fn scan_finished_security_phases(pool: &PgPool) -> Result<(), String> {
    for (phase_id, mission_id) in unscanned_security_phases(pool).await? {
        let pool = pool.clone();
        tokio::spawn(async move {
            match crate::security_scan::run(&pool, mission_id, phase_id).await {
                Ok(n) => eprintln!(
                    "phase_runner: security scan recorded {n} finding(s) for phase \
                     {phase_id} of mission {mission_id}"
                ),
                // Not a phase failure, for the same reason the benchmark
                // baseline is not: a repo the scanners cannot read is a real
                // outcome. Said out loud, though — "no findings" must not be
                // indistinguishable from "never scanned".
                Err(e) => eprintln!(
                    "phase_runner: security scan did not run for phase {phase_id} \
                     of mission {mission_id}: {e}"
                ),
            }
        });
    }
    Ok(())
}

/// Drain the container tier's tool tap into `mission_events`.
///
/// The microVM tier records its tools during the turn, from inside the loop
/// that is watching the VM. A container turn is driven asynchronously by
/// `topology_worker`, so there is no such loop — the tap accumulates in the
/// mission's container and something has to come and collect it.
///
/// Without this the hooks write a file nobody reads, which is the same shape
/// as the gate that is installed and inert: everything looks wired and no
/// evidence ever appears.
///
/// Idempotent by truncation, not by a marker: `drain` clears the file it read,
/// so a second pass finds nothing. That is why this can run every tick without
/// a cursor column.
async fn drain_finished_container_phases(pool: &PgPool) -> Result<(), String> {
    let rows = sqlx::query(
        "SELECT mp.id, mp.mission_id, m.runtime_container_name
           FROM mission_phases mp
           JOIN missions m ON m.id = mp.mission_id
          WHERE mp.status IN ('completed', 'failed')
            AND m.runtime_container_name IS NOT NULL
            AND mp.completed_at > now() - interval '30 minutes'
          ORDER BY mp.completed_at DESC
          LIMIT 4",
    )
    .fetch_all(pool)
    .await
    .map_err(|e| format!("select container phases: {e}"))?;

    for row in rows {
        let phase_id: Uuid = row.get("id");
        let mission_id: Uuid = row.get("mission_id");
        let container: String = row.get("runtime_container_name");

        // `container_exec::connect`, NOT connect_with_local_defaults: the
        // server reaches Docker through a socket proxy (DOCKER_HOST), so the
        // local-socket connector fails — and it failed SILENTLY here, so the
        // sweep did nothing while the tap filled up and every other link in
        // the chain looked correct.
        let docker = match crate::container_exec::connect() {
            Ok(d) => d,
            Err(e) => {
                eprintln!("phase_runner: cannot reach docker to drain tool taps: {e}");
                return Ok(());
            }
        };
        // The gate's own confession, before its tool calls: if it could not
        // parse and allowed everything, every call drained below ran unchecked
        // — and until this read existed, the marker it left saying so was seen
        // by exactly one unit test and no production code.
        if let Some(text) = crate::container_tool_hooks::drain_inert(&docker, &container).await {
            let lines = text.lines().count();
            eprintln!(
                "phase_runner: the tool gate in {container} went INERT {lines} time(s) \
                 during phase {phase_id} — those calls were allowed unchecked"
            );
            crate::mission_events::record(
                pool,
                crate::mission_events::MissionEvent::new(
                    mission_id,
                    crate::container_tool_hooks::GATE_INERT,
                )
                .phase(phase_id)
                .detail(serde_json::json!({ "occurrences": lines, "marker": text })),
            )
            .await;
        }
        let tools = crate::container_tool_hooks::drain(&docker, &container).await;
        if tools.is_empty() {
            continue;
        }
        // The phase's own run, so the events hang off the same row the UI
        // already reads. A phase with no run row still records the tools —
        // losing them because the join came up empty would be the worse trade.
        let run_id: Option<Uuid> =
            sqlx::query_scalar("SELECT id FROM topology_runs WHERE phase_id = $1 LIMIT 1")
                .bind(phase_id)
                .fetch_optional(pool)
                .await
                .ok()
                .flatten();
        // The agent of each turn, in the order the turns ran. `prompt.composed`
        // is written by the tier as it sends each turn, so this IS the running
        // order — not a reconstruction of it.
        let turn_agents: Vec<Uuid> = sqlx::query_scalar(
            "SELECT agent_id FROM mission_events
              WHERE phase_id = $1 AND kind = $2 AND agent_id IS NOT NULL
              ORDER BY id",
        )
        .bind(phase_id)
        .bind(crate::mission_events::PROMPT_COMPOSED)
        .fetch_all(pool)
        .await
        .unwrap_or_default();
        record_vm_tools(
            pool,
            mission_id,
            phase_id,
            run_id.unwrap_or(phase_id),
            &tools,
            &turn_agents,
        )
        .await;
        eprintln!(
            "phase_runner: drained {} tool call(s) from {container} for phase {phase_id}",
            tools.len()
        );
    }
    Ok(())
}

/// Did this phase finish without delivering the work it exists to produce?
///
/// A coding phase that changes no files has done nothing, and until now that
/// was reported as `completed` — the same status as a phase that delivered a
/// tested, reviewed, pushed change. Mission `019fcf62` completed that way while
/// its agents were silently unpinned from the repo, and nothing in the platform
/// said otherwise; the failure was found by a script diffing the forge.
///
/// Three things must all hold before calling it a failure, because a false
/// positive here fails honest work:
///
///   - **The phase is one whose directive tells it to write files.** That is
///     every kind except `review`: `coding` changes the tree, `research` is told
///     to "save findings under /mission/repo/research/", `benchmark` to author
///     benchmarks "under /mission/repo/benches", and `security_scan` to file
///     findings and propose patches. A phase that produced nothing did not do
///     what it was told, whatever its kind.
///
///     This used to be `kind == "coding"` alone, with the reasoning "research
///     phases legitimately write nothing to the tree" — which contradicts the
///     research directive three modules over. The cost: `benchmark` and
///     `security_hardening` missions, whose defining phases are NOT coding, had
///     no delivery guarantee at all and reported success on an empty tree.
///   - **The diff was actually computed.** An uncomputable diff also reports
///     zero files (see `mission_delivery::untrusted_empty_reason`); treating it
///     as an empty delivery would blame the agent for a platform fault.
///   - **`allow_empty` is not set.** The escape hatch for a coding phase whose
///     job is genuinely to verify rather than to change — asserted for, not
///     assumed.
fn empty_delivery_is_a_failure(
    kind: &str,
    files_changed: usize,
    diff_error: Option<&str>,
    config: &serde_json::Value,
) -> bool {
    PRODUCING_KINDS.contains(&kind)
        && files_changed == 0
        && diff_error.is_none()
        && config.get("allow_empty").and_then(|v| v.as_bool()) != Some(true)
}

/// Phase kinds whose directive instructs the agent to leave files behind.
///
/// `review` is absent on purpose: a reviewing phase that changes nothing has
/// done its job, and `vm_stop_gate::per_node` makes the same distinction for the
/// same reason. Any other kind falls through to the generic directive ("Execute
/// this mission phase according to the mission brief"), which promises no files,
/// so it is not held to producing them.
const PRODUCING_KINDS: &[&str] = &["coding", "research", "benchmark", "security_scan"];

/// How long a phase may wait for a VM slot before it is failed.
///
/// Two full turns. A fleet that genuinely frees will free within one, so this
/// only fires when nothing is coming — and a phase that waited two hours must say
/// so rather than sit `pending` forever looking like a bug.
const CAPACITY_WAIT_MAX_SECS: f64 = 2.0 * 3600.0;

/// How long a phase may wait for a judge it cannot reach before it is failed.
///
/// Not consuming a pass for an unreachable judge is right; retrying forever is
/// not, because a permanently dead validator would leave the phase `evaluating`
/// in silence — a wrong failure traded for an invisible hang. Thirty minutes is
/// many sweep ticks, so a transient outage recovers well inside it, and a real
/// one surfaces as a failure that names the transport error.
const JUDGE_WAIT_MAX_SECS: f64 = 30.0 * 60.0;

/// Longest gap between two judge attempts on the same phase.
///
/// The retry used to run on the sweep's own 10s tick, so a phase whose judge
/// was unreachable re-judged 180 times in its 30-minute window. A verdict is
/// not one request — [`crate::evaluator`] loops up to `MAX_TOOL_CALLS + 1`
/// times and resends the whole growing history each round — so that was on the
/// order of 2,000 model requests for one phase nobody could judge, and on a
/// transport error they are billed: the request was processed, only its
/// response failed to decode.
const JUDGE_RETRY_MAX_BACKOFF_SECS: f64 = 5.0 * 60.0;

/// How long to wait before the next judge attempt, given how long this phase
/// has already been blocked.
///
/// Waiting as long as we have already waited doubles the total elapsed time per
/// attempt, so the schedule is exponential without storing an attempt counter:
/// 10, 20, 40, 80, 160, 300, 300 … — about ten attempts across the same
/// 30-minute window instead of a hundred and eighty.
fn judge_backoff_secs(blocked_for: f64) -> f64 {
    blocked_for.clamp(POLL_INTERVAL.as_secs_f64(), JUDGE_RETRY_MAX_BACKOFF_SECS)
}

/// A judge error that retrying cannot fix before a time the error itself names.
///
/// z.ai answers an exhausted plan with a 429 carrying code `1310` and its own
/// reset timestamp. Retrying that is not optimism, it is arithmetic: the reset
/// was two days out when this fired on 2026-09-09, and the phase spent its full
/// 30-minute window asking a question whose answer could not change. Fail
/// immediately instead, and say what is actually wrong — "quota exhausted until
/// X" sends you to the plan, where "the independent validator could not be
/// reached" sends you into the mission.
///
/// Conservative on purpose: an error that does not positively identify itself
/// as an exhausted plan is treated as retryable, because giving up on a
/// transient blip costs a phase that had done nothing wrong.
fn judge_error_is_exhausted_plan(why: &str) -> Option<String> {
    if !(why.contains("Limit Exhausted") || why.contains(r#""code":"1310""#)) {
        return None;
    }
    let until: Option<String> = why.find("reset at ").map(|i| {
        why[i + "reset at ".len()..]
            .chars()
            .take_while(|c| *c != ']' && *c != '"')
            .collect::<String>()
            .trim()
            .to_string()
    });
    Some(match until.filter(|u| !u.is_empty()) {
        Some(u) => format!("the judge provider's plan limit is exhausted until {u}"),
        None => "the judge provider's plan limit is exhausted".to_string(),
    })
}

/// Stamp why a phase is waiting, returning how long it has waited so far.
///
/// The timestamp is set once and preserved across retries, so the wait is
/// measured from the first refusal rather than reset every 10s sweep.
async fn record_capacity_block(pool: &PgPool, phase_id: Uuid, note: &str) -> f64 {
    let waited: Option<f64> = sqlx::query_scalar(
        "UPDATE mission_phases
            SET capacity_blocked_since = COALESCE(capacity_blocked_since, now()),
                capacity_note = $2
          WHERE id = $1
      RETURNING EXTRACT(EPOCH FROM now() - capacity_blocked_since)::float8",
    )
    .bind(phase_id)
    .bind(note)
    .fetch_optional(pool)
    .await
    .ok()
    .flatten();
    waited.unwrap_or(0.0)
}

/// Enqueue topology_runs for every phase whose predecessors are done.
async fn start_pending_phases(
    pool: &PgPool,
    hub: &std::sync::Arc<crate::fleet::NodeHub>,
) -> Result<(), String> {
    // Eligible = pending phase, mission running, all lower-order phases
    // in this mission are 'completed'. `NOT EXISTS ... status <> completed`
    // handles order 0 (no prior rows) + skipped phases naturally.
    let rows = sqlx::query(
        "SELECT mp.id, mp.mission_id, mp.kind, mp.order_idx, mp.iteration,
                mp.config->>'task' AS phase_task,
                -- The whole config, not just the task: the stop gate is built
                -- from `allow_empty` and `done_when_check`, and a phase that
                -- declares a completion check gets it enforced in the agent's
                -- own loop rather than only after it has finished.
                mp.config,
                m.workspace_id, m.title, m.description,
                -- Where and how this mission executes. `runtime_kind` decides
                -- which executor takes the phase; without it 'microvm' is a
                -- value the placement code honours and nothing reads.
                m.runtime_kind, m.backend, m.target_node_id, m.team_engine,
                -- Whether a checkout exists at all. A repo-less mission's
                -- /mission/repo is scratch space, and the task text must say so.
                (m.repo_id IS NOT NULL) AS has_repo
         FROM mission_phases mp
         JOIN missions m ON m.id = mp.mission_id
         WHERE mp.status = 'pending'
           AND m.status = 'running'
           AND NOT EXISTS (
             SELECT 1 FROM mission_phases prior
             WHERE prior.mission_id = mp.mission_id
               AND prior.order_idx < mp.order_idx
               AND prior.status <> 'completed'
           )
         LIMIT 20",
    )
    .fetch_all(pool)
    .await
    .map_err(|e| format!("query eligible phases: {e}"))?;

    for row in rows {
        let phase_id: Uuid = row.get("id");
        let mission_id: Uuid = row.get("mission_id");
        let kind: String = row.get("kind");
        let workspace_id: Uuid = row.get("workspace_id");
        let title: String = row.get("title");
        let description: Option<String> = row.get("description");
        let phase_task: Option<String> = row.get("phase_task");
        let config: serde_json::Value = row.get("config");
        let iteration: i32 = row.get("iteration");
        let runtime_kind: String = row.get("runtime_kind");
        let backend: Option<String> = row.get("backend");
        let target_node_id: Option<Uuid> = row.get("target_node_id");
        let team_engine: Option<String> = row.get("team_engine");
        let has_repo: bool = row.get("has_repo");

        if let Err(e) = launch_phase(
            pool,
            hub,
            PhaseLaunch {
                phase_id,
                mission_id,
                kind: &kind,
                workspace_id,
                title: &title,
                description: description.as_deref(),
                phase_task: phase_task.as_deref(),
                config: &config,
                iteration,
                runtime_kind: &runtime_kind,
                backend: backend.as_deref(),
                target_node_id,
                team_engine: team_engine.as_deref(),
                has_repo,
            },
        )
        .await
        {
            eprintln!("phase_runner: launch phase {phase_id} failed: {e}");
        }
    }
    Ok(())
}

/// Everything `launch_phase` needs about the phase it is starting, gathered
/// from the eligibility query.
struct PhaseLaunch<'a> {
    phase_id: Uuid,
    mission_id: Uuid,
    kind: &'a str,
    workspace_id: Uuid,
    title: &'a str,
    description: Option<&'a str>,
    /// This phase's own instructions, from `mission_phases.config.task`.
    ///
    /// Without it every phase of a mission receives byte-identical task text
    /// and differs only by the kind directive — so a two-phase mission has
    /// both phases do the same work. Mission `019fc42b` demonstrated it: two
    /// coding phases with distinct `task` values both produced the same two
    /// files, because neither phase ever saw its own instructions.
    phase_task: Option<&'a str>,
    /// `mission_phases.config`, for the settings that shape execution rather
    /// than describe the work — currently the stop gate's two.
    config: &'a serde_json::Value,
    /// Which pass this is, 0-based. Stamped onto the runs so the completion
    /// check can tell this pass's work from the previous one's.
    iteration: i32,
    /// `missions.runtime_kind`. Selects the executor.
    runtime_kind: &'a str,
    /// `missions.backend` — which per-CLI image, on the microVM path.
    backend: Option<&'a str>,
    /// Set by `mission_orchestrator` at launch. On the microVM path it is where
    /// the VM boots, and it is not optional there.
    target_node_id: Option<Uuid>,
    /// `missions.team_engine`. NULL = solo.
    team_engine: Option<&'a str>,
    /// Whether `missions.repo_id` is set. Decides whether the agents are told
    /// `/mission/repo` is a git checkout or a scratch workspace whose contents
    /// are captured as artifacts.
    has_repo: bool,
}

/// Which team purposes execute a phase of this kind.
///
/// `pub(crate)` on purpose: the World visualization attributes agents to phase
/// orbs, and it must use the SAME mapping the runner uses to pick executors. A
/// second copy would let the picture disagree with the machine about who is
/// working on what — which presents as a rendering bug and is really a lie.
pub(crate) fn purposes_for(kind: &str) -> &'static [&'static str] {
    match kind {
        "research" => &["research", "mission"],
        "coding" => &["coding", "mission"],
        "benchmark" => &["coding", "mission"],
        "security_scan" => &["security", "coding", "mission"],
        _ => &["mission"],
    }
}

async fn launch_phase(
    pool: &PgPool,
    hub: &std::sync::Arc<crate::fleet::NodeHub>,
    p: PhaseLaunch<'_>,
) -> Result<(), String> {
    let PhaseLaunch {
        phase_id,
        mission_id,
        kind,
        workspace_id,
        title,
        description,
        phase_task,
        has_repo,
        iteration,
        // Destructured but read through `p` below, so the compiler keeps this
        // pattern honest if a field is added.
        config: _,
        runtime_kind: _,
        backend: _,
        target_node_id: _,
        team_engine: _,
    } = p;
    // Which team purposes should execute this phase.
    let purposes: &[&str] = purposes_for(kind);

    // Load teams for this mission matching any of the purposes.
    let team_rows = sqlx::query(
        "SELECT mt.team_id, t.graph
         FROM mission_teams mt
         JOIN teams t ON t.id = mt.team_id
         WHERE mt.mission_id = $1
           AND mt.purpose = ANY($2)",
    )
    .bind(mission_id)
    .bind(purposes)
    .fetch_all(pool)
    .await
    .map_err(|e| format!("query mission teams: {e}"))?;

    // A microVM mission has no teams and needs none: its agent is a `claude -p`
    // inside a VM, not a graph of ZeroClaw claws. Without this exemption the
    // phase would sit `pending` forever while the log said only "no matching
    // teams" — the executor below would never be reached at all.
    //
    // The checkout below still runs, because the VM needs the repository.
    if team_rows.is_empty() && p.runtime_kind != "microvm" {
        eprintln!(
            "phase_runner: mission {mission_id} phase {phase_id} ({kind}) has no matching teams — skipping (staying pending)"
        );
        return Ok(());
    }

    // Ensure the mission's repo is checked out first — the runtime
    // container bind-mounts /var/lib/clawmates-missions/{id}, which
    // must exist before docker start or the mount fails.
    // Idempotent: fetch+reset on existing clones, clone on missing.
    // Non-fatal: research-only missions have no repo and skip cleanly.
    match crate::mission_workspace::ensure_checkout(
        pool,
        cm_domain::WorkspaceId::from(workspace_id),
        mission_id,
    )
    .await
    {
        Ok(Some(path)) => {
            eprintln!(
                "phase_runner: repo checked out at {} for mission {mission_id} phase {phase_id}",
                path.display()
            );
            // From here the checkout belongs to a running phase. Recording it
            // explicitly is what stops the *next* phase's launch from
            // refreshing the tree out from under this one's output — a
            // decision that must not depend on what the agent leaves behind.
            crate::mission_workspace::mark_phase_started(&path);
        }
        Ok(None) => {}
        Err(e) => eprintln!(
            "phase_runner: repo checkout for mission {mission_id} phase {phase_id} failed (continuing): {e}"
        ),
    }

    // Provision the per-mission runtime container if not bound.
    // Idempotent — on_launch sets this on initial launch, but pre-C3
    // missions or retries against a torn-down container land here.
    // Always call ensure_container — the fast path re-mints a fresh
    // one-time pairing code even for existing containers. Old codes
    // expire / are single-use, so a launch that reuses a container
    // still needs a fresh code for the topology_worker's next /pair.
    //
    // Skipped for a microVM mission: its agent runs in a VM on a fleet node, so
    // a container here would be provisioned, have the checkout copied into it,
    // and then sit idle for the life of the mission — while the pairing code and
    // runtime binding it writes describe a runtime nothing is using.
    let needs_container = p.runtime_kind != "microvm";
    if let Some(prov) =
        crate::mission_runtime::MissionRuntimeProvisioner::from_env().filter(|_| needs_container)
    {
        match prov.ensure_container(mission_id).await {
            Ok(ec) => {
                let name = crate::mission_runtime::container_name(mission_id);
                crate::container_tool_hooks::record_install(
                    pool,
                    mission_id,
                    Some(phase_id),
                    ec.hooks.as_deref(),
                )
                .await;
                // Push the checkout into the container. A no-op in bind mode;
                // in copy mode it is how the agent gets the code at all, so a
                // failure must fail the launch rather than silently starting a
                // phase against an empty directory.
                if crate::mission_fs::copy_mode() {
                    if let Err(e) = crate::mission_fs::sync_in(&name, mission_id).await {
                        return Err(format!("copy checkout into {name}: {e}"));
                    }
                }
                // Re-assert this mission's crew in the runtime config.
                //
                // Claws are provisioned once, at on_launch. `ensure_container`
                // RECREATES a container that is not running, and recreation
                // reseeds `.zeroclaw` from the seed directory — which does not
                // contain this mission's claws. The agents still exist in
                // Postgres, so nothing looks wrong, but the alias the turn
                // dials is gone from the daemon and `/ws/chat` answers
                // `400 Bad Request`. That is what a retry of mission 01a00538
                // hit after its container was recreated.
                //
                // provision_claw is idempotent (created:false when present), so
                // this costs one call per claw on the happy path and is the
                // difference between a resumable mission and a dead one.
                // Aim at THIS MISSION's daemon, never the global gateway. Each
                // mission's turns run against its own container, which loads
                // config at boot and never re-reads the file, so provisioning
                // against `from_env()` writes the claws into the shared runtime
                // and leaves this one with none — the failure mode
                // `RuntimeProvisioner::for_gateway`'s doc comment describes, and
                // the one this block exists to repair.
                if let Some(rp) =
                    crate::runtime_provision::RuntimeProvisioner::for_gateway(ec.endpoint.clone())
                {
                    let crew = sqlx::query(
                        "SELECT DISTINCT a.id, a.model_binding, t.risk_profile, t.mcp_bundles
                           FROM team_members tm
                           JOIN mission_teams mt ON mt.team_id = tm.team_id
                           JOIN teams t          ON t.id = tm.team_id
                           JOIN agents a         ON a.id = tm.claw_id
                          WHERE mt.mission_id = $1 AND a.deleted_at IS NULL",
                    )
                    .bind(mission_id)
                    .fetch_all(pool)
                    .await
                    .unwrap_or_default();
                    let mut reasserted: Vec<cm_domain::AgentId> = Vec::new();
                    for row in &crew {
                        let aid: uuid::Uuid = row.get("id");
                        let model: Option<String> = row.get("model_binding");
                        let risk: Option<String> = row.get("risk_profile");
                        // Re-assert the team's OWN bundles. Passing a constant
                        // here would quietly strip `clawmates_skills` from a
                        // crew that had it, and a re-provision that removes a
                        // capability is worse than one that never ran: the
                        // agent keeps working and simply stops being able to
                        // read its skills, halfway through the mission.
                        let bundles: Vec<String> = row
                            .get::<serde_json::Value, _>("mcp_bundles")
                            .as_array()
                            .map(|a| {
                                a.iter()
                                    .filter_map(|v| v.as_str().map(|s| s.to_string()))
                                    .collect()
                            })
                            .unwrap_or_default();
                        match rp
                            .provision_claw(
                                aid,
                                model.as_deref().unwrap_or("claude"),
                                risk.as_deref().unwrap_or("research_readonly"),
                                &bundles,
                            )
                            .await
                        {
                            Ok(_) => reasserted.push(cm_domain::AgentId::from(aid)),
                            Err(e) => eprintln!(
                                "phase_runner: re-provision claw {aid} for {mission_id} failed \
                                 (continuing): {e}"
                            ),
                        }
                    }
                    // Provisioning creates the agent; it does NOT set
                    // workspace.path (the prop-schema cannot). Without this the
                    // claws exist and every one of them runs in its own sandbox
                    // instead of the mission tree.
                    if let Err(e) = prov
                        .pin_agent_workspaces(mission_id, &reasserted, "/mission/repo")
                        .await
                    {
                        return Err(format!("re-pin mission workspaces: {e}"));
                    }
                }
                if let Err(e) = cm_db::repo::missions::set_runtime_binding(
                    pool,
                    mission_id,
                    workspace_id,
                    Some(&name),
                    Some(&ec.endpoint),
                    ec.pairing_code.as_deref(),
                )
                .await
                {
                    eprintln!(
                        "phase_runner: bind runtime container for {mission_id} failed: {e}"
                    );
                } else {
                    eprintln!(
                        "phase_runner: runtime container {name} → {} (paired={}) for mission {mission_id}",
                        ec.endpoint,
                        ec.pairing_code.is_some()
                    );
                }
            }
            Err(e) => eprintln!(
                "phase_runner: provision runtime container for {mission_id} failed (continuing): {e}"
            ),
        }
    }

    // On a second or later pass, tell the agents what the evaluator found
    // missing. This is what makes iteration converge instead of repeat — the
    // same mechanism `/goal` uses when it feeds the evaluator's reason into
    // the next turn, and that swarm.rs uses for rejected work.
    let prior = crate::evaluator::latest(pool, phase_id)
        .await
        .unwrap_or(None);
    let task = phase_task_text(kind, title, description, phase_task, has_repo);
    let task = match prior {
        Some((iter, false, guidance)) => format!(
            "{task}\n\nPASS {} DID NOT SATISFY THE COMPLETION CONDITION. What is \
             still missing:\n{guidance}\n\nDo the work this describes. Producing \
             output that merely looks like it satisfies the check — printing an \
             expected value, weakening a test, stubbing a result — fails the pass, \
             because the condition is verified against the repository itself.",
            iter + 1
        ),
        _ => task,
    };

    // The container tier is deliberately NOT given this: it injects per-turn in
    // `topology_exec`, with the running node's own role, and appending here too
    // would put every crew member's skills in every turn twice.
    let task_with_skills = match phase_skills_text(pool, mission_id).await {
        // Always `Inline` here, and not because it is the default: the solo
        // tiers get no skills door (`install_skills_door` runs only for a
        // mission with its own container), so an index would list uris nothing
        // in the VM can fetch. When the microVM tier folds onto
        // `container_tool_hooks` this becomes a real choice; today it is a fact.
        Some(skills) => crate::topology_exec::compose_turn_prompt(
            &task,
            Some(&skills),
            crate::skill_delivery::Mode::Inline,
        ),
        None => task.clone(),
    };


    // Direct-session executor: run the whole phase as ONE `claude -p` session
    // against the mission checkout, instead of driving turns through ZeroClaw.
    //
    // Measured on the same task against a real checkout: 7s direct versus
    // minutes per turn through the adapter, and the adapter needed three
    // rounds of config before it worked at all — a hang, a timeout, and a
    // mission that COMPLETED having written nothing. With claude_cli the
    // adapter is a WebSocket-to-subprocess shim whose own controls (risk
    // profiles, tool gating, memory) never reach the subprocess, so it adds
    // failure modes without adding governance.
    //
    // It still creates one `topology_runs` row. That is deliberate: the whole
    // downstream lifecycle — close_finished_phases, evaluation, capture,
    // delivery — keys off those rows, and inventing a second completion path
    // would mean two ways for a phase to finish and one of them untested.
    // microVM executor. Checked BEFORE `direct_mode` because it is a property of
    // the mission, not of the deployment: `runtime_kind='microvm'` was chosen for
    // this mission specifically, and an env var that happens to be set must not
    // silently run it somewhere else.
    if p.runtime_kind == "microvm" {
        // PLACEMENT, for both the solo and composed paths, in one place so they
        // cannot disagree about which node this phase runs on.
        //
        // Per phase rather than per mission: re-placing is free because mission
        // state lives on the gateway checkout (inject -> run -> collect ->
        // destroy), so a node that filled or drained since the last phase simply
        // is not chosen for the next one.
        // The node this phase will use. Starts as the mission's current pin (a
        // request), becomes whatever placement actually chose.
        let mut chosen_node = p.target_node_id;
        // A composed graph runs on ONE node, and its nodes may each name their
        // own backend — an independent verifier on another provider is the
        // roster's whole purpose. So the node must hold EVERY rootfs the graph
        // asks for, not just the mission's. Read here rather than carried on
        // `PhaseLaunch` because it is only the microVM path that cares.
        let roster: Option<serde_json::Value> =
            sqlx::query_scalar("SELECT config -> 'roster' FROM missions WHERE id = $1")
                .bind(mission_id)
                .fetch_optional(pool)
                .await
                .ok()
                .flatten();
        let backends = crate::vm_placement::required_backends(p.backend, roster.as_ref());
        if backends.len() > 1 {
            eprintln!(
                "phase_runner: mission {mission_id} phase {phase_id} needs {backends:?} \
                 on a single node (composed graph with per-node backends)"
            );
        }
        match crate::vm_placement::choose(pool, hub, workspace_id, &backends, chosen_node).await {
            Ok(node) => {
                if chosen_node != Some(node.as_uuid()) {
                    eprintln!(
                        "phase_runner: mission {mission_id} phase {phase_id} placed on node {} \
                         (was {:?})",
                        node.as_uuid(),
                        chosen_node
                    );
                }
                // Pin for the executors, which read `missions.target_node_id`,
                // and clear any capacity wait now that one is satisfied.
                let _ = sqlx::query(
                    "UPDATE missions SET target_node_id = $1, updated_at = now() WHERE id = $2",
                )
                .bind(node.as_uuid())
                .bind(mission_id)
                .execute(pool)
                .await;
                let _ = sqlx::query(
                    "UPDATE mission_phases
                        SET capacity_blocked_since = NULL, capacity_note = NULL
                      WHERE id = $1 AND capacity_blocked_since IS NOT NULL",
                )
                .bind(phase_id)
                .execute(pool)
                .await;
                chosen_node = Some(node.as_uuid());
            }
            // TRANSIENT: the fleet is full, or we cannot read it. Leave the phase
            // `pending` — `start_pending_phases` retries every 10s, and that loop
            // IS the queue. No topology_runs row is created, so nothing downstream
            // sees a run that never happened.
            Err(e) if e.is_transient() => {
                let msg = e.message();
                // Recorded, not just logged: a phase waiting for capacity and a
                // phase nothing is working on look identical from the outside,
                // and this codebase has paid for that confusion repeatedly.
                let blocked_for = record_capacity_block(pool, phase_id, &msg).await;
                if blocked_for >= CAPACITY_WAIT_MAX_SECS {
                    eprintln!(
                        "phase_runner: mission {mission_id} phase {phase_id} waited \
                         {blocked_for:.0}s for a VM slot — failing it.\n{msg}"
                    );
                    let _ = sqlx::query(
                        "UPDATE mission_phases SET status = 'failed', completed_at = now(),
                            capacity_note = $2 WHERE id = $1",
                    )
                    .bind(phase_id)
                    .bind(format!(
                        "waited {blocked_for:.0}s for fleet capacity:\n{msg}"
                    ))
                    .execute(pool)
                    .await;
                    return Err(format!("phase {phase_id} exhausted its capacity wait"));
                }
                eprintln!(
                    "phase_runner: mission {mission_id} phase {phase_id} WAITING for a VM \
                     slot ({blocked_for:.0}s so far)\n{msg}"
                );
                return Ok(());
            }
            // Not transient — no node could ever run this backend. Waiting will
            // not fix it, so fail the phase rather than queue it forever.
            Err(e) => return Err(e.message()),
        }

        // A composed mission's graph comes from the same `mission_teams` row a
        // ZeroClaw mission would use — only its nodes run as VMs instead of
        // claws. That is the whole point of composing the engines: the graph,
        // the planners and the durability are Engine Z's, unchanged.
        if wants_composed(p.team_engine) {
            let team = team_rows
                .iter()
                .map(|r| {
                    (
                        r.get::<Uuid, _>("team_id"),
                        r.get::<serde_json::Value, _>("graph"),
                    )
                })
                .collect::<Vec<_>>();
            return launch_composed_microvm_phase(
                pool,
                mission_id,
                phase_id,
                workspace_id,
                iteration,
                &task_with_skills,
                team,
                purposes,
            )
            .await;
        }
        return launch_microvm_phase(
            pool,
            hub,
            mission_id,
            phase_id,
            kind,
            workspace_id,
            iteration,
            &task_with_skills,
            p.backend,
            chosen_node,
            p.team_engine,
            crate::vm_stop_gate::StopGate::for_phase(kind, p.config),
            has_repo,
        )
        .await;
    }

    if crate::session_executor::direct_mode() {
        return launch_direct_session(
            pool,
            mission_id,
            phase_id,
            workspace_id,
            iteration,
            &task_with_skills,
        )
            .await;
    }

    // Purge prior failed / cancelled runs for this phase so the card
    // starts fresh on re-attempts. Completed runs are kept for
    // auditability (a mission that succeeded once and got re-run
    // still shows both), but the failure noise from earlier attempts
    // doesn't clutter the retry.
    sqlx::query(
        "DELETE FROM topology_runs
          WHERE mission_phase_id = $1
            AND status IN ('failed', 'cancelled')",
    )
    .bind(phase_id)
    .execute(pool)
    .await
    .map_err(|e| format!("purge prior failed runs for phase {phase_id}: {e}"))?;

    for r in &team_rows {
        let team_id: Uuid = r.get("team_id");
        let graph: serde_json::Value = r.get("graph");
        // Inject each node's explicit `agent` alias so the executor
        // dials the exact claw provisioned for THIS team's role,
        // instead of falling through to the env-based ZEROCLAW_AGENT_MAP
        // (which points at ambient names that don't exist per-team).
        let graph = inject_node_agents(pool, team_id, graph).await;
        let run_id = Uuid::now_v7();
        sqlx::query(
            "INSERT INTO topology_runs
                (id, workspace_id, task, kind, status, graph, tier,
                 team_id, mission_id, mission_phase_id, iteration)
             VALUES ($1, $2, $3, 'run', 'queued', $4, 'team', $5, $6, $7, $8)",
        )
        .bind(run_id)
        .bind(workspace_id)
        .bind(&task)
        .bind(&graph)
        .bind(team_id)
        .bind(mission_id)
        .bind(phase_id)
        // Stamps which pass produced this run, so the "all runs finished?"
        // check can't be satisfied by a previous pass's completed rows.
        .bind(iteration)
        .execute(pool)
        .await
        .map_err(|e| format!("enqueue run for team {team_id}: {e}"))?;
    }

    // Flip phase to running.
    mark_phase_running(pool, mission_id, phase_id).await?;
    eprintln!(
        "phase_runner: mission {mission_id} phase {phase_id} ({kind}) launched with {} team(s)",
        team_rows.len()
    );
    Ok(())
}

/// Launch a phase as a single headless session.
///
/// Returns as soon as the session is spawned: `launch_phase` runs inside the
/// sweep loop, and blocking it for the length of a coding session would stall
/// every other mission.
/// Does this mission want the composed engines — a durable ZeroClaw graph whose
/// every node is a Claude-Code-in-a-microVM session?
///
/// The third `team_engine` name migration 0069 anticipated. Exact match, like
/// `wants_claude_code_team`: a typo must run solo rather than half-compose.
fn wants_composed(team_engine: Option<&str>) -> bool {
    matches!(team_engine, Some("composed"))
}

/// Enqueue a composed run — Slice 4's producer side.
///
/// Unlike every other executor here this one does **not** run the work: it emits
/// ONE `queued` row carrying the real graph and lets `topology_worker` claim it.
/// That is where the durability comes from — the checkpoint, the resume after a
/// crash, and the cancellation all belong to the worker, and a run that drove
/// itself from a `tokio::spawn` (as the solo microVM path does) would have none
/// of them.
///
/// One row, not one per team, and the constraint is physical: every node injects
/// from and collects back over the SAME host checkout, so two concurrent runs of
/// one phase would be two VMs writing one directory. A mission with two matching
/// teams is refused, visibly, rather than silently running only the first.
#[allow(clippy::too_many_arguments)]
async fn launch_composed_microvm_phase(
    pool: &PgPool,
    mission_id: Uuid,
    phase_id: Uuid,
    workspace_id: Uuid,
    iteration: i32,
    task: &str,
    teams: Vec<(Uuid, serde_json::Value)>,
    purposes: &[&str],
) -> Result<(), String> {
    record_phase_prompt(pool, mission_id, phase_id, "composed_microvm", task).await;
    sqlx::query(
        "DELETE FROM topology_runs
          WHERE mission_phase_id = $1 AND status IN ('failed', 'cancelled')",
    )
    .bind(phase_id)
    .execute(pool)
    .await
    .map_err(|e| format!("purge prior runs for phase {phase_id}: {e}"))?;

    // Refusals are recorded as a failed run rather than returned as an error:
    // `launch_phase` is called from the sweep every ten seconds, so a returned
    // error is a phase that retries forever while the log repeats itself. A
    // failed row closes the phase and says why on the card.
    let refuse = |why: String| async move {
        eprintln!("phase_runner: composed phase {phase_id} of mission {mission_id} refused: {why}");
        let _ = sqlx::query(
            "INSERT INTO topology_runs
                (id, workspace_id, task, kind, status, graph, tier,
                 mission_id, mission_phase_id, iteration, error)
             VALUES ($1, $2, $3, 'run', 'failed', $4, 'microvm_graph', $5, $6, $7, $8)",
        )
        .bind(Uuid::now_v7())
        .bind(workspace_id)
        .bind(task)
        .bind(serde_json::json!({ "nodes": [], "edges": [] }))
        .bind(mission_id)
        .bind(phase_id)
        .bind(iteration)
        .bind(&why)
        .execute(pool)
        .await;
        // The phase must still leave `pending`, or the sweep re-launches it.
        let _ = mark_phase_running(pool, mission_id, phase_id).await;
        Ok(())
    };

    let (team_id, graph) = match teams.len() {
        1 => {
            let (id, g) = teams.into_iter().next().expect("len == 1");
            (Some(id), g)
        }
        // The usual case, and not an error: `mission_orchestrator::on_launch`
        // deliberately mints NO team for a microVM mission, because claws in
        // containers are exactly what a VM mission does not use. A composed run
        // needs the template's shape, not its claws, so it builds the graph from
        // the template directly.
        0 => match crate::mission_orchestrator::composed_graph(pool, mission_id, purposes).await {
            Ok(Some(g)) => (None, g),
            Ok(None) => {
                return refuse(
                    "team_engine='composed' needs a team template to give the run its \
                     shape, and this mission picked none — a composed mission is a \
                     ZeroClaw graph whose nodes happen to be VMs, so without the graph \
                     there is nothing to compose"
                        .to_string(),
                )
                .await
            }
            Err(e) => return refuse(format!("could not build the composed graph: {e}")).await,
        },
        n => {
            return refuse(format!(
                "{n} teams match this phase, and a composed run must be exactly one: \
                 every node injects from and collects back over the same host \
                 checkout, so two runs would be two VMs writing one directory"
            ))
            .await
        }
    };

    // Parse here rather than letting the worker discover it: a graph the
    // orchestrator cannot plan would otherwise be claimed, fail with "missing or
    // invalid graph", and look like a runtime fault instead of a bad team.
    if let Err(e) = serde_json::from_value::<cm_topology::TopologyGraph>(graph.clone()) {
        return refuse(format!(
            "the graph for this composed phase (team {team_id:?}) is not a runnable \
             topology: {e}"
        ))
        .await;
    }

    let run_id = Uuid::now_v7();
    sqlx::query(
        "INSERT INTO topology_runs
            (id, workspace_id, task, kind, status, graph, tier,
             team_id, mission_id, mission_phase_id, iteration)
         VALUES ($1, $2, $3, 'run', 'queued', $4, 'microvm_graph', $5, $6, $7, $8)",
    )
    .bind(run_id)
    .bind(workspace_id)
    .bind(task)
    .bind(&graph)
    .bind(team_id)
    .bind(mission_id)
    .bind(phase_id)
    .bind(iteration)
    .execute(pool)
    .await
    .map_err(|e| format!("enqueue composed run for phase {phase_id}: {e}"))?;

    mark_phase_running(pool, mission_id, phase_id).await?;

    eprintln!(
        "phase_runner: mission {mission_id} phase {phase_id} queued as a COMPOSED run \
         {run_id} (team {team_id:?}) — a ZeroClaw graph with microVM nodes"
    );
    Ok(())
}

/// Launch a phase inside a Firecracker microVM on the mission's placed node.
///
/// Mirrors [`launch_direct_session`] on purpose, down to creating exactly ONE
/// `topology_runs` row: `close_finished_phases`, evaluation, capture and delivery
/// all key off those rows, and a second completion path would be a second way for
/// a phase to finish with one of them untested.
///
/// The run row and the phase flip happen **before** any fallible VM work, so a
/// configuration error (no subscription token, a node that lost its capability)
/// surfaces as a failed run an operator can see — not as a phase that stays
/// `pending` and is retried every ten seconds forever.
#[allow(clippy::too_many_arguments)]
async fn launch_microvm_phase(
    pool: &PgPool,
    hub: &std::sync::Arc<crate::fleet::NodeHub>,
    mission_id: Uuid,
    phase_id: Uuid,
    // The phase kind, used to label this run's single checkpoint record.
    kind: &str,
    workspace_id: Uuid,
    iteration: i32,
    task: &str,
    backend: Option<&str>,
    target_node_id: Option<Uuid>,
    team_engine: Option<&str>,
    gate: Option<crate::vm_stop_gate::StopGate>,
    // Whether the mission has a repository. A repo-less mission gets an EMPTY
    // workspace at the same guest path instead of a checkout — see
    // `VmPhase::has_repo`.
    has_repo: bool,
) -> Result<(), String> {
    record_phase_prompt(pool, mission_id, phase_id, "microvm", task).await;
    sqlx::query(
        "DELETE FROM topology_runs
          WHERE mission_phase_id = $1 AND status IN ('failed', 'cancelled')",
    )
    .bind(phase_id)
    .execute(pool)
    .await
    .map_err(|e| format!("purge prior runs for phase {phase_id}: {e}"))?;

    let run_id = Uuid::now_v7();
    sqlx::query(
        "INSERT INTO topology_runs
            (id, workspace_id, task, kind, status, graph, tier,
             mission_id, mission_phase_id, iteration)
         VALUES ($1, $2, $3, 'run', 'running', $4, 'microvm', $5, $6, $7)",
    )
    .bind(run_id)
    .bind(workspace_id)
    .bind(task)
    .bind(serde_json::json!({ "nodes": [], "edges": [], "executor": "microvm" }))
    .bind(mission_id)
    .bind(phase_id)
    .bind(iteration)
    .execute(pool)
    .await
    .map_err(|e| format!("enqueue microvm run for phase {phase_id}: {e}"))?;

    mark_phase_running(pool, mission_id, phase_id).await?;

    let repo = crate::mission_workspace::checkout_path(mission_id);
    let task = task.to_string();
    let backend = backend.map(str::to_string);
    let team_engine = team_engine.map(str::to_string);
    let pool2 = pool.clone();
    let hub = hub.clone();
    // Moved in for the checkpoint record below: the reader labels each record by
    // role, and for a solo run the phase kind is the only role there is.
    let phase_kind = kind.to_string();
    tokio::spawn(async move {
        // Everything fallible lives in here, so every outcome closes the run.
        let outcome = async {
            let node = target_node_id.ok_or_else(|| {
                "mission has no target_node_id — placement should have set one at \
                 launch; a microvm mission cannot run on the gateway, which has no \
                 /dev/kvm"
                    .to_string()
            })?;
            // A repo-BACKED mission with no checkout is a real fault: something
            // went wrong before the phase, and booting a VM to hand the agent an
            // empty directory would turn it into a confusing agent report.
            //
            // A repo-LESS mission is a different thing entirely, and this guard
            // used to refuse it too — "a microvm phase needs a repository" was
            // the third of three places that assumed so, alongside the inject
            // and the readiness probe. It gets an empty workspace instead, made
            // here so the executor's inject has something to pack.
            if has_repo && !repo.is_dir() {
                return Err(format!(
                    "mission has no checkout at {} — a repo-backed microvm phase \
                     needs one",
                    repo.display()
                ));
            }
            if !has_repo {
                std::fs::create_dir_all(&repo)
                    .map_err(|e| format!("create empty workspace {}: {e}", repo.display()))?;
            }
            crate::microvm_executor::run_phase_in_vm(
                &hub,
                crate::microvm_executor::VmPhase {
                    // Attribution for live output: this is the run a browser
                    // subscribes to for this phase.
                    run_id: Some(run_id),
                    node_id: cm_domain::NodeId::from(node),
                    mission_id,
                    phase_id,
                    iteration,
                    task: &task,
                    backend: backend.as_deref(),
                    repo: &repo,
                    has_repo,
                    team_engine: team_engine.as_deref(),
                    gate: gate.as_ref(),
                    // The solo path is one VM for the whole phase; only a
                    // composed run needs the id qualified per graph node.
                    step: None,
                    // Live tool motion: the tap is drained WHILE the turn runs,
                    // so the World shows a coding phase touching files as it
                    // happens rather than an hour later, all at once.
                    tap_sink: Some(vm_tool_recorder(&pool2, mission_id, phase_id, run_id)),
                },
            )
            .await
        }
        .await;

        // The agent's own account is diagnostic only. Whether the phase succeeded
        // is decided downstream by capture + delivery against the repository.
        // Counted from the guest's own per-subagent transcripts. "-" means the
        // probe could not run, which is not the same as "it delegated to nobody".
        let subagents = match &outcome {
            Ok(o) => o
                .subagents
                .map(|n| n.to_string())
                .unwrap_or_else(|| "?".into()),
            Err(_) => "-".into(),
        };
        // Only meaningful for a mission that asked for a team; "-" otherwise.
        let teammates = match &outcome {
            Ok(o) => o
                .teammates
                .map(|n| n.to_string())
                .unwrap_or_else(|| "-".into()),
            Err(_) => "-".into(),
        };
        // How often the completion gate sent the agent back inside its own turn.
        // "-" is no gate; a number is how many second chances it took, which is
        // the whole measurement of whether the gate is worth its hook.
        let blocked = match &outcome {
            Ok(o) => o
                .stop_blocks
                .map(|n| n.to_string())
                .unwrap_or_else(|| "-".into()),
            Err(_) => "-".into(),
        };
        // What the agent touched inside the VM, recorded before the outcome is
        // consumed. Recorded whatever the phase's verdict: a phase that failed
        // still did work, and the map of what it touched is exactly what makes
        // the failure legible.
        // With a sink attached, `tools` comes back EMPTY by contract — the
        // recorder task has already written every batch, including the final
        // one. Kept as a no-op rather than deleted so a future sink-less path
        // still records; recording the same calls twice is what the empty
        // contract exists to prevent.
        if let Ok(o) = &outcome {
            record_vm_tools(&pool2, mission_id, phase_id, run_id, &o.tools, &[]).await;
            // The gate's own record, on the mission, in the container tier's
            // vocabulary: `gate.inert` when it gave up parsing and allowed
            // calls unchecked, `gate.denied` per call it refused. The guest
            // wrote both files from day one; this is the first reader.
            if let Some(g) = &o.tool_gate {
                if g.inert > 0 {
                    crate::mission_events::record(
                        &pool2,
                        crate::mission_events::MissionEvent::new(
                            mission_id,
                            crate::container_tool_hooks::GATE_INERT,
                        )
                        .phase(phase_id)
                        .run(run_id)
                        .detail(serde_json::json!({ "occurrences": g.inert, "tier": "microvm" })),
                    )
                    .await;
                }
                for line in &g.denied {
                    let detail = serde_json::from_str::<serde_json::Value>(line)
                        .unwrap_or_else(|_| serde_json::json!({ "raw": line }));
                    crate::mission_events::record(
                        &pool2,
                        crate::mission_events::MissionEvent::new(mission_id, "gate.denied")
                            .phase(phase_id)
                            .run(run_id)
                            .detail(detail),
                    )
                    .await;
                }
            }
        }
        // What actually ran: the rootfs the node booted and the CLI the guest
        // reported. Persisted on the run so "which image and version did this
        // mission use" is a query, not an inference from file mtimes — on
        // 2026-09-18 every fleet rootfs had sat on 2.1.223–2.1.226 for a month
        // while the container tier moved on, and nothing had recorded either.
        let vm = serde_json::json!({
            "vm_id": crate::microvm_executor::vm_id_for(phase_id, iteration, None),
            "node_id": target_node_id,
            "backend": backend,
            "rootfs": outcome.as_ref().ok().and_then(|o| o.rootfs.clone()),
            "cli_version": outcome.as_ref().ok().and_then(|o| o.cli_version.clone()),
        });
        let (status, note) = match outcome {
            // The gate gave up. It is the ONLY thing that runs a
            // `done_when_check`, so a release at the cap means the phase's own
            // completion condition was still failing when the agent stopped and
            // nothing downstream will ever re-run it. Completing here is the
            // silent-success shape: green phase, unmet condition, no error.
            Ok(o) if o.released_at_cap == Some(true) => (
                "failed",
                format!(
                    "the completion gate released the agent after {} refusal(s) with its \
                     check still failing: {}",
                    crate::vm_stop_gate::MAX_BLOCKS,
                    o.summary
                ),
            ),
            Ok(o) if o.rc == 0 && o.collected => ("completed", o.summary),
            // A turn that ran and could not be collected is a failure even when
            // the agent was satisfied: the work did not reach the host, so there
            // is nothing for delivery to find.
            Ok(o) if !o.collected => (
                "failed",
                format!(
                    "the agent's work could not be collected from the VM: {}",
                    o.summary
                ),
            ),
            Ok(o) => ("failed", o.summary),
            Err(e) => ("failed", e),
        };
        eprintln!(
            "phase_runner: microvm phase {phase_id} of mission {mission_id} → {status} \
             (subagents: {subagents}, teammates: {teammates}, stop-gate blocks: \
             {blocked}; rootfs: {}, cli: {}) — {}",
            vm["rootfs"].as_str().unwrap_or("?"),
            vm["cli_version"].as_str().unwrap_or("?"),
            note.chars().take(300).collect::<String>()
        );
        // Never overwrite a cancellation. The operator asking to stop is a decision;
        // this task reporting how the VM turned out is an observation, and it may
        // land minutes later. Without the guard a cancelled run silently reappears
        // as completed or failed.
        // Persist the turn as a checkpoint RECORD, not just a status.
        //
        // Everything the UI shows of a run's content reads
        // `topology_runs.checkpoint.records`: `/api/missions/{id}/documents`
        // (the output reader) and `/api/topology-runs/{id}/events` (the live
        // pane). The composed and team tiers write it; the SOLO microVM path
        // never did — measured as `checkpoint IS NULL, records = 0` for every
        // `tier='microvm'` run, against 5 records for `team`.
        //
        // So a solo microVM mission produced real work and showed the operator
        // an empty Live tab and an empty Output tab, with the agent's own
        // account of the turn going to stderr and nowhere else.
        //
        // Shape matches what those two readers already parse — role, node_id,
        // output — so no reader changes.
        let record = serde_json::json!({
            "records": [{
                "node_id": "n0",
                "role": phase_kind,
                "phase": "work",
                "output": note,
                "tokens": 0,
                "gated": [],
            }],
            // Beside `records`, not inside: the two readers parse only `records`.
            "vm": vm,
        });
        if let Err(e) = sqlx::query(
            "UPDATE topology_runs
                SET status = $2,
                    checkpoint = COALESCE(checkpoint, '{}'::jsonb) || $3::jsonb,
                    updated_at = now()
              WHERE id = $1 AND status <> 'cancelled'",
        )
        .bind(run_id)
        .bind(status)
        .bind(&record)
        .execute(&pool2)
        .await
        {
            eprintln!("phase_runner: could not close microvm run {run_id}: {e}");
        }
    });

    eprintln!(
        "phase_runner: mission {mission_id} phase {phase_id} launched in a MICROVM \
         on node {target_node_id:?}"
    );
    Ok(())
}

async fn launch_direct_session(
    pool: &PgPool,
    mission_id: Uuid,
    phase_id: Uuid,
    workspace_id: Uuid,
    iteration: i32,
    task: &str,
) -> Result<(), String> {
    record_phase_prompt(pool, mission_id, phase_id, "session", task).await;
    sqlx::query(
        "DELETE FROM topology_runs
          WHERE mission_phase_id = $1 AND status IN ('failed', 'cancelled')",
    )
    .bind(phase_id)
    .execute(pool)
    .await
    .map_err(|e| format!("purge prior runs for phase {phase_id}: {e}"))?;

    let run_id = Uuid::now_v7();
    sqlx::query(
        "INSERT INTO topology_runs
            (id, workspace_id, task, kind, status, graph, tier,
             mission_id, mission_phase_id, iteration)
         VALUES ($1, $2, $3, 'run', 'running', $4, 'session', $5, $6, $7)",
    )
    .bind(run_id)
    .bind(workspace_id)
    .bind(task)
    .bind(serde_json::json!({ "nodes": [], "edges": [], "executor": "session" }))
    .bind(mission_id)
    .bind(phase_id)
    .bind(iteration)
    .execute(pool)
    .await
    .map_err(|e| format!("enqueue session run for phase {phase_id}: {e}"))?;

    mark_phase_running(pool, mission_id, phase_id).await?;

    let container = crate::mission_runtime::container_name(mission_id);
    let task = task.to_string();
    let pool = pool.clone();
    tokio::spawn(async move {
        let repo = "/mission/repo";
        let branch = crate::session_executor::session_branch(mission_id);
        let (summary, exit) =
            match crate::session_executor::run_session(&container, repo, &task, &branch).await {
                Ok(v) => v,
                Err(e) => (format!("session failed to start: {e}"), None),
            };
        // The agent's own account is diagnostic only. Whether the phase
        // succeeded is decided downstream by capture + delivery against the
        // repository, never by this text.
        let ok = exit == Some(0);
        eprintln!(
            "phase_runner: session for mission {mission_id} phase {phase_id} exited {exit:?} — {}",
            summary.chars().take(200).collect::<String>()
        );
        let status = if ok { "completed" } else { "failed" };

        // The agent's account, kept. Until now this tier wrote NO checkpoint
        // record — the same defect the solo microVM path was fixed for, in the
        // one remaining tier: `summary` went to stderr above and nowhere else,
        // so the Live and Output tabs were empty for a session that did real
        // work, and nothing downstream could read what the agent said it did.
        //
        // Shape matches the microVM path exactly, so the two existing readers
        // (`/api/missions/{id}/documents` and `/api/topology-runs/{id}/events`)
        // need no change.
        let record = serde_json::json!({
            "records": [{
                "node_id": "n0",
                "role": "session",
                "phase": "work",
                "output": summary,
                "tokens": 0,
                "gated": [],
            }]
        });
        // Also as a durable event, so the narrative is queryable per mission
        // rather than only by walking topology_runs JSON.
        let mut ev = crate::mission_events::MissionEvent::new(
            mission_id,
            crate::mission_events::REASONING,
        );
        ev.phase_id = Some(phase_id);
        ev.run_id = Some(run_id);
        ev.target = Some("session".to_string());
        ev.detail = serde_json::json!({ "text": summary });
        crate::mission_events::record(&pool, ev).await;

        // Never overwrite a cancellation. The operator asking to stop is a decision;
        // this task reporting how the VM turned out is an observation, and it may
        // land minutes later. Without the guard a cancelled run silently reappears
        // as completed or failed.
        if let Err(e) = sqlx::query(
            "UPDATE topology_runs
                SET status = $2,
                    checkpoint = COALESCE(checkpoint, '{}'::jsonb) || $3::jsonb,
                    updated_at = now()
              WHERE id = $1 AND status <> 'cancelled'",
        )
        .bind(run_id)
        .bind(status)
        .bind(&record)
        .execute(&pool)
        .await
        {
            eprintln!("phase_runner: could not close session run {run_id}: {e}");
        }
    });

    eprintln!("phase_runner: mission {mission_id} phase {phase_id} launched as a DIRECT SESSION");
    Ok(())
}

/// The pinned skills for a phase's crew, rendered for the task text.
///
/// The container tier gets skills per TURN (`topology_exec::pinned_skills_text`),
/// where each node carries its own claw alias and therefore its own role's
/// skills. The microVM and direct-session tiers have no per-turn alias — the
/// phase runs as one `claude -p` session — so their skills have to be resolved
/// per PHASE and appended to the task instead. Without this, those three tiers
/// deliver no skill at all, which is what they did until now.
///
/// Union across the crew, deduplicated. A solo tier does not know which crew
/// member's turn it is running, and a procedure that applies to the role doing
/// the work still applies when one agent does all of it. Erring toward the
/// union is safe here in a way it would not be on the container tier, where
/// per-role precision is available and used.
/// Record the prompt a phase is about to run with.
///
/// Called from inside each tier rather than at the dispatch point, because the
/// tiers do not all send the same text: the three solo tiers send the
/// skills-bearing task, and the container tier sends the bare task and appends
/// skills per turn in `topology_exec`. Recording at the fork wrote a `solo`
/// prompt for container missions too — a prompt that was composed and never
/// sent. A provenance record of something that did not happen is worse than no
/// record: it is the wrong answer, delivered confidently.
async fn record_phase_prompt(
    pool: &PgPool,
    mission_id: Uuid,
    phase_id: Uuid,
    tier: &str,
    text: &str,
) {
    let mut ev = crate::mission_events::MissionEvent::new(
        mission_id,
        crate::mission_events::PROMPT_COMPOSED,
    );
    ev.phase_id = Some(phase_id);
    ev.target = Some(tier.to_string());
    ev.detail = serde_json::json!({ "text": text, "tier": tier });
    crate::mission_events::record(pool, ev).await;
}

pub async fn phase_skills_text(pool: &PgPool, mission_id: Uuid) -> Option<String> {
    let crew = sqlx::query(
        "SELECT DISTINCT a.id
           FROM team_members tm
           JOIN mission_teams mt ON mt.team_id = tm.team_id
           JOIN agents a         ON a.id = tm.claw_id
          WHERE mt.mission_id = $1 AND a.deleted_at IS NULL",
    )
    .bind(mission_id)
    .fetch_all(pool)
    .await
    .unwrap_or_default();

    let mut seen: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
    let mut out = String::new();
    for row in &crew {
        let agent_id: Uuid = row.get("id");
        let link = cm_db::repo::agent_template_link::get(pool, agent_id)
            .await
            .ok()
            .flatten();
        let (tpl_id, slot) = link
            .as_ref()
            .map(|l| (Some(l.template_id), Some(l.role_slot.as_str())))
            .unwrap_or((None, None));
        let Ok(bindings) =
            cm_db::repo::skills_catalog::effective_for_agent(pool, agent_id, tpl_id, slot).await
        else {
            continue;
        };
        for b in bindings.iter().filter(|b| b.pin_in_context) {
            if !seen.insert(b.skill.name.clone()) {
                continue;
            }
            // Bounded, and truncation is STATED. A silently clipped procedure is
            // worse than an absent one: the agent follows the half it can see
            // and reports success against a rule it never read.
            if out.len() + b.skill.body.len() > crate::topology_exec::MAX_PINNED_SKILL_BYTES {
                out.push_str(&format!(
                    "\n[skill \"{}\" omitted — the pinned set exceeded {} bytes]\n",
                    b.skill.name,
                    crate::topology_exec::MAX_PINNED_SKILL_BYTES
                ));
                continue;
            }
            out.push_str(&crate::topology_exec::render_pinned_skill(
                &b.skill.name,
                &b.skill.body,
            ));
        }
    }
    if seen.is_empty() {
        return None;
    }
    Some(out)
}

fn phase_task_text(
    kind: &str,
    title: &str,
    description: Option<&str>,
    phase_task: Option<&str>,
    // Whether the mission has a repository bound. A repo-less mission's
    // `/mission/repo` is a scratch directory, and telling its agents otherwise
    // is how eight research documents were written into a container that was
    // then reaped unread.
    has_repo: bool,
) -> String {
    let base = description.unwrap_or("").trim();
    // These are Claude Code's OWN tool names, because every executor that runs a
    // mission turn ends in `claude -p`: the microVM passes
    // `--allowedTools Read Edit Write Bash Agent`
    // (`microvm_executor::LEAD_TOOLS`), the direct path passes
    // `Read Edit Write Bash` (`session_executor::ALLOWED_TOOLS`), and the
    // container tier binds every claw to `claude_cli.default`
    // (`runtime_provision::provider_alias_for`), whose subprocess gets Claude
    // Code's native toolset — ZeroClaw's own tool gating "never reaches the
    // subprocess" (see the note above `direct_mode`).
    //
    // This block previously advertised ZeroClaw tool names (`file_edit`,
    // `content_search`, …) and explicitly told the agent that `bash` did NOT
    // exist. On every tier in service that was backwards: those tools were the
    // ones absent, and Bash was one of the ones present. Agents answered by
    // describing the mismatch and asking what to do — five of them, on one
    // mission, for 7.4k tokens and zero artifacts.
    let tool_preamble = format!(
        "\
        TOOLS AVAILABLE (Claude Code's standard tools — use these exact names):\n\
        - Read   — read a file\n\
        - Edit   — modify an existing file\n\
        - Write  — create or overwrite a file\n\
        - Bash   — run a shell command\n\
        - Glob   — find files by path glob\n\
        - Grep   — search file contents\n\
        \n\
        {workspace}\n\
        All file operations resolve there — use absolute paths under it, or cd\n\
        there first. Write your outputs as REAL files with Write/Edit — do NOT\n\
        paste code blocks in your reply expecting the platform to save them;\n\
        nothing else writes files for you.\n",
        workspace = if has_repo {
            "WORKSPACE: Your working directory is /mission/repo. That path is the\n\
             mission's git checkout."
        } else {
            // Saying "git checkout" here to a mission that has none is what
            // produced the eight destroyed ClawHDF5 documents: the agent looked,
            // found no repo, wrote the files anyway, and nothing collected them.
            // Now `mission_outputs` DOES collect them, and the agent is told so
            // — an instruction the platform can actually keep.
            "WORKSPACE: Your working directory is /mission/repo. It EXISTS and is writable.\n\
             This mission has NO git repository — that path is a scratch workspace, so\n\
             there is nothing to commit or push. Every file you leave there is collected\n\
             when the phase ends and published as a mission artifact, so write your\n\
             output as files exactly as you would in a repo."
        }
    );
    // The INT-XX markers are a machine contract, not a style preference:
    // task_card_parser.rs scans turn output line-by-line for these literals and
    // materializes `mission_tasks` rows from them. The rules used to live only
    // in the team-template role prompts -- which are never injected into mission
    // turns (runtime_provision.rs writes model/risk_profile/mcp_bundles and
    // nothing else) -- and in a skill the agent had to choose to fetch. So the
    // parser's contract was stated nowhere the agent reliably saw it. It is
    // stated here because this is the one text every mission turn receives.
    let marker_protocol = "\
        TASK MARKERS (parsed literally, line by line — this is a machine contract):\n\
        Emit these on their own line, with the colon, no bold, no code fence,\n\
        exactly one INT id per line, at the END of a substantive turn:\n\
        - TASK: INT-NN — <title>          open a new item\n\
        - WORK: INT-NN                    started implementing\n\
        - HANDOFF: INT-NN                 passed to test/review\n\
        - TEST_PASS: INT-NN               tests green\n\
        - TEST_FAIL: INT-NN — <reason>    build/tests failed\n\
        - REVIEW_APPROVE: INT-NN          diff approved\n\
        - REVIEW_BLOCK: INT-NN — <reason> changes requested\n\
        - COMPLETED: INT-NN               done and pushed\n\
        Never emit a marker you can't back up — COMPLETED without a corresponding\n\
        commit desynchronizes the mission from the repo.\n";
    let directive = match kind {
        "research" => {
            "Your team is running the RESEARCH phase of this mission. \
             Investigate the topic, gather sources, and produce a \
             sectioned Markdown brief the coding phase can implement \
             directly. Save findings under /mission/repo/research/ \
             using Write — one Markdown file per topic. Emit INT-XX \
             task markers in the last file for concrete follow-ups."
        }
        "coding" => {
            "Your team is running the CODING phase of this mission. \
             Implement the mission's acceptance criteria against the \
             /mission/repo checkout using Write/Edit for every source \
             file, then git via Bash to commit small focused changes \
             with test coverage. Emit COMPLETED: <INT-id> markers as you \
             close research-produced tasks. Do NOT respond with source \
             code in text — write it as files."
        }
        "benchmark" => {
            "Your team is running the BENCHMARK phase of this mission. \
             Author or extend benchmarks under /mission/repo/benches or \
             the crate's bench harness using Write/Edit. Baseline the \
             pre-change performance, apply the change (or use the \
             mission's committed diff), then measure after."
        }
        "security_scan" => {
            "Your team is running the SECURITY SCAN phase of this mission. \
             Run cargo-audit, gitleaks, trivy, and semgrep against \
             /mission/repo. Triage findings, file INT-XX task markers \
             for remediation, propose patches for the coding phase."
        }
        _ => "Execute this mission phase according to the mission brief.",
    };
    // The mission brief is shared by every phase; this block is not. It goes
    // last and says so explicitly, because the failure it fixes was agents
    // re-doing the whole mission in each phase rather than their slice of it.
    let scope = match phase_task.map(str::trim).filter(|t| !t.is_empty()) {
        Some(t) => format!(
            "\n\nTHIS PHASE'S TASK — do this and only this. The brief above is \
             the mission's full scope across all phases; the following is your \
             share of it:\n{t}"
        ),
        None => String::new(),
    };
    format!(
        "MISSION: {title}\n\n{tool_preamble}\n{marker_protocol}\n{directive}\n\nBRIEF:\n{base}{scope}"
    )
}

/// Record a microVM turn's drained tool tap.
///
/// `agent_id` is deliberately absent: a microVM phase has no platform agent, so
/// there is no pawn to attribute the touch to. Inventing one would put a named
/// crew member's face on work a VM did alone.
/// A sink that records tool observations into `mission_events` as they arrive,
/// plus the task draining it.
///
/// The channel exists so the VM executor stays free of the database: it
/// observes, this records. The task ends when the sender is dropped, which
/// happens when the phase's `VmPhase` goes out of scope — so there is no
/// lifetime to manage and no way to leak one per phase.
pub(crate) fn vm_tool_recorder(
    pool: &PgPool,
    mission_id: Uuid,
    phase_id: Uuid,
    run_id: Uuid,
) -> tokio::sync::mpsc::UnboundedSender<Vec<crate::vm_tool_tap::Observed>> {
    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<Vec<crate::vm_tool_tap::Observed>>();
    let pool = pool.clone();
    tokio::spawn(async move {
        while let Some(batch) = rx.recv().await {
            record_vm_tools(&pool, mission_id, phase_id, run_id, &batch, &[]).await;
        }
    });
    tx
}

/// Which agent each observed call belongs to, by session.
///
/// The tap is per-CONTAINER and every role in a phase shares one, so a phase's
/// calls arrive as one undifferentiated stream and `agent_id` was written
/// `None` for all of them. That is why every Skill-Use score is per-mission
/// rather than per-role, and why the World's per-agent view gets nothing from
/// this tier.
///
/// One `claude -p` invocation is one turn is one agent, and Claude Code stamps
/// each invocation with a `session_id`. So the distinct sessions, in the order
/// they first appear, are the phase's turns in the order they ran — and
/// `prompt.composed` already records the agent of each turn in that same order.
///
/// # It attributes nothing rather than guessing
///
/// Only when the counts match exactly. A phase whose sessions and turns differ
/// in number has something this correlation does not model — a retry, a turn
/// that called no tool, two agents genuinely concurrent — and a
/// plausible-looking wrong attribution is worse here than none: it would put
/// one agent's `git push` on another agent's record, which is the sort of thing
/// a person later reasons from.
pub(crate) fn attribute_sessions(
    tools: &[crate::vm_tool_tap::Observed],
    turn_agents: &[Uuid],
) -> Vec<Option<Uuid>> {
    let mut order: Vec<&str> = Vec::new();
    for t in tools {
        let Some(sid) = t.session.as_deref() else {
            // A single unattributable call means the sequence has a hole in it,
            // and a hole shifts every later session onto the wrong turn.
            return vec![None; tools.len()];
        };
        if !order.contains(&sid) {
            order.push(sid);
        }
    }
    if order.len() != turn_agents.len() || order.is_empty() {
        return vec![None; tools.len()];
    }
    tools
        .iter()
        .map(|t| {
            let sid = t.session.as_deref()?;
            let idx = order.iter().position(|s| *s == sid)?;
            turn_agents.get(idx).copied()
        })
        .collect()
}

pub(crate) async fn record_vm_tools(
    pool: &PgPool,
    mission_id: Uuid,
    phase_id: Uuid,
    run_id: Uuid,
    tools: &[crate::vm_tool_tap::Observed],
    turn_agents: &[Uuid],
) {
    if tools.is_empty() {
        return;
    }
    let owners = attribute_sessions(tools, turn_agents);
    if owners.iter().all(Option::is_none) && !turn_agents.is_empty() {
        eprintln!(
            "phase_runner: {} tool call(s) for phase {phase_id} could not be \
             attributed to an agent ({} session(s) across {} turn(s)) — recorded \
             unattributed rather than guessed",
            tools.len(),
            tools
                .iter()
                .filter_map(|t| t.session.as_deref())
                .collect::<std::collections::HashSet<_>>()
                .len(),
            turn_agents.len()
        );
    }
    let mut events = Vec::new();
    for (t, owner) in tools.iter().zip(owners) {
        events.push(crate::mission_events::MissionEvent {
            mission_id,
            phase_id: Some(phase_id),
            run_id: Some(run_id),
            agent_id: owner,
            kind: crate::mission_events::TOOL_CALL.to_string(),
            target: Some(t.tool.clone()),
            // `path` because the World's SSE reads `detail.path` for this kind
            // and was handed a null on every container-tier call; `input`
            // because the tool name alone cannot answer a single behavioural
            // question about the phase.
            detail: serde_json::json!({
                "path": t.path,
                "input": t.input,
                // Only commands carry one; `bounded_response` returns null for
                // everything else, and a null key here is noise.
                "response": t.response,
                // Null unless a SUBAGENT made this call. It shares its parent's
                // session id, so `agent_id` above names the agent that spawned
                // it and this is the only thing saying the parent did not run
                // it itself.
                "subagent": t.subagent,
                "subagent_id": t.subagent_id,
            }),
        });
        if let Some(path) = &t.path {
            events.push(crate::mission_events::MissionEvent {
                mission_id,
                phase_id: Some(phase_id),
                run_id: Some(run_id),
                agent_id: owner,
                kind: crate::mission_events::FILE_TOUCH.to_string(),
                target: Some(crate::mission_events::repo_relative(
                    path,
                    &["/mission/repo", "/workspace"],
                )),
                // The ABSOLUTE path as well as the repo-relative one. `target`
                // is normalised for the map, where a `mission` → `repo` pair of
                // directory orbs means nothing to a reader — but normalising is
                // exactly what destroys the question "did this write land
                // outside the checkout", which is the one boundary a skill can
                // be scored on.
                detail: serde_json::json!({ "tool": t.tool, "abs": path }),
            });
        }
    }
    crate::mission_events::record_all(pool, events).await;
}

/// Flip a phase from `pending` to `running`, and say so on the wire.
///
/// The `WHERE … AND status = 'pending'` guard means this UPDATE is a claim, not
/// an assignment: a phase already claimed by another launcher matches nothing.
/// `RETURNING` is what turns that into information — without it the statement
/// reports the same `Ok(())` whether it started a phase or lost the race, and
/// the World would announce a start that never happened.
///
/// Five copies of this UPDATE existed, one per launch path. They were identical
/// and independently maintained, which is how a sixth path would have been
/// written with no event at all.
async fn mark_phase_running(pool: &PgPool, mission_id: Uuid, phase_id: Uuid) -> Result<(), String> {
    let claimed = sqlx::query(
        "UPDATE mission_phases
            SET status = 'running', started_at = now()
          WHERE id = $1 AND status = 'pending'
      RETURNING id",
    )
    .bind(phase_id)
    .fetch_optional(pool)
    .await
    .map_err(|e| format!("mark phase {phase_id} running: {e}"))?;
    if claimed.is_some() {
        crate::mission_events::record(
            pool,
            crate::mission_events::MissionEvent::new(
                mission_id,
                crate::mission_events::PHASE_STARTED,
            )
            .phase(phase_id),
        )
        .await;
    }
    Ok(())
}

/// Close phases whose topology_runs are all terminal.
///
/// A phase that declares a `done_when` condition lands in `evaluating` instead
/// of `completed`; [`evaluate_finished_phases`] judges it and decides whether
/// to finish or run another pass. A failed run still fails the phase outright
/// — there is nothing to evaluate — and a phase with no condition completes
/// exactly as it always did, so untouched missions are unaffected.
async fn close_finished_phases(pool: &PgPool) -> Result<(), String> {
    let closed = sqlx::query(
        "UPDATE mission_phases mp
            SET status =
              CASE
                WHEN EXISTS (
                  SELECT 1 FROM topology_runs r
                  WHERE r.mission_phase_id = mp.id
                    AND r.iteration = mp.iteration
                    AND r.status = 'failed'
                ) THEN 'failed'
                WHEN mp.done_when IS NOT NULL AND btrim(mp.done_when) <> '' THEN 'evaluating'
                ELSE 'completed'
              END,
              completed_at =
                CASE
                  WHEN mp.done_when IS NOT NULL AND btrim(mp.done_when) <> ''
                       AND NOT EXISTS (
                         SELECT 1 FROM topology_runs r
                         WHERE r.mission_phase_id = mp.id
                           AND r.iteration = mp.iteration
                           AND r.status = 'failed'
                       )
                  THEN NULL ELSE now()
                END
          WHERE mp.status = 'running'
            AND EXISTS (
              SELECT 1 FROM topology_runs r
              WHERE r.mission_phase_id = mp.id AND r.iteration = mp.iteration
            )
            AND NOT EXISTS (
              SELECT 1 FROM topology_runs r
              WHERE r.mission_phase_id = mp.id
                AND r.iteration = mp.iteration
                AND r.status NOT IN ('completed', 'failed', 'cancelled')
            )
      RETURNING mp.id, mp.mission_id, mp.status",
    )
    .fetch_all(pool)
    .await
    .map_err(|e| format!("close finished phases: {e}"))?;
    // `RETURNING` rather than a follow-up SELECT, and this is the only way to
    // get these rows: the `CASE` decides each phase's status INSIDE the
    // statement, from `topology_runs` rows whose state the statement itself
    // does not change — so re-deriving it afterwards would be a second
    // implementation of that CASE, free to disagree with the first. Without it
    // this function emits zero `phase.completed` events and reports success.
    for row in closed {
        let phase_id: Uuid = row.get("id");
        let mission_id: Uuid = row.get("mission_id");
        let status: String = row.get("status");
        // `evaluating` is not terminal — the judge has not spoken yet — so it
        // is not a completion. `evaluate_finished_phases` closes those.
        if status == "evaluating" {
            continue;
        }
        crate::mission_events::record(
            pool,
            crate::mission_events::MissionEvent::new(
                mission_id,
                crate::mission_events::PHASE_COMPLETED,
            )
            .phase(phase_id)
            .detail(serde_json::json!({ "status": status })),
        )
        .await;
    }
    Ok(())
}

/// Judge every phase sitting in `evaluating` against its `done_when`.
///
/// Met, or out of iterations → `completed`. Otherwise the phase goes back to
/// `pending` with `iteration` bumped, and [`start_pending_phases`] relaunches
/// it; the verdict's reason is carried into the next pass's task text by
/// [`phase_task_text`] so the agents are told what was missing.
async fn evaluate_finished_phases(
    pool: &PgPool,
    runtime: &cm_runtime::Runtime,
) -> Result<(), String> {
    let rows = sqlx::query(
        "SELECT mp.id, mp.mission_id, mp.kind, mp.done_when, mp.max_iterations, mp.iteration,
                m.runtime_kind
           FROM mission_phases mp
           JOIN missions m ON m.id = mp.mission_id
          WHERE mp.status = 'evaluating' AND m.status = 'running'
            AND (mp.judge_retry_after IS NULL OR mp.judge_retry_after <= now())
          LIMIT 5",
    )
    .fetch_all(pool)
    .await
    .map_err(|e| format!("select evaluating phases: {e}"))?;

    for row in rows {
        let phase_id: Uuid = row.get("id");
        let mission_id: Uuid = row.get("mission_id");
        let kind: String = row.get("kind");
        let condition: String = row
            .get::<Option<String>, _>("done_when")
            .unwrap_or_default();
        let max_iterations: i32 = row.get("max_iterations");
        let iteration: i32 = row.get("iteration");
        let runtime_kind: String = row.get("runtime_kind");

        // Pull the agent's work onto the host BEFORE judging it.
        //
        // `Sandbox::for_mission` copies from the HOST checkout. In copy mode the
        // agents write inside the container, and their work only reaches the host
        // when `sync_out` runs — which used to happen in the capture sweep AFTER
        // the phase closed. So a phase was judged against a tree that did not yet
        // contain the pass being judged, and the judge truthfully reported
        // nothing there.
        //
        // Mission 01a00cfa is the proof: pass 2 wrote a 434-line
        // IMPLEMENTATION_BRIEF.md whose tests passed and which was pushed to a
        // clean branch, and the verdict on it was
        // "research/IMPLEMENTATION_BRIEF.md does not exist anywhere" — logged
        // one line BEFORE the capture that collected it.
        //
        // Skipped for microVM, whose work is collected out of the guest by
        // `microvm_executor` over this same host path before the VM is
        // destroyed — the identical carve-out the capture sweep makes.
        if crate::mission_fs::copy_mode() && runtime_kind != "microvm" {
            let container = crate::mission_runtime::container_name(mission_id);
            if let Err(e) = crate::mission_fs::sync_out(&container, mission_id).await {
                // Judging now would evaluate a stale tree and, on a last pass,
                // FAIL a phase that succeeded. Leave the phase `evaluating` so
                // the next sweep retries rather than recording a verdict nobody
                // could stand behind.
                eprintln!(
                    "phase_runner: could NOT collect work from {container} before judging \
                     phase {phase_id} ({e}) — leaving it to the next sweep rather than \
                     judging a stale tree"
                );
                continue;
            }
        }

        let evidence = crate::phase_summarizer::collect_evidence(pool, mission_id, phase_id)
            .await
            .unwrap_or_else(|e| format!("(evidence collection failed: {e})"));

        let verdict = crate::evaluator::evaluate(runtime, mission_id, &condition, &evidence).await;
        if let Err(e) =
            crate::evaluator::record(pool, mission_id, phase_id, iteration, &verdict).await
        {
            eprintln!("phase_runner: recording evaluation for {phase_id} failed: {e}");
        }

        // A judge that could not be REACHED has not judged. `Verdict.error` is
        // set only when the evaluator itself failed — "could not judge" as
        // distinct from "judged incomplete" — and spending one of the phase's
        // passes on it charges the agent for an outage it had no part in.
        //
        // Mission 01a011bf lost its script phase exactly this way: the work was
        // done and committed, glm-5.3 returned "transport error: error decoding
        // response body", and with two passes budgeted that single unreachable
        // judge was enough to fail the phase. Leave it `evaluating` so the next
        // sweep re-judges the same pass; the phase is not re-run, only re-read.
        //
        // Deliberately NOT a fallback to the agent's own provider: `evaluate`
        // already refuses that, because a judge from the same family is not an
        // independent check and quietly becoming one is worse than waiting.
        let mut judge_gave_up = false;
        if let Some(why) = verdict.error.as_deref() {
            let blocked_for: Option<f64> = sqlx::query_scalar(
                "UPDATE mission_phases
                    SET judge_blocked_since = COALESCE(judge_blocked_since, now())
                  WHERE id = $1
              RETURNING EXTRACT(EPOCH FROM now() - judge_blocked_since)::float8",
            )
            .bind(phase_id)
            .fetch_optional(pool)
            .await
            .map_err(|e| format!("mark judge-blocked {phase_id}: {e}"))?
            .flatten();

            // An exhausted plan names the time it resets. Waiting cannot
            // reach it, so stop now rather than spending the window — and say
            // which of the two very different problems this is.
            if let Some(plain) = judge_error_is_exhausted_plan(why) {
                eprintln!(
                    "phase_runner: phase {phase_id} ({kind}) — NOT retrying: {plain}. \
                     Pass {} of {} NOT consumed; the agent's work is untouched and the \
                     phase is failing on the judge, not on itself.",
                    iteration + 1,
                    max_iterations,
                );
                judge_gave_up = true;
            } else if blocked_for.unwrap_or(0.0) < JUDGE_WAIT_MAX_SECS {
                let wait = judge_backoff_secs(blocked_for.unwrap_or(0.0));
                let _ = sqlx::query(
                    "UPDATE mission_phases
                        SET judge_retry_after = now() + make_interval(secs => $2)
                      WHERE id = $1",
                )
                .bind(phase_id)
                .bind(wait)
                .execute(pool)
                .await;
                eprintln!(
                    "phase_runner: phase {phase_id} ({kind}) — judge unreachable ({why}); \
                     retrying in {wait:.0}s. Pass {} of {} NOT consumed; blocked {:.0}s of \
                     {JUDGE_WAIT_MAX_SECS:.0}s.",
                    iteration + 1,
                    max_iterations,
                    blocked_for.unwrap_or(0.0)
                );
                continue;
            } else {
                // Waited long enough. Fail with the transport reason rather
                // than sitting `evaluating` forever — an invisible hang is
                // worse than an honest failure that names what could not be
                // reached.
                eprintln!(
                    "phase_runner: phase {phase_id} ({kind}) — judge unreachable for {:.0}s, \
                     giving up: {why}",
                    blocked_for.unwrap_or(0.0)
                );
                // Fail NOW rather than requeueing. Re-running the phase would
                // spend a container and a model budget re-doing work that was
                // never the problem — the judge was.
                judge_gave_up = true;
            }
        } else {
            // A real verdict landed: stop the clock.
            let _ = sqlx::query(
                "UPDATE mission_phases SET judge_blocked_since = NULL, judge_retry_after = NULL
                  WHERE id = $1 AND (judge_blocked_since IS NOT NULL
                                     OR judge_retry_after IS NOT NULL)",
            )
            .bind(phase_id)
            .execute(pool)
            .await;
        }

        let last_pass = iteration + 1 >= max_iterations;
        if verdict.met || last_pass || judge_gave_up {
            // A phase that ran out of passes WITHOUT meeting its condition did not
            // succeed, and must not say it did. This used to mark both outcomes
            // `completed`: the verdict recorded met=false while the phase — and
            // through `close_finished_missions`, the whole mission — reported
            // success. Anything reading mission status rather than digging into the
            // verdict saw a goal that was never reached as a goal achieved.
            //
            // Found by the Goodhart test for the independent judge: glm-4.7
            // correctly refused a phase whose suite had a failing test, and the
            // mission still closed `completed`.
            let status = if verdict.met { "completed" } else { "failed" };
            let closed = sqlx::query(
                "UPDATE mission_phases SET status = $2, completed_at = now()
                  WHERE id = $1 AND status = 'evaluating'
              RETURNING id",
            )
            .bind(phase_id)
            .bind(status)
            .fetch_optional(pool)
            .await
            .map_err(|e| format!("close phase {phase_id}: {e}"))?;
            if closed.is_some() {
                crate::mission_events::record(
                    pool,
                    crate::mission_events::MissionEvent::new(
                        mission_id,
                        crate::mission_events::PHASE_COMPLETED,
                    )
                    .phase(phase_id)
                    .detail(serde_json::json!({
                        "status": status,
                        "judge": verdict.model,
                        "reason": verdict.reason,
                    })),
                )
                .await;
            }
            eprintln!(
                "phase_runner: phase {phase_id} ({kind}) {status} after {} pass(es) — met={} \
                 (judge={}, independent={}) — {}",
                iteration + 1,
                verdict.met,
                verdict.model,
                verdict.independent,
                verdict.reason
            );
        } else {
            sqlx::query(
                "UPDATE mission_phases
                    SET status = 'pending', iteration = iteration + 1, started_at = NULL
                  WHERE id = $1 AND status = 'evaluating'",
            )
            .bind(phase_id)
            .execute(pool)
            .await
            .map_err(|e| format!("requeue phase {phase_id}: {e}"))?;
            eprintln!(
                "phase_runner: phase {phase_id} ({kind}) not met after pass {} of {max_iterations} — {}",
                iteration + 1,
                verdict.reason
            );
        }
    }
    Ok(())
}

/// Close missions whose phases are all terminal.
/// Mark as `skipped` the phases a failed phase has made unreachable.
///
/// `start_pending_phases` launches a phase only when EVERY lower-order phase is
/// `completed`, so once one fails the phases after it can never run. They sat at
/// `pending` forever — and `close_finished_missions` requires no phase to be
/// non-terminal, so the MISSION never finished either. It stayed `running`
/// indefinitely, which meant `mission_runtime`'s sweeper (which fires N minutes
/// after a terminal state) never reaped its container.
///
/// Found by counting containers, not by a test: gw-04 was holding a runtime
/// container for a mission whose only run failed three days earlier, phases
/// `pending,failed`. One leaked container per failed multi-phase mission,
/// accumulating silently.
///
/// `skipped` is not a new concept — `close_finished_missions` already treats it
/// as terminal, and it is the honest word: those phases were not run and never
/// will be, which is different from having failed.
async fn skip_unreachable_phases(pool: &PgPool) -> Result<(), String> {
    let n = sqlx::query(
        "UPDATE mission_phases mp
            SET status = 'skipped', completed_at = now()
          WHERE mp.status = 'pending'
            AND EXISTS (
              SELECT 1 FROM missions m
               WHERE m.id = mp.mission_id AND m.status = 'running'
            )
            -- Strictly EARLIER, because order is what makes a phase
            -- unreachable. A failure later in the list says nothing about a
            -- phase that is still waiting its turn ahead of it.
            AND EXISTS (
              SELECT 1 FROM mission_phases prior
               WHERE prior.mission_id = mp.mission_id
                 AND prior.order_idx < mp.order_idx
                 AND prior.status = 'failed'
            )",
    )
    .execute(pool)
    .await
    .map_err(|e| format!("skip unreachable phases: {e}"))?
    .rows_affected();
    if n > 0 {
        eprintln!(
            "phase_runner: skipped {n} phase(s) made unreachable by an earlier failure —              retrying the failed phase reopens them"
        );
    }
    Ok(())
}

async fn close_finished_missions(pool: &PgPool) -> Result<(), String> {
    sqlx::query(
        "UPDATE missions m
            SET status =
              CASE
                WHEN EXISTS (
                  SELECT 1 FROM mission_phases mp
                  WHERE mp.mission_id = m.id AND mp.status = 'failed'
                ) THEN 'failed'
                ELSE 'completed'
              END,
              completed_at = now(),
              updated_at = now()
          WHERE m.status = 'running'
            AND EXISTS (SELECT 1 FROM mission_phases mp WHERE mp.mission_id = m.id)
            AND NOT EXISTS (
              SELECT 1 FROM mission_phases mp
              WHERE mp.mission_id = m.id
                AND mp.status NOT IN ('completed', 'failed', 'skipped')
            )
            -- A repo-bearing mission must not be declared finished before its
            -- work has been captured. Capture is batched (CAPTURE_BATCH per
            -- tick) and runs AFTER a phase completes, so without this a
            -- backlogged mission closes as 'completed' and only then does
            -- capture discover the phase delivered nothing — leaving a
            -- 'completed' mission holding a 'failed' phase, with the mission
            -- status unfixable because the CASE above only touches 'running'
            -- rows. Waiting a tick costs nothing; the alternative is a mission
            -- whose headline outcome contradicts its own phases.
            AND NOT EXISTS (
              SELECT 1 FROM mission_phases mp
              WHERE mp.mission_id = m.id
                AND mp.status = 'completed'
                AND m.repo_id IS NOT NULL
                AND NOT EXISTS (
                  SELECT 1 FROM mission_artifacts a
                   WHERE a.mission_id = mp.mission_id
                     AND a.phase_id = mp.id
                     AND a.kind = 'code_diff'
                )
            )",
    )
    .execute(pool)
    .await
    .map_err(|e| format!("close finished missions: {e}"))?;
    Ok(())
}

/// Walk the team's graph nodes and bind each to its claw as
/// `node.attrs["agent"] = "claw_<hex>"`, based on the
/// `team_members(team_id, node_id, claw_id)` map. Nodes without a
/// matching member row are left alone (the executor will fall through to
/// the env alias map / default).
///
/// IMPORTANT — the alias MUST live under `attrs`, not at the node's top
/// level. `cm_topology::graph::Node` only deserializes `{id, role, level,
/// attrs}`, so a top-level `"agent"` key is silently dropped by serde,
/// `TurnRequest::agent` comes back `None`, and every turn falls back to
/// `ZEROCLAW_DEFAULT_AGENT` (`scout`) — which is jailed to scout's own
/// workspace and cannot see `/mission/repo`. That produced whole missions
/// of agents burning tokens while reporting an "empty greenfield"
/// workspace. The top-level key is still written for display/debug, but
/// `attrs` is what actually binds. See `topology_exec::run_turn`.
///
/// The graph shape we care about: `{ "nodes": [ { "id": "n0", ... } ] }`.
/// Non-object graphs (or graphs without a nodes array) are returned
/// unchanged.
async fn inject_node_agents(
    pool: &sqlx::PgPool,
    team_id: Uuid,
    graph: serde_json::Value,
) -> serde_json::Value {
    let members = match sqlx::query("SELECT node_id, claw_id FROM team_members WHERE team_id = $1")
        .bind(team_id)
        .fetch_all(pool)
        .await
    {
        Ok(rows) => rows,
        Err(e) => {
            eprintln!("phase_runner: load team_members({team_id}) failed: {e}");
            return graph;
        }
    };
    let mut by_node: std::collections::HashMap<String, Uuid> = std::collections::HashMap::new();
    for row in members {
        let node_id: String = row.get("node_id");
        let claw_id: Uuid = row.get("claw_id");
        by_node.insert(node_id, claw_id);
    }
    if by_node.is_empty() {
        return graph;
    }
    apply_node_agents(graph, &by_node)
}

/// Pure core of [`inject_node_agents`] — the DB-free half, so the binding
/// contract can be regression-tested against the real `TopologyGraph`
/// deserializer.
fn apply_node_agents(
    graph: serde_json::Value,
    by_node: &std::collections::HashMap<String, Uuid>,
) -> serde_json::Value {
    let mut graph = graph;
    if let Some(nodes) = graph.get_mut("nodes").and_then(|v| v.as_array_mut()) {
        for node in nodes {
            let Some(obj) = node.as_object_mut() else {
                continue;
            };
            let id = obj.get("id").and_then(|v| v.as_str()).map(str::to_string);
            let Some(id) = id else { continue };
            if let Some(claw_id) = by_node.get(&id) {
                let alias = crate::runtime_provision::claw_alias(*claw_id);
                // The binding that actually takes effect (see doc comment).
                match obj.get_mut("attrs").and_then(|v| v.as_object_mut()) {
                    Some(attrs) => {
                        attrs.insert(
                            "agent".to_string(),
                            serde_json::Value::String(alias.clone()),
                        );
                    }
                    None => {
                        let mut attrs = serde_json::Map::new();
                        attrs.insert(
                            "agent".to_string(),
                            serde_json::Value::String(alias.clone()),
                        );
                        obj.insert("attrs".to_string(), serde_json::Value::Object(attrs));
                    }
                }
                // Kept for display/debug only — serde drops it on load.
                obj.insert("agent".to_string(), serde_json::Value::String(alias));
            }
        }
    }
    graph
}

#[cfg(test)]
mod tests {

    /// An unreachable judge must not spend the phase's iteration budget, and
    /// must not retry forever either. Both halves are the property: the first
    /// stops an outage failing work that was done, the second stops a dead
    /// validator leaving a phase `evaluating` in silence.
    #[test]
    fn the_judge_wait_is_bounded_and_shorter_than_the_capacity_wait() {
        assert!(
            JUDGE_WAIT_MAX_SECS > 0.0,
            "a zero wait would fail on the first transient error"
        );
        assert!(
            JUDGE_WAIT_MAX_SECS < CAPACITY_WAIT_MAX_SECS,
            "waiting on a judge is cheaper to abandon than waiting on a VM slot"
        );
        // Many sweep ticks, so a blip recovers well inside the window.
        assert!(
            JUDGE_WAIT_MAX_SECS >= 10.0 * 60.0,
            "too short and a normal provider blip fails the phase"
        );
    }

    /// The give-up path must FAIL, never requeue: re-running the phase spends a
    /// container and a model budget re-doing work that was never the problem.
    /// The real 429 z.ai returns for an exhausted plan. Retrying it is not
    /// optimism, it is arithmetic: on 2026-09-09 the reset was two days out and
    /// the phase spent its whole 30-minute window asking anyway.
    #[test]
    fn an_exhausted_plan_is_recognised_and_names_its_reset() {
        let why = r#"provider returned an error: 429 Too Many Requests: {"type":"error","error":{"type":"rate_limit_error","code":"1310","message":"[1310][Weekly/Monthly Limit Exhausted. Your limit will reset at 2026-09-11 10:01:33][2026090911555755ef730ec0404849]"}}"#;
        let plain = super::judge_error_is_exhausted_plan(why).expect("recognised");
        assert!(plain.contains("2026-09-11 10:01:33"), "{plain}");
        assert!(plain.contains("exhausted"), "{plain}");
    }

    /// Conservative by design. A transport blip must stay retryable — giving up
    /// on one costs a phase that had done nothing wrong, which is the failure
    /// mission 01a011bf actually suffered.
    #[test]
    fn a_transient_error_stays_retryable() {
        assert!(super::judge_error_is_exhausted_plan(
            "transport error: error decoding response body"
        )
        .is_none());
        assert!(super::judge_error_is_exhausted_plan("429 Too Many Requests").is_none());
        assert!(super::judge_error_is_exhausted_plan("").is_none());
    }

    /// Waiting as long as we have already waited doubles total elapsed per
    /// attempt, so the window holds ~10 attempts instead of 180.
    #[test]
    fn the_backoff_is_exponential_and_capped() {
        assert_eq!(super::judge_backoff_secs(0.0), 10.0, "first retry is one sweep");
        assert_eq!(super::judge_backoff_secs(20.0), 20.0);
        assert_eq!(super::judge_backoff_secs(160.0), 160.0);
        assert_eq!(
            super::judge_backoff_secs(1_000.0),
            super::JUDGE_RETRY_MAX_BACKOFF_SECS,
            "capped, or a long outage stops retrying at all"
        );

        // Count the attempts the 30-minute window now allows.
        let mut elapsed = 0.0f64;
        let mut attempts = 1;
        while elapsed < super::JUDGE_WAIT_MAX_SECS {
            elapsed += super::judge_backoff_secs(elapsed);
            attempts += 1;
        }
        assert!(
            (5..=15).contains(&attempts),
            "expected roughly ten attempts, got {attempts}"
        );
    }

    #[test]
    fn giving_up_on_the_judge_closes_the_phase() {
        let src = include_str!("phase_runner.rs");
        let block = src
            .split("let mut judge_gave_up = false;")
            .nth(1)
            .expect("the guard exists");
        let head = &block[..block.find("let last_pass").unwrap_or(block.len())];
        assert!(
            head.contains("judge_gave_up = true;"),
            "the timeout branch must set the flag"
        );
        assert!(
            block.contains("verdict.met || last_pass || judge_gave_up"),
            "the flag must reach the close decision, or it requeues instead"
        );
    }

    use super::*;

    fn by_node(pairs: &[(&str, Uuid)]) -> std::collections::HashMap<String, Uuid> {
        pairs.iter().map(|(k, v)| (k.to_string(), *v)).collect()
    }

    /// Composing the engines must be asked for by name. A mission that says
    /// nothing — or says something adjacent — runs the solo VM path it ran
    /// before, one VM for the whole phase.
    #[test]
    fn only_an_explicit_request_composes_the_engines() {
        assert!(wants_composed(Some("composed")));
        for engine in [
            None,
            Some(""),
            Some("claude_code"),
            Some("zeroclaw"),
            Some("Composed"),
            Some("compose"),
        ] {
            assert!(!wants_composed(engine), "{engine:?} must not compose");
        }
    }

    /// The composed tier the producer inserts and the tier the worker dispatches
    /// on are the same string, and nothing but this test connects them. A typo
    /// would enqueue a row no worker ever claims: the phase would sit `running`
    /// with a `queued` run behind it and no error anywhere.
    #[test]
    fn the_composed_tier_is_one_the_worker_actually_drives() {
        assert!(
            cm_db::repo::topology_runs::WORKER_DRIVEN_TIERS.contains(&"microvm_graph"),
            "the producer would enqueue a run nobody claims"
        );
    }

    /// The regression from mission `019fcf62`: a coding phase whose agents were
    /// silently unpinned from the repo wrote nothing and reported `completed`,
    /// the same status a fully delivered phase gets.
    #[test]
    fn a_coding_phase_that_delivers_nothing_is_a_failure() {
        let none = serde_json::json!({});
        assert!(
            empty_delivery_is_a_failure("coding", 0, None, &none),
            "zero files from a coding phase is not success"
        );
        assert!(
            !empty_delivery_is_a_failure("coding", 1, None, &none),
            "a phase that changed a file delivered"
        );
    }

    /// Every phase whose directive tells it to write files is held to it.
    ///
    /// This test used to assert the opposite, on the reasoning that "a research
    /// phase legitimately writes nothing to the tree". That was contradicted by
    /// the research directive in this very file, which tells the agent to save
    /// findings under `/mission/repo/research/` — and it cost `benchmark` and
    /// `security_hardening` missions any delivery guarantee at all, because
    /// their defining phases are not `coding`. A `benchmark` mission is ONE
    /// benchmark phase; with that phase exempt, nothing in the platform could
    /// fail it.
    #[test]
    fn every_producing_kind_is_held_to_delivering_files() {
        let none = serde_json::json!({});
        for kind in ["coding", "research", "benchmark", "security_scan"] {
            assert!(
                empty_delivery_is_a_failure(kind, 0, None, &none),
                "{kind} is told to write files, so producing none is a failure"
            );
            assert!(
                !empty_delivery_is_a_failure(kind, 1, None, &none),
                "{kind} that wrote a file delivered"
            );
        }
    }

    /// A reviewing phase is SUPPOSED to leave the tree alone.
    ///
    /// The same distinction `vm_stop_gate::per_node` makes when it drops
    /// `require_changes` for a composed graph's verifier node: holding a
    /// reviewer to changing files fails it for doing exactly its job.
    #[test]
    fn a_reviewing_phase_may_change_nothing() {
        let none = serde_json::json!({});
        for kind in ["review", "something_unrecognised"] {
            assert!(
                !empty_delivery_is_a_failure(kind, 0, None, &none),
                "{kind} is not promised to produce files"
            );
        }
    }

    /// The distinction this whole defect class turns on: an uncomputable diff
    /// ALSO reports zero files. Blaming the agent for a platform fault would
    /// re-encode the failure as an ordinary outcome — the exact mistake the
    /// `diff_error` field exists to prevent.
    #[test]
    fn an_uncomputable_diff_does_not_fail_the_phase() {
        let none = serde_json::json!({});
        assert!(
            !empty_delivery_is_a_failure("coding", 0, Some("patch: fatal: bad object"), &none),
            "a diff we could not compute is a platform fault, not an empty delivery"
        );
    }

    /// The escape hatch must be asserted for, not assumed: only an explicit
    /// `true` opts out, so a typo leaves the check armed.
    #[test]
    fn allow_empty_must_be_an_explicit_true() {
        assert!(!empty_delivery_is_a_failure(
            "coding",
            0,
            None,
            &serde_json::json!({"allow_empty": true})
        ));
        for wrong in [
            serde_json::json!({"allow_empty": "true"}),
            serde_json::json!({"allow_empty": 1}),
            serde_json::json!({"allow_empty": false}),
            serde_json::json!({"allowEmpty": true}),
        ] {
            assert!(
                empty_delivery_is_a_failure("coding", 0, None, &wrong),
                "{wrong} must not disable the check"
            );
        }
    }

    /// A phase's own task must reach the agent, and two phases of one mission
    /// must not receive identical text.
    ///
    /// This is the regression from mission `019fc42b`: `config.task` was
    /// accepted by the API, stored in the DB, and read by nothing. Both coding
    /// phases got byte-identical instructions and both produced the same two
    /// files. Asserting the texts *differ* is the part that matters — asserting
    /// only that the task appears would still pass if the brief carried it.
    #[test]
    fn phase_task_reaches_the_agent_and_distinguishes_phases() {
        let brief = Some("Add two marker files.");
        let alpha = phase_task_text("coding", "Demo", brief, Some("Create ALPHA.md"), true);
        let beta = phase_task_text("coding", "Demo", brief, Some("Create BETA.md"), true);

        assert!(
            alpha.contains("Create ALPHA.md"),
            "phase task must be injected"
        );
        assert!(beta.contains("Create BETA.md"));
        assert!(
            !alpha.contains("BETA.md"),
            "a phase must not see its sibling's task"
        );
        assert_ne!(
            alpha, beta,
            "sibling phases received identical instructions"
        );

        // A phase with no task of its own is unchanged from before the fix.
        let bare = phase_task_text("coding", "Demo", brief, None, true);
        assert!(!bare.contains("THIS PHASE'S TASK"));
        // Empty and whitespace-only configs take the same path as absent.
        assert_eq!(
            bare,
            phase_task_text("coding", "Demo", brief, Some("   "), true)
        );
    }

    /// The marker syntax we hand the agent must be the syntax we parse back.
    ///
    /// These two sides used to live far apart — the rules were in team-template
    /// role prompts that mission turns never receive — so nothing caught a
    /// drift between what we asked for and what `task_card_parser` accepts.
    /// Every example line in the prompt is fed through the real parser here.
    #[test]
    fn task_text_marker_examples_parse() {
        let text = phase_task_text("coding", "Demo", Some("brief"), None, true);

        let examples: Vec<&str> = text
            .lines()
            .map(str::trim)
            .filter(|l| l.starts_with("- ") && l.contains("INT-NN"))
            .map(|l| l.trim_start_matches("- "))
            .collect();
        assert!(
            examples.len() >= 8,
            "expected the full marker ladder in the prompt, found {}: {examples:?}",
            examples.len()
        );

        for ex in examples {
            // Strip the trailing prose column ("open a new item") and the
            // <placeholder>, leaving a marker line an agent would actually emit.
            let line = ex.replace("INT-NN", "INT-05");
            let line = line.split("  ").next().unwrap_or(&line).trim();
            let line = line
                .replace("<title>", "Add retry")
                .replace("<reason>", "compile error");
            let parsed = crate::task_card_parser::parse(&line);
            assert_eq!(
                parsed.len(),
                1,
                "prompt advertises a marker the parser does not accept: {line:?}"
            );
            assert_eq!(parsed[0].int_id, "INT-05", "wrong id parsed from {line:?}");
        }
    }

    /// The regression guard: the alias must survive a round-trip through the
    /// real `TopologyGraph` deserializer and land in `attrs`. A top-level
    /// `"agent"` key alone is dropped by serde, which silently routed every
    /// mission turn to the default `scout` agent.
    #[test]
    fn bound_alias_survives_topology_graph_deserialization() {
        let claw = Uuid::nil();
        let graph = serde_json::json!({
            "kind": "pipeline",
            "nodes": [{"id": "n0", "role": "coder"}],
            "edges": [],
        });

        let out = apply_node_agents(graph, &by_node(&[("n0", claw)]));
        let parsed: cm_topology::TopologyGraph =
            serde_json::from_value(out).expect("graph deserializes");

        assert_eq!(
            parsed.nodes[0].attrs.get("agent").map(String::as_str),
            Some(crate::runtime_provision::claw_alias(claw).as_str()),
            "alias must be readable from attrs after a real deserialize"
        );
    }

    #[test]
    fn binding_preserves_existing_attrs() {
        let claw = Uuid::nil();
        let graph = serde_json::json!({
            "kind": "pipeline",
            "nodes": [{"id": "n0", "role": "coder", "attrs": {"budget": "5"}}],
            "edges": [],
        });

        let out = apply_node_agents(graph, &by_node(&[("n0", claw)]));
        let attrs = &out["nodes"][0]["attrs"];
        assert_eq!(attrs["budget"], "5");
        assert_eq!(attrs["agent"], crate::runtime_provision::claw_alias(claw));
    }

    #[test]
    fn unmapped_nodes_are_left_unbound() {
        let graph = serde_json::json!({
            "kind": "pipeline",
            "nodes": [{"id": "n0", "role": "coder"}, {"id": "n1", "role": "tester"}],
            "edges": [],
        });

        let out = apply_node_agents(graph, &by_node(&[("n0", Uuid::nil())]));
        assert!(out["nodes"][0]["attrs"]["agent"].is_string());
        // n1 has no member row — the executor falls back to the alias map.
        assert!(out["nodes"][1].get("attrs").is_none());
    }
}