feat(missions): goal conditions and phase iteration, judged on the subscription model

A phase used to complete when its topology_runs reached a terminal state --
purely structural. It marked itself done whether the agents produced the
artifact or wrote nothing at all, and it ran exactly once: execute_resumable's
skip(start) is resume, not repeat, and the only re-run path was a human
hitting the retry endpoint.

A phase can now carry `done_when`, a completion condition judged after each
pass against the evidence the agents actually surfaced. Not met and passes
remain -> the phase goes back to pending with iteration bumped, and the
verdict's reason is appended to the next pass's task text. That feedback is
what makes iteration converge rather than repeat -- the same mechanism /goal
uses, and that swarm.rs already uses for rejected work.

The evaluator runs on the SUBSCRIPTION model. CLAWMATES_EVALUATOR_MODEL
defaults to judge_model(), and a `runtime:<alias>` spec routes through
ZeroClawDriveExecutor -- a container agent on claude_cli, i.e. Claude Code on
the OAuth subscription, needing no platform API key. Same routing the door
governor uses.

Two deliberate departures from the governor's contract, both required:

- FAIL-CLOSED. Runtime::judge is fail-open and reads a verdict by
  !contains("DENY"), so a model explaining why it *would* deny reads as
  approval and an empty reply reads as approval. For completion that is
  backwards: unsure must mean not done. The contract is swarm.rs's strict
  JSON {"met","reason"} with .unwrap_or(false). Six tests cover the closed
  paths -- prose, empty, missing field, non-boolean, transport error.
- judge_raw returns the raw reply; judge collapses to a bool too early to
  carry a structured verdict.

Iteration scoping is the subtle part and has its own test: on pass 2 the
phase's own iteration is 1 but pass 1's completed run is still in the table,
so "are this phase's runs all finished?" must ask about the CURRENT pass or
that stale row closes out pass 2 the instant it is enqueued.

Evidence comes from phase_summarizer::collect_evidence, extracted from the
existing collect_material so the evaluator and the summary card cannot
disagree about what a phase produced.

done_when/max_iterations are promoted from phase config into columns (the
sweep filters on them every tick) and max_iterations is clamped to 20 at
insert -- the UI limits it too, but a runaway loop must not be one crafted
request away.

A phase with no condition completes exactly as before; that regression guard
is the first test in the file.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>

crates/bins/clawmates-server/src/main.rs

@@ -295,7 +295,7 @@ async fn run() -> Result<(), String> {
         let recipes = cm_api::workflow_registry::load();
         eprintln!("workflow_registry: {} recipe(s) available", recipes.len());
     }
-    cm_api::phase_runner::spawn(pool.clone());
+    cm_api::phase_runner::spawn(pool.clone(), runtime.clone());
     // Per-mission runtime container sweeper (C3): tears down mission
     // runtime containers 30 min after the mission reaches a terminal
     // state so operators have a window to pull final artifacts.

crates/cm-api/src/evaluator.rs

@@ -0,0 +1,281 @@
+//! Phase completion evaluation — the `/goal` analogue.
+//!
+//! A mission phase can carry a `done_when` condition. After every pass, this
+//! module asks a model whether the condition holds against what the agents
+//! actually surfaced, and returns a verdict plus a reason. The reason is used
+//! twice: shown to the operator, and fed into the next pass as guidance —
+//! which is what makes iteration converge rather than merely repeat.
+//!
+//! ## Two properties that are not negotiable
+//!
+//! **Fail-closed.** An unparseable reply, an empty reply, or a transport
+//! error means *not done*. The door governor ([`Runtime::judge`]) is
+//! deliberately fail-open — a governor outage must not halt agents — but the
+//! opposite is right here: a judge outage must not declare work finished. The
+//! verdict contract is `swarm.rs`'s (`{"passed":..}` → `.unwrap_or(false)`),
+//! not the governor's `!contains("DENY")`, which reads a model that explains
+//! *why it would deny* as a denial and an empty string as approval.
+//!
+//! **The judge cannot run commands.** It sees only the transcript material we
+//! hand it. Conditions must therefore be demonstrable from turn output —
+//! "`cargo test` passes and the output shows 0 failures" works because the
+//! agent runs the tests and the result lands in the transcript; "the code is
+//! well factored" does not. This constraint is surfaced in the mission wizard
+//! and in the planner prompt.
+
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use uuid::Uuid;
+
+/// The model's verdict on one pass.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Verdict {
+    pub met: bool,
+    pub reason: String,
+    /// The model spec that judged, recorded for attribution.
+    pub model: String,
+    /// Set when the evaluator itself failed rather than judging "not met" —
+    /// distinguishes "judged incomplete" from "could not judge".
+    pub error: Option<String>,
+}
+
+impl Verdict {
+    fn not_met(model: &str, reason: impl Into<String>, error: Option<String>) -> Self {
+        Verdict {
+            met: false,
+            reason: reason.into(),
+            model: model.to_string(),
+            error,
+        }
+    }
+}
+
+const EVAL_SYSTEM: &str = "\
+You judge whether a phase of automated work is complete.
+
+You are given the phase's COMPLETION CONDITION and the EVIDENCE its agents \
+produced — their turn output, task states, and artifacts. Decide whether the \
+condition holds.
+
+You cannot run commands or read files. Judge only what the evidence shows. If \
+the evidence does not positively demonstrate the condition, it is not met — \
+absence of evidence is not satisfaction. Do not assume work happened because \
+an agent said it would.
+
+Respond with STRICT JSON ONLY, no prose and no code fence:
+{\"met\": true|false, \"reason\": \"one or two sentences\"}
+
+When met is false, the reason is handed to the agents as guidance for their \
+next attempt, so state specifically what is still missing.";
+
+/// The model spec to judge with.
+///
+/// Defaults to [`cm_runtime::judge_model`] so a single knob configures both
+/// the door governor and this. A `runtime:<alias>` spec drives a ZeroClaw
+/// container agent — which on this deployment is `claude_cli`, i.e. Claude
+/// Code on the OAuth **subscription**, needing no platform API key. Anything
+/// else resolves through the provider registry.
+pub fn evaluator_model() -> String {
+    std::env::var("CLAWMATES_EVALUATOR_MODEL").unwrap_or_else(|_| cm_runtime::judge_model())
+}
+
+/// Judge whether `condition` holds given `evidence`.
+///
+/// Never returns `Err`: a failure to judge is a `Verdict` with `met: false`
+/// and `error` set, so the caller records the attempt and keeps iterating
+/// rather than silently completing the phase.
+pub async fn evaluate(
+    runtime: &cm_runtime::Runtime,
+    condition: &str,
+    evidence: &str,
+) -> Verdict {
+    let model = evaluator_model();
+    let user = format!("COMPLETION CONDITION:\n{condition}\n\nEVIDENCE:\n{evidence}");
+
+    // Same routing as the door governor (mcp_door.rs): `runtime:<alias>` goes
+    // through the container agent so a subscription-only model can judge.
+    let raw: Result<String, String> = if let Some(alias) = model.strip_prefix("runtime:") {
+        match crate::topology_exec::ZeroClawDriveExecutor::from_env() {
+            Ok(exec) => exec.judge_raw(alias.trim(), EVAL_SYSTEM, &user).await,
+            Err(e) => Err(format!("runtime executor unavailable: {e}")),
+        }
+    } else {
+        runtime
+            .complete(EVAL_SYSTEM, &user, &model, 512, false)
+            .await
+    };
+
+    match raw {
+        Err(e) => Verdict::not_met(
+            &model,
+            "could not evaluate the completion condition this pass",
+            Some(e),
+        ),
+        Ok(text) => parse_verdict(&model, &text),
+    }
+}
+
+/// Parse the model's reply into a verdict, failing closed.
+fn parse_verdict(model: &str, text: &str) -> Verdict {
+    let trimmed = text.trim();
+    if trimmed.is_empty() {
+        return Verdict::not_met(
+            model,
+            "evaluator returned an empty reply",
+            Some("empty reply".into()),
+        );
+    }
+    let Some(v): Option<Value> = crate::routes::claws::extract_json(trimmed) else {
+        return Verdict::not_met(
+            model,
+            "evaluator reply was not valid JSON",
+            Some(format!("unparseable reply: {}", head(trimmed, 200))),
+        );
+    };
+    // `.unwrap_or(false)` is the fail-closed hinge: a reply missing `met`, or
+    // with a non-boolean `met`, is treated as not done.
+    let met = v.get("met").and_then(|m| m.as_bool()).unwrap_or(false);
+    let reason = v
+        .get("reason")
+        .and_then(|r| r.as_str())
+        .map(str::trim)
+        .filter(|r| !r.is_empty())
+        .unwrap_or(if met {
+            "condition met"
+        } else {
+            "evaluator gave no reason"
+        })
+        .to_string();
+    Verdict {
+        met,
+        reason,
+        model: model.to_string(),
+        error: None,
+    }
+}
+
+fn head(s: &str, n: usize) -> String {
+    // Truncate on a char boundary so multi-byte output can't panic here.
+    match s.char_indices().nth(n) {
+        Some((i, _)) => format!("{}…", &s[..i]),
+        None => s.to_string(),
+    }
+}
+
+/// Persist one verdict. Best-effort at the call site; a lost evaluation row
+/// costs an operator the audit trail, not correctness.
+pub async fn record(
+    pool: &sqlx::PgPool,
+    mission_id: Uuid,
+    phase_id: Uuid,
+    iteration: i32,
+    v: &Verdict,
+) -> Result<(), sqlx::Error> {
+    sqlx::query(
+        "INSERT INTO mission_phase_evaluations
+             (id, mission_id, phase_id, iteration, met, reason, model, error)
+         VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+         ON CONFLICT (phase_id, iteration) DO UPDATE
+            SET met = EXCLUDED.met, reason = EXCLUDED.reason,
+                model = EXCLUDED.model, error = EXCLUDED.error",
+    )
+    .bind(Uuid::now_v7())
+    .bind(mission_id)
+    .bind(phase_id)
+    .bind(iteration)
+    .bind(v.met)
+    .bind(&v.reason)
+    .bind(&v.model)
+    .bind(v.error.as_deref())
+    .execute(pool)
+    .await
+    .map(|_| ())
+}
+
+/// The most recent verdict for a phase, used to carry guidance into the next
+/// pass and to render the operator-facing strip.
+pub async fn latest(
+    pool: &sqlx::PgPool,
+    phase_id: Uuid,
+) -> Result<Option<(i32, bool, String)>, sqlx::Error> {
+    use sqlx::Row;
+    let row = sqlx::query(
+        "SELECT iteration, met, reason FROM mission_phase_evaluations
+          WHERE phase_id = $1 ORDER BY iteration DESC LIMIT 1",
+    )
+    .bind(phase_id)
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.map(|r| {
+        (
+            r.get::<i32, _>("iteration"),
+            r.get::<bool, _>("met"),
+            r.get::<String, _>("reason"),
+        )
+    }))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_a_well_formed_verdict() {
+        let v = parse_verdict("m", r#"{"met": true, "reason": "tests pass"}"#);
+        assert!(v.met);
+        assert_eq!(v.reason, "tests pass");
+        assert!(v.error.is_none());
+    }
+
+    #[test]
+    fn tolerates_a_code_fence() {
+        let v = parse_verdict("m", "```json\n{\"met\": false, \"reason\": \"no brief\"}\n```");
+        assert!(!v.met);
+        assert_eq!(v.reason, "no brief");
+    }
+
+    // ── The fail-closed contract. Each of these once meant "allow" under the
+    //    governor's !contains("DENY") parse; here they must all mean NOT done.
+
+    #[test]
+    fn unparseable_reply_is_not_met() {
+        let v = parse_verdict("m", "I think the phase is basically finished, yes.");
+        assert!(!v.met, "prose must not be read as completion");
+        assert!(v.error.is_some(), "should record why it could not judge");
+    }
+
+    #[test]
+    fn empty_reply_is_not_met() {
+        let v = parse_verdict("m", "   ");
+        assert!(!v.met);
+        assert!(v.error.is_some());
+    }
+
+    #[test]
+    fn missing_met_field_is_not_met() {
+        let v = parse_verdict("m", r#"{"reason": "looks good to me"}"#);
+        assert!(!v.met, "a verdict with no `met` must not complete the phase");
+    }
+
+    #[test]
+    fn non_boolean_met_is_not_met() {
+        let v = parse_verdict("m", r#"{"met": "yes", "reason": "done"}"#);
+        assert!(!v.met, "a stringly-typed `met` must not complete the phase");
+    }
+
+    #[test]
+    fn a_verdict_always_carries_a_reason() {
+        assert!(!parse_verdict("m", r#"{"met": false}"#).reason.is_empty());
+        assert!(!parse_verdict("m", r#"{"met": true}"#).reason.is_empty());
+        assert!(!parse_verdict("m", r#"{"met": false, "reason": "  "}"#)
+            .reason
+            .is_empty());
+    }
+
+    #[test]
+    fn head_truncates_on_a_char_boundary() {
+        let s = "é".repeat(300);
+        let _ = head(&s, 200); // must not panic
+        assert!(head("abc", 200).ends_with('c'));
+    }
+}

crates/cm-api/src/lib.rs

@@ -5,6 +5,7 @@ pub mod beszel;
 pub mod brain_seed;
 pub mod cleanup_sweeper;
 mod error;
+pub mod evaluator;
 mod extract;
 pub mod fleet;
 pub mod fleet_herdr;

crates/cm-api/src/phase_runner.rs

@@ -33,23 +33,28 @@ use uuid::Uuid;
 
 const POLL_INTERVAL: Duration = Duration::from_secs(10);
 
-pub fn spawn(pool: PgPool) {
+/// `runtime` is needed only by the completion evaluator; phases without a
+/// `done_when` never touch it.
+pub fn spawn(pool: PgPool, runtime: cm_runtime::Runtime) {
     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).await {
+            if let Err(e) = sweep_once(&pool, &runtime).await {
                 eprintln!("phase_runner: sweep failed: {e}");
             }
         }
     });
 }
 
-async fn sweep_once(pool: &PgPool) -> Result<(), String> {
+async fn sweep_once(pool: &PgPool, runtime: &cm_runtime::Runtime) -> Result<(), String> {
     start_pending_phases(pool).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?;
     close_finished_missions(pool).await?;
     Ok(())
 }
@@ -60,7 +65,7 @@ async fn start_pending_phases(pool: &PgPool) -> Result<(), String> {
     // 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,
+        "SELECT mp.id, mp.mission_id, mp.kind, mp.order_idx, mp.iteration,
                 m.workspace_id, m.title, m.description
          FROM mission_phases mp
          JOIN missions m ON m.id = mp.mission_id
@@ -85,15 +90,19 @@ async fn start_pending_phases(pool: &PgPool) -> Result<(), String> {
         let workspace_id: Uuid = row.get("workspace_id");
         let title: String = row.get("title");
         let description: Option<String> = row.get("description");
+        let iteration: i32 = row.get("iteration");
 
         if let Err(e) = launch_phase(
             pool,
+            PhaseLaunch {
                 phase_id,
                 mission_id,
-            &kind,
+                kind: &kind,
                 workspace_id,
-            &title,
+                title: &title,
-            description.as_deref(),
+                description: description.as_deref(),
+                iteration,
+            },
         )
         .await
         {
@@ -103,15 +112,30 @@ async fn start_pending_phases(pool: &PgPool) -> Result<(), String> {
     Ok(())
 }
 
-async fn launch_phase(
+/// Everything `launch_phase` needs about the phase it is starting, gathered
-    pool: &PgPool,
+/// from the eligibility query.
+struct PhaseLaunch<'a> {
     phase_id: Uuid,
     mission_id: Uuid,
-    kind: &str,
+    kind: &'a str,
     workspace_id: Uuid,
-    title: &str,
+    title: &'a str,
-    description: Option<&str>,
+    description: Option<&'a str>,
-) -> Result<(), String> {
+    /// 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,
+}
+
+async fn launch_phase(pool: &PgPool, p: PhaseLaunch<'_>) -> Result<(), String> {
+    let PhaseLaunch {
+        phase_id,
+        mission_id,
+        kind,
+        workspace_id,
+        title,
+        description,
+        iteration,
+    } = p;
     // Which team purposes should execute this phase.
     let purposes: &[&str] = match kind {
         "research" => &["research", "mission"],
@@ -202,7 +226,20 @@ async fn launch_phase(
         }
     }
 
+    // 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);
+    let task = match prior {
+        Some((iter, false, reason)) => format!(
+            "{task}\n\nPREVIOUS ATTEMPT (pass {}) DID NOT SATISFY THE COMPLETION \
+             CONDITION:\n{reason}\n\nAddress this specifically in this pass.",
+            iter + 1
+        ),
+        _ => task,
+    };
 
     // Purge prior failed / cancelled runs for this phase so the card
     // starts fresh on re-attempts. Completed runs are kept for
@@ -231,8 +268,8 @@ async fn launch_phase(
         sqlx::query(
             "INSERT INTO topology_runs
                 (id, workspace_id, task, kind, status, graph, tier,
-                 team_id, mission_id, mission_phase_id)
+                 team_id, mission_id, mission_phase_id, iteration)
-             VALUES ($1, $2, $3, 'run', 'queued', $4, 'team', $5, $6, $7)",
+             VALUES ($1, $2, $3, 'run', 'queued', $4, 'team', $5, $6, $7, $8)",
         )
         .bind(run_id)
         .bind(workspace_id)
@@ -241,6 +278,9 @@ async fn launch_phase(
         .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}"))?;
@@ -349,6 +389,12 @@ fn phase_task_text(kind: &str, title: &str, description: Option<&str>) -> String
 }
 
 /// 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> {
     sqlx::query(
         "UPDATE mission_phases mp
@@ -356,16 +402,33 @@ async fn close_finished_phases(pool: &PgPool) -> Result<(), String> {
               CASE
                 WHEN EXISTS (
                   SELECT 1 FROM topology_runs r
-                  WHERE r.mission_phase_id = mp.id AND r.status = 'failed'
+                  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 = now()
+              completed_at =
-          WHERE mp.status = 'running'
+                CASE
-            AND EXISTS (SELECT 1 FROM topology_runs r WHERE r.mission_phase_id = mp.id)
+                  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')
             )",
     )
@@ -375,6 +438,82 @@ async fn close_finished_phases(pool: &PgPool) -> Result<(), String> {
     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
+           FROM mission_phases mp
+           JOIN missions m ON m.id = mp.mission_id
+          WHERE mp.status = 'evaluating' AND m.status = 'running'
+          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 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, &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}");
+        }
+
+        let last_pass = iteration + 1 >= max_iterations;
+        if verdict.met || last_pass {
+            sqlx::query(
+                "UPDATE mission_phases SET status = 'completed', completed_at = now()
+                  WHERE id = $1 AND status = 'evaluating'",
+            )
+            .bind(phase_id)
+            .execute(pool)
+            .await
+            .map_err(|e| format!("complete phase {phase_id}: {e}"))?;
+            eprintln!(
+                "phase_runner: phase {phase_id} ({kind}) completed after {} pass(es) — met={} — {}",
+                iteration + 1,
+                verdict.met,
+                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.
 async fn close_finished_missions(pool: &PgPool) -> Result<(), String> {
     sqlx::query(

crates/cm-api/src/phase_summarizer.rs

@@ -170,6 +170,46 @@ struct TaskRef {
     status: String,
 }
 
+/// Render this phase's material as plain evidence text.
+///
+/// Shared with the completion evaluator (`crate::evaluator`), which judges a
+/// `done_when` condition against exactly the same material the summarizer
+/// writes its card from — turn outputs, task counts, artifacts. Reusing this
+/// keeps the two from disagreeing about what the phase actually produced, and
+/// the truncation/aggregation logic only has to be right once.
+pub async fn collect_evidence(
+    pool: &PgPool,
+    mission_id: Uuid,
+    phase_id: Uuid,
+) -> Result<String, String> {
+    let m = collect_material(pool, mission_id, phase_id).await?;
+    let mut s = String::with_capacity(m.outputs.len() + 512);
+    s.push_str(&format!(
+        "turns: {}\ntokens: {}\nagent outputs: {}\ntasks: {} created, {} completed, {} failed\n",
+        m.turns, m.tokens, m.output_count, m.tasks_created, m.tasks_completed, m.tasks_failed,
+    ));
+    if !m.artifacts.is_empty() {
+        s.push_str("\nartifacts written:\n");
+        for a in m.artifacts.iter().take(40) {
+            s.push_str(&format!("- {} ({})\n", a.path, a.kind));
+        }
+    }
+    if !m.task_summaries.is_empty() {
+        s.push_str("\ntask states:\n");
+        for t in m.task_summaries.iter().take(40) {
+            s.push_str(&format!(
+                "- {} [{}] {}\n",
+                t.external_id.as_deref().unwrap_or("-"),
+                t.status,
+                t.title
+            ));
+        }
+    }
+    s.push_str("\nagent turn output:\n");
+    s.push_str(&m.outputs);
+    Ok(s)
+}
+
 async fn collect_material(
     pool: &PgPool,
     mission_id: Uuid,

crates/cm-api/src/topology_exec.rs

@@ -243,6 +243,22 @@ impl ZeroClawDriveExecutor {
         }
     }
 
+    /// Drive `alias` with a judging prompt and return its **raw** reply.
+    ///
+    /// [`Self::judge`] collapses the reply to a bool by substring-matching
+    /// `DENY`, which only suits the governor's ALLOW/DENY contract and is
+    /// fail-open. Callers that need a structured verdict — the phase
+    /// completion evaluator wants `{"met":bool,"reason":string}` and must fail
+    /// **closed** — need the text, and need the error rather than a
+    /// synthesized permissive answer.
+    pub async fn judge_raw(&self, alias: &str, system: &str, user: &str) -> Result<String, String> {
+        let prompt = format!("{system}\n\n{user}");
+        self.drive(alias, &prompt)
+            .await
+            .map(|outcome| outcome.output.trim().to_string())
+            .map_err(|e| e.to_string())
+    }
+
     /// Drive agent `alias` as a delegated sub-task and return its result. Reuses
     /// the same gateway drive as topology turns + the governor, so a delegated
     /// turn carries the same blocked-action / token instrumentation in its

crates/cm-api/tests/phase_conditions.rs

@@ -0,0 +1,333 @@
+//! Coverage for goal conditions and phase iteration (migration 0061).
+//!
+//! These tests exercise the SQL directly rather than the sweep loop, because
+//! the part that is easy to get wrong is the *iteration scoping*: "are this
+//! phase's runs all finished?" must ask about the CURRENT pass. Without that,
+//! pass 1's completed rows satisfy pass 2 the instant it is enqueued and the
+//! phase completes without doing any work.
+//!
+//! What this locks in:
+//!   * A phase with no `done_when` still goes running -> completed on terminal
+//!     runs (the regression guard: existing missions are unaffected).
+//!   * A phase with `done_when` goes running -> evaluating instead.
+//!   * A failed run fails the phase outright, condition or not.
+//!   * Pass 2 is not satisfied by pass 1's completed runs.
+//!   * `mission_phase_evaluations` is unique per (phase, iteration) and
+//!     upserts.
+
+use cm_db::repo::workspaces;
+use cm_domain::{Workspace, WorkspaceId};
+use sqlx::Row;
+use uuid::Uuid;
+
+async fn seed_workspace(pool: &sqlx::PgPool) -> WorkspaceId {
+    let ws = Workspace {
+        id: WorkspaceId::new(),
+        name: "Phase Conditions Test".into(),
+        plan: "team".into(),
+    };
+    workspaces::insert(pool, &ws).await.unwrap();
+    ws.id
+}
+
+async fn seed_mission(pool: &sqlx::PgPool, ws: WorkspaceId) -> Uuid {
+    let id = Uuid::now_v7();
+    sqlx::query(
+        "INSERT INTO missions (id, workspace_id, title, template_kind, status)
+         VALUES ($1, $2, 'test mission', 'research_only', 'running')",
+    )
+    .bind(id)
+    .bind(ws.as_uuid())
+    .execute(pool)
+    .await
+    .unwrap();
+    id
+}
+
+/// A phase in `running`, optionally carrying a completion condition.
+async fn seed_phase(
+    pool: &sqlx::PgPool,
+    mission_id: Uuid,
+    done_when: Option<&str>,
+    max_iterations: i32,
+    iteration: i32,
+) -> Uuid {
+    let id = Uuid::now_v7();
+    sqlx::query(
+        "INSERT INTO mission_phases
+             (id, mission_id, kind, order_idx, status, done_when, max_iterations, iteration)
+         VALUES ($1, $2, 'research', 0, 'running', $3, $4, $5)",
+    )
+    .bind(id)
+    .bind(mission_id)
+    .bind(done_when)
+    .bind(max_iterations)
+    .bind(iteration)
+    .execute(pool)
+    .await
+    .unwrap();
+    id
+}
+
+async fn seed_run(
+    pool: &sqlx::PgPool,
+    ws: WorkspaceId,
+    mission_id: Uuid,
+    phase_id: Uuid,
+    status: &str,
+    iteration: i32,
+) {
+    sqlx::query(
+        "INSERT INTO topology_runs
+             (id, workspace_id, task, kind, status, tier, mission_id, mission_phase_id, iteration)
+         VALUES ($1, $2, 'task', 'run', $3, 'team', $4, $5, $6)",
+    )
+    .bind(Uuid::now_v7())
+    .bind(ws.as_uuid())
+    .bind(status)
+    .bind(mission_id)
+    .bind(phase_id)
+    .bind(iteration)
+    .execute(pool)
+    .await
+    .unwrap();
+}
+
+/// The exact statement `phase_runner::close_finished_phases` runs.
+async fn close_finished_phases(pool: &sqlx::PgPool) {
+    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')
+            )",
+    )
+    .execute(pool)
+    .await
+    .unwrap();
+}
+
+async fn phase_status(pool: &sqlx::PgPool, phase_id: Uuid) -> String {
+    sqlx::query("SELECT status FROM mission_phases WHERE id = $1")
+        .bind(phase_id)
+        .fetch_one(pool)
+        .await
+        .unwrap()
+        .get::<String, _>("status")
+}
+
+/// The regression guard. A mission that never opts into a condition must
+/// behave exactly as it did before conditions existed.
+#[tokio::test]
+async fn phase_without_condition_completes_as_before() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = seed_workspace(&pool).await;
+    let mission = seed_mission(&pool, ws).await;
+    let phase = seed_phase(&pool, mission, None, 1, 0).await;
+    seed_run(&pool, ws, mission, phase, "completed", 0).await;
+
+    close_finished_phases(&pool).await;
+    assert_eq!(phase_status(&pool, phase).await, "completed");
+}
+
+#[tokio::test]
+async fn phase_with_condition_goes_to_evaluating() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = seed_workspace(&pool).await;
+    let mission = seed_mission(&pool, ws).await;
+    let phase = seed_phase(&pool, mission, Some("a brief exists"), 3, 0).await;
+    seed_run(&pool, ws, mission, phase, "completed", 0).await;
+
+    close_finished_phases(&pool).await;
+    assert_eq!(
+        phase_status(&pool, phase).await,
+        "evaluating",
+        "a phase with a condition must be judged before it can complete"
+    );
+    // completed_at must stay NULL while the phase is still being judged.
+    let completed_at: Option<time::OffsetDateTime> =
+        sqlx::query("SELECT completed_at FROM mission_phases WHERE id = $1")
+            .bind(phase)
+            .fetch_one(&pool)
+            .await
+            .unwrap()
+            .get("completed_at");
+    assert!(completed_at.is_none(), "not finished, so not timestamped");
+}
+
+/// A blank condition is not a condition — otherwise a UI that sends "" would
+/// silently park every phase in `evaluating` forever.
+#[tokio::test]
+async fn blank_condition_is_treated_as_none() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = seed_workspace(&pool).await;
+    let mission = seed_mission(&pool, ws).await;
+    let phase = seed_phase(&pool, mission, Some("   "), 3, 0).await;
+    seed_run(&pool, ws, mission, phase, "completed", 0).await;
+
+    close_finished_phases(&pool).await;
+    assert_eq!(phase_status(&pool, phase).await, "completed");
+}
+
+#[tokio::test]
+async fn failed_run_fails_the_phase_even_with_a_condition() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = seed_workspace(&pool).await;
+    let mission = seed_mission(&pool, ws).await;
+    let phase = seed_phase(&pool, mission, Some("a brief exists"), 3, 0).await;
+    seed_run(&pool, ws, mission, phase, "failed", 0).await;
+
+    close_finished_phases(&pool).await;
+    assert_eq!(
+        phase_status(&pool, phase).await,
+        "failed",
+        "there is nothing to evaluate when the work itself failed"
+    );
+}
+
+/// The subtle one. On pass 2 the phase has `iteration = 1`, but pass 1's
+/// completed run is still in the table. Without scoping the check to the
+/// current iteration, that stale row satisfies "all runs finished" and the
+/// phase completes having done no work on this pass.
+#[tokio::test]
+async fn second_pass_is_not_satisfied_by_first_pass_runs() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = seed_workspace(&pool).await;
+    let mission = seed_mission(&pool, ws).await;
+    // Phase is on pass 2 (iteration=1) and running.
+    let phase = seed_phase(&pool, mission, Some("a brief exists"), 3, 1).await;
+    // Pass 1 left a completed run behind.
+    seed_run(&pool, ws, mission, phase, "completed", 0).await;
+    // Pass 2's run is still queued.
+    seed_run(&pool, ws, mission, phase, "queued", 1).await;
+
+    close_finished_phases(&pool).await;
+    assert_eq!(
+        phase_status(&pool, phase).await,
+        "running",
+        "pass 1's completed run must not close out pass 2"
+    );
+
+    // Finish pass 2 for real.
+    sqlx::query(
+        "UPDATE topology_runs SET status = 'completed'
+          WHERE mission_phase_id = $1 AND iteration = 1",
+    )
+    .bind(phase)
+    .execute(&pool)
+    .await
+    .unwrap();
+    close_finished_phases(&pool).await;
+    assert_eq!(phase_status(&pool, phase).await, "evaluating");
+}
+
+/// A phase whose current pass has enqueued nothing yet must not be closed by
+/// an earlier pass's rows either.
+#[tokio::test]
+async fn phase_with_no_runs_this_pass_stays_running() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = seed_workspace(&pool).await;
+    let mission = seed_mission(&pool, ws).await;
+    let phase = seed_phase(&pool, mission, None, 3, 1).await;
+    seed_run(&pool, ws, mission, phase, "completed", 0).await;
+
+    close_finished_phases(&pool).await;
+    assert_eq!(phase_status(&pool, phase).await, "running");
+}
+
+#[tokio::test]
+async fn evaluations_are_unique_per_iteration_and_upsert() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = seed_workspace(&pool).await;
+    let mission = seed_mission(&pool, ws).await;
+    let phase = seed_phase(&pool, mission, Some("done"), 3, 0).await;
+
+    let first = cm_api::evaluator::Verdict {
+        met: false,
+        reason: "no brief yet".into(),
+        model: "runtime:coordinator".into(),
+        error: None,
+    };
+    cm_api::evaluator::record(&pool, mission, phase, 0, &first)
+        .await
+        .unwrap();
+    // Same iteration again — upsert, not a duplicate row or a constraint error.
+    let second = cm_api::evaluator::Verdict {
+        met: true,
+        reason: "brief written".into(),
+        model: "runtime:coordinator".into(),
+        error: None,
+    };
+    cm_api::evaluator::record(&pool, mission, phase, 0, &second)
+        .await
+        .unwrap();
+
+    let count: i64 = sqlx::query("SELECT count(*) AS n FROM mission_phase_evaluations WHERE phase_id = $1")
+        .bind(phase)
+        .fetch_one(&pool)
+        .await
+        .unwrap()
+        .get("n");
+    assert_eq!(count, 1, "one row per (phase, iteration)");
+
+    let latest = cm_api::evaluator::latest(&pool, phase).await.unwrap();
+    assert_eq!(latest, Some((0, true, "brief written".into())));
+}
+
+/// `latest` must return the newest pass, which is what feeds guidance into the
+/// next attempt.
+#[tokio::test]
+async fn latest_returns_the_most_recent_iteration() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = seed_workspace(&pool).await;
+    let mission = seed_mission(&pool, ws).await;
+    let phase = seed_phase(&pool, mission, Some("done"), 3, 0).await;
+
+    for (i, reason) in [(0, "first"), (1, "second"), (2, "third")] {
+        cm_api::evaluator::record(
+            &pool,
+            mission,
+            phase,
+            i,
+            &cm_api::evaluator::Verdict {
+                met: false,
+                reason: reason.into(),
+                model: "m".into(),
+                error: None,
+            },
+        )
+        .await
+        .unwrap();
+    }
+    let latest = cm_api::evaluator::latest(&pool, phase).await.unwrap();
+    assert_eq!(latest, Some((2, false, "third".into())));
+}

crates/cm-db/src/repo/missions.rs

@@ -132,6 +132,12 @@ pub struct NewMissionPhase {
     pub config: Value,
 }
 
+/// Hard ceiling on phase passes, applied at insert regardless of what the
+/// caller asked for. Each pass is a full team run against a live model, so an
+/// unbounded loop is an unbounded bill; the evaluator deciding "not yet"
+/// forever must still terminate.
+pub const MAX_PHASE_ITERATIONS: i64 = 20;
+
 // ── Missions ─────────────────────────────────────────────────────
 
 /// Insert a mission + its phases in a single transaction.
@@ -164,16 +170,38 @@ pub async fn insert(pool: &PgPool, m: NewMission<'_>) -> Result<Uuid, DbError> {
     .await?;
 
     for p in &m.phases {
+        // `done_when` / `max_iterations` are promoted out of the phase config
+        // into real columns: the phase-runner sweep filters on them in SQL on
+        // every tick, and a JSONB probe in that hot path would be both slower
+        // and untypeable. The config blob remains the authoring surface (it is
+        // what the workflow recipe and the wizard write).
+        let done_when = p
+            .config
+            .get("done_when")
+            .and_then(|v| v.as_str())
+            .map(str::trim)
+            .filter(|s| !s.is_empty());
+        // Clamp server-side. The UI limits this too, but a runaway loop must
+        // not be one crafted request away.
+        let max_iterations = p
+            .config
+            .get("max_iterations")
+            .and_then(|v| v.as_i64())
+            .unwrap_or(1)
+            .clamp(1, MAX_PHASE_ITERATIONS);
+
         sqlx::query(
             "INSERT INTO mission_phases
-               (id, mission_id, kind, order_idx, status, config)
+               (id, mission_id, kind, order_idx, status, config, done_when, max_iterations)
-             VALUES ($1,$2,$3,$4,'pending',$5)",
+             VALUES ($1,$2,$3,$4,'pending',$5,$6,$7)",
         )
         .bind(Uuid::now_v7())
         .bind(mission_id)
         .bind(&p.kind)
         .bind(p.order_idx)
         .bind(&p.config)
+        .bind(done_when)
+        .bind(max_iterations as i32)
         .execute(&mut *tx)
         .await?;
     }

migrations/0061_phase_conditions.sql

@@ -0,0 +1,65 @@
+-- Goal conditions for mission phases.
+--
+-- Until now a phase completed when its topology_runs reached a terminal
+-- state -- purely structural, with no notion of whether the work was any
+-- good. `close_finished_phases` marked a phase `completed` whether the
+-- agents produced the artifact or wrote nothing at all.
+--
+-- `done_when` is a natural-language completion condition, judged after each
+-- pass by a model (see crates/cm-api/src/evaluator.rs). It follows the
+-- constraint the evaluator operates under: the judge cannot run commands, so
+-- the condition must be demonstrable from what the agents surfaced in their
+-- turn output.
+--
+-- NULL `done_when` preserves today's behaviour exactly: terminal runs ->
+-- completed, no evaluation, no extra model spend. Existing missions are
+-- unaffected.
+
+ALTER TABLE mission_phases
+    -- The completion condition. NULL = no evaluation (legacy behaviour).
+    ADD COLUMN done_when      TEXT,
+    -- Upper bound on passes. 1 = run once, matching current behaviour.
+    -- Capped server-side as well; this is a backstop against a runaway loop.
+    ADD COLUMN max_iterations INT NOT NULL DEFAULT 1,
+    -- Which pass the phase is on, 0-based.
+    ADD COLUMN iteration      INT NOT NULL DEFAULT 0;
+
+-- One verdict per (phase, iteration). Modelled on mission_phase_summaries
+-- (0060): the model emits structured JSON, we persist it with the model name
+-- so a verdict can be attributed, and keep the reason because it is both the
+-- explanation shown to the operator AND the guidance fed into the next pass.
+CREATE TABLE mission_phase_evaluations (
+    id         UUID PRIMARY KEY,
+    mission_id UUID NOT NULL REFERENCES missions(id) ON DELETE CASCADE,
+    phase_id   UUID NOT NULL REFERENCES mission_phases(id) ON DELETE CASCADE,
+    iteration  INT  NOT NULL,
+    -- Whether the condition held. Fail-closed: an unparseable or missing
+    -- verdict is recorded as false, never as "done".
+    met        BOOLEAN NOT NULL,
+    reason     TEXT NOT NULL,
+    -- The model that judged, e.g. "runtime:coordinator" or "claude-opus-4-8".
+    model      TEXT NOT NULL,
+    -- Set when the evaluator itself failed (transport, parse). `met` is false
+    -- in that case; this distinguishes "judged not done" from "could not judge".
+    error      TEXT,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    UNIQUE (phase_id, iteration)
+);
+
+CREATE INDEX mission_phase_evaluations_phase_idx
+    ON mission_phase_evaluations (phase_id, iteration DESC);
+CREATE INDEX mission_phase_evaluations_mission_idx
+    ON mission_phase_evaluations (mission_id, created_at DESC);
+
+-- Which pass produced a run. Without this, "are this phase's runs all
+-- finished?" matches pass 1's completed rows forever and a second pass would
+-- be declared done the instant it was enqueued.
+--
+-- (An `iteration` column existed on this table once and was dropped in 0053
+-- along with the legacy loops backend. This one is for mission phases.)
+ALTER TABLE topology_runs
+    ADD COLUMN iteration INT NOT NULL DEFAULT 0;
+
+CREATE INDEX topology_runs_phase_iteration_idx
+    ON topology_runs (mission_phase_id, iteration)
+    WHERE mission_phase_id IS NOT NULL;