missions: phase_runner — actually execute mission phases
Root-cause fix for "we hit launch, waited overnight, nothing ran."
mission_orchestrator materialized teams + agents fine, but nothing
enqueued the actual work — mission_phases stayed 'pending' forever
and topology_runs count for the mission was 0.
New crates/cm-api/src/phase_runner.rs — background worker on 10s
poll that does three things:
1. start_pending_phases — for every mission_phase with
status='pending' AND parent mission.status='running' AND all
lower-order phases already 'completed', enqueue one
topology_runs row per team whose (mission_id, purpose) matches
the phase kind:
phase=research → teams with purpose='research'
phase=coding → teams with purpose='coding'
phase=benchmark → teams with purpose='coding' (fallback)
phase=security_scan → teams with purpose 'security' | 'coding'
Each run gets a phase-kind-specific task text combining the
mission title/description + a directive for that phase.
Flips phase to 'running' after enqueue.
2. close_finished_phases — SQL sweep that flips phases whose
topology_runs are all terminal to 'completed' (or 'failed' if
any run failed).
3. close_finished_missions — same shape for missions whose phases
are all terminal.
Spawned alongside task_card_worker in clawmates-server main.rs.
Ordering enforced by mission_phases.order_idx — a coding phase
doesn't fire until its research phase completes.
Idempotent: every state transition is guarded so double-firing on a
race is safe. When a mission has no matching teams for a phase (bad
wizard state), the phase stays pending and the runner logs a skip
rather than getting stuck in a fail loop.
Existing topology_worker picks up the queued runs and drives them
through the ZeroClaw executor as usual.
This commit is contained in:
@@ -288,6 +288,7 @@ async fn run() -> Result<(), String> {
|
|||||||
// for INT-XX markers in event payloads and upserts mission_tasks
|
// for INT-XX markers in event payloads and upserts mission_tasks
|
||||||
// rows so the canvas renders a live status timeline.
|
// rows so the canvas renders a live status timeline.
|
||||||
cm_api::task_card_worker::spawn(pool.clone());
|
cm_api::task_card_worker::spawn(pool.clone());
|
||||||
|
cm_api::phase_runner::spawn(pool.clone());
|
||||||
// PDF renderer worker (Slice 6): watches mission_artifacts for
|
// PDF renderer worker (Slice 6): watches mission_artifacts for
|
||||||
// MD entries with render_pdf_status='pending', calls the
|
// MD entries with render_pdf_status='pending', calls the
|
||||||
// configured LLM (default Gemini 2.5 Flash) for styled HTML,
|
// configured LLM (default Gemini 2.5 Flash) for styled HTML,
|
||||||
|
|||||||
@@ -16,6 +16,7 @@ pub mod mission_refiner;
|
|||||||
pub mod mission_workspace;
|
pub mod mission_workspace;
|
||||||
pub mod node_rules;
|
pub mod node_rules;
|
||||||
pub mod pdf_renderer;
|
pub mod pdf_renderer;
|
||||||
|
pub mod phase_runner;
|
||||||
pub mod quota;
|
pub mod quota;
|
||||||
mod recursive_exec;
|
mod recursive_exec;
|
||||||
mod routes;
|
mod routes;
|
||||||
|
|||||||
@@ -0,0 +1,274 @@
|
|||||||
|
//! 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);
|
||||||
|
|
||||||
|
pub fn spawn(pool: PgPool) {
|
||||||
|
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 {
|
||||||
|
eprintln!("phase_runner: sweep failed: {e}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn sweep_once(pool: &PgPool) -> Result<(), String> {
|
||||||
|
start_pending_phases(pool).await?;
|
||||||
|
close_finished_phases(pool).await?;
|
||||||
|
close_finished_missions(pool).await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Enqueue topology_runs for every phase whose predecessors are done.
|
||||||
|
async fn start_pending_phases(pool: &PgPool) -> 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,
|
||||||
|
m.workspace_id, m.title, m.description
|
||||||
|
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");
|
||||||
|
|
||||||
|
if let Err(e) = launch_phase(
|
||||||
|
pool,
|
||||||
|
phase_id,
|
||||||
|
mission_id,
|
||||||
|
&kind,
|
||||||
|
workspace_id,
|
||||||
|
&title,
|
||||||
|
description.as_deref(),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
eprintln!("phase_runner: launch phase {phase_id} failed: {e}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn launch_phase(
|
||||||
|
pool: &PgPool,
|
||||||
|
phase_id: Uuid,
|
||||||
|
mission_id: Uuid,
|
||||||
|
kind: &str,
|
||||||
|
workspace_id: Uuid,
|
||||||
|
title: &str,
|
||||||
|
description: Option<&str>,
|
||||||
|
) -> Result<(), String> {
|
||||||
|
// Which team purposes should execute this phase.
|
||||||
|
let purposes: &[&str] = match kind {
|
||||||
|
"research" => &["research", "mission"],
|
||||||
|
"coding" => &["coding", "mission"],
|
||||||
|
"benchmark" => &["coding", "mission"],
|
||||||
|
"security_scan" => &["security", "coding", "mission"],
|
||||||
|
_ => &["mission"],
|
||||||
|
};
|
||||||
|
|
||||||
|
// 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}"))?;
|
||||||
|
|
||||||
|
if team_rows.is_empty() {
|
||||||
|
eprintln!(
|
||||||
|
"phase_runner: mission {mission_id} phase {phase_id} ({kind}) has no matching teams — skipping (staying pending)"
|
||||||
|
);
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
|
||||||
|
let task = phase_task_text(kind, title, description);
|
||||||
|
|
||||||
|
for r in &team_rows {
|
||||||
|
let team_id: Uuid = r.get("team_id");
|
||||||
|
let graph: serde_json::Value = r.get("graph");
|
||||||
|
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)
|
||||||
|
VALUES ($1, $2, $3, 'run', 'queued', $4, 'team', $5, $6, $7)",
|
||||||
|
)
|
||||||
|
.bind(run_id)
|
||||||
|
.bind(workspace_id)
|
||||||
|
.bind(&task)
|
||||||
|
.bind(&graph)
|
||||||
|
.bind(team_id)
|
||||||
|
.bind(mission_id)
|
||||||
|
.bind(phase_id)
|
||||||
|
.execute(pool)
|
||||||
|
.await
|
||||||
|
.map_err(|e| format!("enqueue run for team {team_id}: {e}"))?;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Flip phase to running.
|
||||||
|
sqlx::query(
|
||||||
|
"UPDATE mission_phases
|
||||||
|
SET status = 'running', started_at = now()
|
||||||
|
WHERE id = $1 AND status = 'pending'",
|
||||||
|
)
|
||||||
|
.bind(phase_id)
|
||||||
|
.execute(pool)
|
||||||
|
.await
|
||||||
|
.map_err(|e| format!("mark phase {phase_id} running: {e}"))?;
|
||||||
|
eprintln!(
|
||||||
|
"phase_runner: mission {mission_id} phase {phase_id} ({kind}) launched with {} team(s)",
|
||||||
|
team_rows.len()
|
||||||
|
);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn phase_task_text(kind: &str, title: &str, description: Option<&str>) -> String {
|
||||||
|
let base = description.unwrap_or("").trim();
|
||||||
|
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 to the Obsidian vault or the \
|
||||||
|
mission's artifact directory. Emit INT-XX task markers \
|
||||||
|
for concrete follow-ups."
|
||||||
|
}
|
||||||
|
"coding" => {
|
||||||
|
"Your team is running the CODING phase of this mission. \
|
||||||
|
Implement the mission's acceptance criteria against the \
|
||||||
|
checked-out repo. Follow the workspace-repo-commit-protocol: \
|
||||||
|
small focused commits with test coverage. Emit COMPLETED: <INT-id> \
|
||||||
|
markers as you close research-produced tasks."
|
||||||
|
}
|
||||||
|
"benchmark" => {
|
||||||
|
"Your team is running the BENCHMARK phase of this mission. \
|
||||||
|
Author or extend benchmarks that measure the target change. \
|
||||||
|
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 the \
|
||||||
|
checked-out 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.",
|
||||||
|
};
|
||||||
|
format!("MISSION: {title}\n\n{directive}\n\nBRIEF:\n{base}")
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Close phases whose topology_runs are all terminal.
|
||||||
|
async fn close_finished_phases(pool: &PgPool) -> Result<(), String> {
|
||||||
|
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.status = 'failed'
|
||||||
|
) THEN 'failed'
|
||||||
|
ELSE 'completed'
|
||||||
|
END,
|
||||||
|
completed_at = now()
|
||||||
|
WHERE mp.status = 'running'
|
||||||
|
AND EXISTS (SELECT 1 FROM topology_runs r WHERE r.mission_phase_id = mp.id)
|
||||||
|
AND NOT EXISTS (
|
||||||
|
SELECT 1 FROM topology_runs r
|
||||||
|
WHERE r.mission_phase_id = mp.id
|
||||||
|
AND r.status NOT IN ('completed', 'failed', 'cancelled')
|
||||||
|
)",
|
||||||
|
)
|
||||||
|
.execute(pool)
|
||||||
|
.await
|
||||||
|
.map_err(|e| format!("close finished phases: {e}"))?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Close missions whose phases are all terminal.
|
||||||
|
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')
|
||||||
|
)",
|
||||||
|
)
|
||||||
|
.execute(pool)
|
||||||
|
.await
|
||||||
|
.map_err(|e| format!("close finished missions: {e}"))?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user