Omar Sobh
94fecb526c
Every re-attempted phase now starts with a clean slate:
- phase_runner::launch_phase DELETEs prior status IN ('failed',
'cancelled') topology_runs for the phase before enqueuing the
new ones. Completed runs are kept for audit; only the failure
noise from earlier attempts goes.
- POST /api/missions/{id}/phases/{phase_id}/retry — resets a
failed/cancelled phase to 'pending' (auth-scoped to the calling
workspace + guarded on mission.status='running'). phase_runner
picks it up on the next 10s tick.
- MissionCanvas phase card grows a coral 'Retry' button, visible
only when phase.status='failed' and mission.status='running'.
Click → resets + refreshes; the prior failed run rows disappear
from the card as soon as phase_runner enqueues the new attempt.
Design: auto-purge in phase_runner rather than a separate 'clear
failed runs' endpoint. Users don't have to manually clean up before
retrying; the runner does it as part of the natural work of firing
a fresh attempt.
Verified: cargo check + tsc + eslint --quiet all green.
560 lines
18 KiB
Rust
560 lines
18 KiB
Rust
//! `/api/missions/*` — the unified workflow surface (Slice 1).
|
||
//!
|
||
//! This is a skeleton: create/list/get/status only. Slices 4–8 layer
|
||
//! richer behavior on top (template dispatch, phase execution, task
|
||
//! parsing, artifact rendering). The old `/api/research/*` +
|
||
//! `/api/loops/*` surfaces stay live in parallel until Slice 9.
|
||
|
||
use axum::{
|
||
extract::{Path, Query, State},
|
||
Json,
|
||
};
|
||
use cm_db::repo::missions::{
|
||
BenchmarkSnapshot, Mission, MissionArtifact, MissionPhase, MissionTask, NewMission,
|
||
NewMissionPhase,
|
||
};
|
||
use serde::{Deserialize, Serialize};
|
||
use serde_json::Value;
|
||
use uuid::Uuid;
|
||
|
||
use crate::{ApiError, AppState, Authed};
|
||
|
||
// ── Requests ─────────────────────────────────────────────────────
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
pub struct CreateMissionRequest {
|
||
pub title: String,
|
||
pub template_kind: String,
|
||
pub team_id: Option<Uuid>,
|
||
pub team_template_id: Option<Uuid>,
|
||
pub repo_id: Option<Uuid>,
|
||
#[serde(default = "default_schedule")]
|
||
pub schedule: Value,
|
||
pub description: Option<String>,
|
||
#[serde(default)]
|
||
pub config: Value,
|
||
#[serde(default)]
|
||
pub phases: Vec<PhaseSpec>,
|
||
/// Defaults to "zeroclaw". "local_herdr" requires target_node_id.
|
||
pub runtime_kind: Option<String>,
|
||
pub target_node_id: Option<Uuid>,
|
||
}
|
||
|
||
fn default_schedule() -> Value {
|
||
serde_json::json!({ "kind": "one_shot" })
|
||
}
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
pub struct PhaseSpec {
|
||
pub kind: String,
|
||
pub order_idx: i32,
|
||
#[serde(default)]
|
||
pub config: Value,
|
||
}
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
pub struct ListQuery {
|
||
#[serde(default = "default_limit")]
|
||
pub limit: i64,
|
||
}
|
||
fn default_limit() -> i64 {
|
||
50
|
||
}
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
pub struct SetStatusRequest {
|
||
pub status: String,
|
||
}
|
||
|
||
// ── Responses ────────────────────────────────────────────────────
|
||
|
||
#[derive(Debug, Serialize)]
|
||
pub struct MissionDetail {
|
||
#[serde(flatten)]
|
||
pub mission: Mission,
|
||
pub phases: Vec<MissionPhase>,
|
||
pub tasks: Vec<MissionTask>,
|
||
pub artifacts: Vec<MissionArtifact>,
|
||
pub benchmarks: Vec<BenchmarkSnapshot>,
|
||
}
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
pub struct BenchmarkTriggerRequest {
|
||
pub phase_id: Uuid,
|
||
/// Slot: "baseline" (records iteration 0) or "after"
|
||
/// (records iteration N + delta vs baseline).
|
||
pub slot: String,
|
||
#[serde(default)]
|
||
pub iteration: Option<i32>,
|
||
}
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
pub struct SecurityScanRequest {
|
||
pub phase_id: Uuid,
|
||
}
|
||
|
||
#[derive(Debug, Serialize)]
|
||
pub struct SecurityScanResponse {
|
||
pub findings: usize,
|
||
pub tasks: Vec<MissionTask>,
|
||
}
|
||
|
||
// ── Handlers ─────────────────────────────────────────────────────
|
||
|
||
pub async fn list(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Query(q): Query<ListQuery>,
|
||
) -> Result<Json<Vec<Mission>>, ApiError> {
|
||
let rows = cm_db::repo::missions::list_by_workspace(
|
||
&state.pool,
|
||
user.workspace_id.as_uuid(),
|
||
q.limit.clamp(1, 500),
|
||
)
|
||
.await?;
|
||
Ok(Json(rows))
|
||
}
|
||
|
||
pub async fn create(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Json(body): Json<CreateMissionRequest>,
|
||
) -> Result<Json<Mission>, ApiError> {
|
||
if body.title.trim().is_empty() {
|
||
return Err(ApiError::BadRequest);
|
||
}
|
||
// Validate runtime_kind + require target_node when local_herdr.
|
||
let runtime_kind = body.runtime_kind.as_deref().unwrap_or("zeroclaw");
|
||
match runtime_kind {
|
||
"zeroclaw" => {}
|
||
"local_herdr" => {
|
||
if body.target_node_id.is_none() {
|
||
return Err(ApiError::BadRequest);
|
||
}
|
||
}
|
||
_ => return Err(ApiError::BadRequest),
|
||
}
|
||
|
||
let new = NewMission {
|
||
workspace_id: user.workspace_id.as_uuid(),
|
||
title: body.title.trim(),
|
||
template_kind: body.template_kind.trim(),
|
||
team_id: body.team_id,
|
||
team_template_id: body.team_template_id,
|
||
repo_id: body.repo_id,
|
||
schedule: body.schedule,
|
||
description: body.description.as_deref(),
|
||
config: body.config,
|
||
runtime_kind: Some(runtime_kind),
|
||
target_node_id: body.target_node_id,
|
||
phases: body
|
||
.phases
|
||
.into_iter()
|
||
.map(|p| NewMissionPhase {
|
||
kind: p.kind,
|
||
order_idx: p.order_idx,
|
||
config: p.config,
|
||
})
|
||
.collect(),
|
||
};
|
||
let id = cm_db::repo::missions::insert(&state.pool, new).await?;
|
||
let mission = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::Internal)?;
|
||
Ok(Json(mission))
|
||
}
|
||
|
||
pub async fn get(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Path(id): Path<Uuid>,
|
||
) -> Result<Json<MissionDetail>, ApiError> {
|
||
let mission = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
let phases = cm_db::repo::missions::phases_for(&state.pool, id).await?;
|
||
let tasks = cm_db::repo::missions::tasks_for(&state.pool, id).await?;
|
||
let artifacts = cm_db::repo::missions::artifacts_for(&state.pool, id).await?;
|
||
let benchmarks = cm_db::repo::missions::benchmark_snapshots_for(&state.pool, id).await?;
|
||
Ok(Json(MissionDetail {
|
||
mission,
|
||
phases,
|
||
tasks,
|
||
artifacts,
|
||
benchmarks,
|
||
}))
|
||
}
|
||
|
||
/// POST /api/missions/{id}/benchmark — run the benchmark harness
|
||
/// against a phase. Slot='baseline' records iteration 0's
|
||
/// before_metrics; slot='after' with iteration=N records the
|
||
/// after_metrics + computes delta against baseline.
|
||
pub async fn trigger_benchmark(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Path(id): Path<Uuid>,
|
||
Json(body): Json<BenchmarkTriggerRequest>,
|
||
) -> Result<Json<Vec<BenchmarkSnapshot>>, ApiError> {
|
||
// Workspace scope check on the mission — 404 if not visible.
|
||
let _ = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
let result = match body.slot.as_str() {
|
||
"baseline" => crate::benchmark_runner::baseline(&state.pool, id, body.phase_id).await,
|
||
"after" => {
|
||
let iter = body.iteration.unwrap_or(1);
|
||
crate::benchmark_runner::after_iteration(&state.pool, id, body.phase_id, iter).await
|
||
}
|
||
_ => return Err(ApiError::BadRequest),
|
||
};
|
||
if let Err(e) = result {
|
||
eprintln!("benchmark trigger for mission {id}: {e}");
|
||
return Err(ApiError::Internal);
|
||
}
|
||
let snaps = cm_db::repo::missions::benchmark_snapshots_for(&state.pool, id).await?;
|
||
Ok(Json(snaps))
|
||
}
|
||
|
||
/// POST /api/missions/{id}/security-scan — run the security phase's
|
||
/// tool set (cargo-audit / gitleaks / trivy fs / semgrep) inside
|
||
/// the mission's team container and materialize each finding as a
|
||
/// mission_task keyed on the tool's canonical id.
|
||
pub async fn trigger_security_scan(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Path(id): Path<Uuid>,
|
||
Json(body): Json<SecurityScanRequest>,
|
||
) -> Result<Json<SecurityScanResponse>, ApiError> {
|
||
let _ = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
let findings = crate::security_scan::run(&state.pool, id, body.phase_id)
|
||
.await
|
||
.map_err(|e| {
|
||
eprintln!("security_scan for mission {id}: {e}");
|
||
ApiError::Internal
|
||
})?;
|
||
let tasks = cm_db::repo::missions::tasks_for(&state.pool, id).await?;
|
||
Ok(Json(SecurityScanResponse { findings, tasks }))
|
||
}
|
||
|
||
#[derive(Debug, Serialize)]
|
||
pub struct RefineResponse {
|
||
pub original: String,
|
||
pub refined: String,
|
||
}
|
||
|
||
/// POST /api/missions/{id}/refine — generate a coherent, sectioned
|
||
/// Markdown rewrite of the current description WITHOUT persisting.
|
||
/// Frontend renders a before/after diff; user hits Accept (PATCH
|
||
/// /description) or Cancel. Draft-only.
|
||
pub async fn refine(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Path(id): Path<Uuid>,
|
||
) -> Result<Json<RefineResponse>, ApiError> {
|
||
let result = crate::mission_refiner::refine(&state.pool, user.workspace_id, id)
|
||
.await
|
||
.map_err(|e| {
|
||
eprintln!("mission {id}: refine failed: {e}");
|
||
if e.contains("not found") {
|
||
ApiError::NotFound
|
||
} else if e.contains("empty") || e.contains("only allowed on draft") {
|
||
ApiError::BadRequest
|
||
} else {
|
||
ApiError::Internal
|
||
}
|
||
})?;
|
||
Ok(Json(RefineResponse {
|
||
original: result.original,
|
||
refined: result.refined,
|
||
}))
|
||
}
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
pub struct SetDescriptionRequest {
|
||
pub description: String,
|
||
}
|
||
|
||
/// PATCH /api/missions/{id}/description — commit a new description.
|
||
/// Draft-only. Used by the Refine Accept flow (and any future
|
||
/// direct-edit surface).
|
||
pub async fn set_description(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Path(id): Path<Uuid>,
|
||
Json(body): Json<SetDescriptionRequest>,
|
||
) -> Result<Json<Mission>, ApiError> {
|
||
let mission = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
if mission.status != "draft" {
|
||
return Err(ApiError::BadRequest);
|
||
}
|
||
cm_db::repo::missions::set_description(
|
||
&state.pool,
|
||
id,
|
||
user.workspace_id.as_uuid(),
|
||
&body.description,
|
||
)
|
||
.await?;
|
||
let mission = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
Ok(Json(mission))
|
||
}
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
pub struct UpdateMissionRequest {
|
||
#[serde(default)]
|
||
pub title: Option<String>,
|
||
#[serde(default)]
|
||
pub description: Option<String>,
|
||
}
|
||
|
||
/// PATCH /api/missions/{id} — edit title + description. Draft-only.
|
||
pub async fn update_meta(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Path(id): Path<Uuid>,
|
||
Json(body): Json<UpdateMissionRequest>,
|
||
) -> Result<Json<Mission>, ApiError> {
|
||
let mission = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
if mission.status != "draft" {
|
||
return Err(ApiError::BadRequest);
|
||
}
|
||
let title = body
|
||
.title
|
||
.as_deref()
|
||
.map(str::trim)
|
||
.filter(|s| !s.is_empty());
|
||
let description = body.description.as_deref();
|
||
cm_db::repo::missions::update_meta(
|
||
&state.pool,
|
||
id,
|
||
user.workspace_id.as_uuid(),
|
||
title,
|
||
description,
|
||
)
|
||
.await?;
|
||
let mission = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
Ok(Json(mission))
|
||
}
|
||
|
||
/// DELETE /api/missions/{id} — hard-delete. Allowed in any status;
|
||
/// the operator is expected to Cancel first if a run is in flight
|
||
/// (cascades will still fire either way).
|
||
pub async fn delete(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Path(id): Path<Uuid>,
|
||
) -> Result<Json<serde_json::Value>, ApiError> {
|
||
let deleted =
|
||
cm_db::repo::missions::delete(&state.pool, id, user.workspace_id.as_uuid()).await?;
|
||
if deleted == 0 {
|
||
return Err(ApiError::NotFound);
|
||
}
|
||
Ok(Json(serde_json::json!({ "deleted": true })))
|
||
}
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
pub struct HerdrDispatchRequest {
|
||
pub cli: String,
|
||
pub prompt: String,
|
||
}
|
||
|
||
#[derive(Debug, Serialize)]
|
||
pub struct HerdrDispatchResponse {
|
||
pub pane_id: String,
|
||
pub node_id: Uuid,
|
||
}
|
||
|
||
/// POST /api/missions/{id}/herdr-dispatch — manually spawn a Herdr
|
||
/// pane on the mission's target_node running `cli` with `prompt`.
|
||
/// Requires mission.runtime_kind = 'local_herdr' + target_node_id set.
|
||
/// Wizard integration + auto-dispatch land in later phases; this
|
||
/// exists so Phase 1b's fleet_herdr module can be exercised end-to-end
|
||
/// against a real node while the rest of the arc builds out.
|
||
pub async fn herdr_dispatch(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Path(id): Path<Uuid>,
|
||
Json(body): Json<HerdrDispatchRequest>,
|
||
) -> Result<Json<HerdrDispatchResponse>, ApiError> {
|
||
let mission = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
if mission.runtime_kind != "local_herdr" {
|
||
return Err(ApiError::BadRequest);
|
||
}
|
||
let node_id = mission.target_node_id.ok_or(ApiError::BadRequest)?;
|
||
let handle = crate::fleet_herdr::dispatch(
|
||
state.node_hub.clone(),
|
||
cm_domain::NodeId::from(node_id),
|
||
id,
|
||
body.cli.trim(),
|
||
body.prompt.trim(),
|
||
)
|
||
.await
|
||
.map_err(|e| {
|
||
eprintln!("herdr_dispatch mission {id}: {e}");
|
||
ApiError::Internal
|
||
})?;
|
||
Ok(Json(HerdrDispatchResponse {
|
||
pane_id: handle.pane_id,
|
||
node_id,
|
||
}))
|
||
}
|
||
|
||
/// GET /api/missions/{id}/teams — teams materialized for this mission,
|
||
/// grouped by purpose (research / coding / etc). Returns
|
||
/// [{ purpose, team_id, team_name }] so the Team tab can render
|
||
/// sections. The legacy single-team view falls back to
|
||
/// mission.team_id when this array is empty.
|
||
pub async fn list_teams(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Path(id): Path<Uuid>,
|
||
) -> Result<Json<Value>, ApiError> {
|
||
let _ = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
use sqlx::Row;
|
||
let rows = sqlx::query(
|
||
"SELECT mt.team_id::text AS team_id, mt.purpose, t.name AS team_name
|
||
FROM mission_teams mt
|
||
JOIN teams t ON t.id = mt.team_id
|
||
WHERE mt.mission_id = $1
|
||
ORDER BY mt.created_at ASC",
|
||
)
|
||
.bind(id)
|
||
.fetch_all(&state.pool)
|
||
.await?;
|
||
let teams: Vec<Value> = rows
|
||
.into_iter()
|
||
.map(|r| {
|
||
serde_json::json!({
|
||
"team_id": r.get::<String, _>("team_id"),
|
||
"purpose": r.get::<String, _>("purpose"),
|
||
"team_name": r.get::<String, _>("team_name"),
|
||
})
|
||
})
|
||
.collect();
|
||
Ok(Json(serde_json::json!({ "teams": teams })))
|
||
}
|
||
|
||
/// POST /api/missions/{id}/phases/{phase_id}/retry — reset a
|
||
/// failed / cancelled phase back to 'pending' so the phase_runner
|
||
/// picks it up on the next tick. The runner purges old failed
|
||
/// topology_runs for the phase before re-enqueuing, so the phase
|
||
/// card starts fresh on the retry.
|
||
pub async fn retry_phase(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Path((id, phase_id)): Path<(Uuid, Uuid)>,
|
||
) -> Result<Json<Value>, ApiError> {
|
||
// Scope check on the mission.
|
||
let mission = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
if mission.status != "running" {
|
||
return Err(ApiError::BadRequest);
|
||
}
|
||
let r = sqlx::query(
|
||
"UPDATE mission_phases
|
||
SET status = 'pending', started_at = NULL, completed_at = NULL
|
||
WHERE id = $1 AND mission_id = $2
|
||
AND status IN ('failed', 'cancelled')",
|
||
)
|
||
.bind(phase_id)
|
||
.bind(id)
|
||
.execute(&state.pool)
|
||
.await?;
|
||
if r.rows_affected() == 0 {
|
||
return Err(ApiError::NotFound);
|
||
}
|
||
Ok(Json(serde_json::json!({ "reset": true })))
|
||
}
|
||
|
||
/// GET /api/missions/{id}/runs — topology_runs bound to this mission,
|
||
/// newest first. Used by the Live tab to subscribe to per-run SSE.
|
||
pub async fn list_runs(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Path(id): Path<Uuid>,
|
||
) -> Result<Json<Value>, ApiError> {
|
||
// Scope check — 404 if the mission doesn't belong to this workspace.
|
||
let _ = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
let runs = cm_db::repo::topology_runs::list_by_mission(&state.pool, id, 50).await?;
|
||
Ok(Json(serde_json::json!({ "runs": runs })))
|
||
}
|
||
|
||
pub async fn set_status(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Path(id): Path<Uuid>,
|
||
Json(body): Json<SetStatusRequest>,
|
||
) -> Result<Json<Mission>, ApiError> {
|
||
let allowed = ["draft", "running", "completed", "failed", "cancelled"];
|
||
if !allowed.contains(&body.status.as_str()) {
|
||
return Err(ApiError::BadRequest);
|
||
}
|
||
|
||
// Snapshot prior state so we can detect the draft→running edge
|
||
// and fire the launch orchestrator (Slice 4).
|
||
let prior = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
|
||
// Draft→running requires a materializable team. Run the orchestrator
|
||
// BEFORE flipping status so a materialization failure keeps the
|
||
// mission in draft (no orphaned "running" mission with no agents).
|
||
if prior.status == "draft" && body.status == "running" {
|
||
// Materializable when we have any of:
|
||
// - team_id (already exists)
|
||
// - team_template_id (legacy single-team path)
|
||
// - config.phase_teams with at least one non-empty list (new multi-team)
|
||
let has_phase_teams = prior
|
||
.config
|
||
.get("phase_teams")
|
||
.and_then(|v| v.as_object())
|
||
.map(|obj| {
|
||
obj.values()
|
||
.any(|v| v.as_array().map(|a| !a.is_empty()).unwrap_or(false))
|
||
})
|
||
.unwrap_or(false);
|
||
if prior.team_id.is_none() && prior.team_template_id.is_none() && !has_phase_teams {
|
||
eprintln!(
|
||
"mission {id}: launch rejected — no team_id, no team_template_id, no config.phase_teams"
|
||
);
|
||
return Err(ApiError::BadRequest);
|
||
}
|
||
if let Err(e) = crate::mission_orchestrator::on_launch(
|
||
&state.pool,
|
||
user.workspace_id,
|
||
user.user_id,
|
||
id,
|
||
Some(state.node_hub.clone()),
|
||
)
|
||
.await
|
||
{
|
||
eprintln!("mission {id}: on_launch failed: {e}");
|
||
return Err(ApiError::Internal);
|
||
}
|
||
}
|
||
|
||
cm_db::repo::missions::set_status(&state.pool, id, user.workspace_id.as_uuid(), &body.status)
|
||
.await?;
|
||
|
||
let mission = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
Ok(Json(mission))
|
||
}
|