Omar SobhandClaude Opus 5
c573480955
Agent containers leaked two independent ways.
1. The four-step teardown (deprovision ZeroClaw -> reap_sandbox -> unlink
.brain/.onion -> hard_purge) was inlined at three call sites and two had
drifted. missions.rs::reap_mission_resources skipped reap_sandbox;
topology_worker::maybe_teardown_ephemeral_team skipped it and the brain
unlink; DELETE /api/claws/{id} (soft delete) released nothing at all, so an
offline claw that can never run again kept its container and bind mount
forever. All four now funnel through claws::purge_agent, with
release_claw_resources for the soft-delete case (containers gone, rows kept).
2. Both orphan reapers listed only the local driver, so a container placed on a
fleet node was invisible to the only backstop that could find it -- this is
what accumulated 144 tc-agent-* orphans on one node. NodeDriverProvider gains
node_ids() (backed by NodeHub::online_ids) and both reapers now sweep every
connected node. The remote sweep is TTL-only on purpose: the boot pass runs
with Duration::ZERO and would otherwise kill a container another instance is
mid-provision on.
Why it was invisible: agent_containers.agent_id is ON DELETE CASCADE, so
hard_purge took the registry row with the agent and left the container
permanently unreferenceable.
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
971 lines
33 KiB
Rust
971 lines
33 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 ─────────────────────────────────────────────────────
|
||
|
||
/// A mission plus the phase progress the list card needs. `mission` is
|
||
/// flattened, so the JSON is a strict SUPERSET of `Mission` — existing
|
||
/// consumers keep working and simply gain fields.
|
||
#[derive(Debug, Serialize)]
|
||
pub struct MissionListItem {
|
||
#[serde(flatten)]
|
||
pub mission: Mission,
|
||
pub phases_total: i64,
|
||
pub phases_done: i64,
|
||
/// Kind of the phase currently running, if any.
|
||
pub current_phase: Option<String>,
|
||
}
|
||
|
||
pub async fn list(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Query(q): Query<ListQuery>,
|
||
) -> Result<Json<Vec<MissionListItem>>, ApiError> {
|
||
let rows = cm_db::repo::missions::list_by_workspace(
|
||
&state.pool,
|
||
user.workspace_id.as_uuid(),
|
||
q.limit.clamp(1, 500),
|
||
)
|
||
.await?;
|
||
// One extra grouped query for the whole page, not one per mission.
|
||
let ids: Vec<Uuid> = rows.iter().map(|m| m.id).collect();
|
||
let progress = cm_db::repo::missions::phase_progress(&state.pool, &ids).await?;
|
||
let by_id: std::collections::HashMap<Uuid, (i64, i64, Option<String>)> = progress
|
||
.into_iter()
|
||
.map(|(id, total, done, running)| (id, (total, done, running)))
|
||
.collect();
|
||
Ok(Json(
|
||
rows.into_iter()
|
||
.map(|m| {
|
||
let (phases_total, phases_done, current_phase) =
|
||
by_id.get(&m.id).cloned().unwrap_or((0, 0, None));
|
||
MissionListItem {
|
||
mission: m,
|
||
phases_total,
|
||
phases_done,
|
||
current_phase,
|
||
}
|
||
})
|
||
.collect(),
|
||
))
|
||
}
|
||
|
||
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 ws = user.workspace_id.as_uuid();
|
||
// Verify the mission exists in this workspace before we start reaping.
|
||
let exists: Option<Uuid> =
|
||
sqlx::query_scalar("SELECT id FROM missions WHERE id = $1 AND workspace_id = $2")
|
||
.bind(id)
|
||
.bind(ws)
|
||
.fetch_optional(&state.pool)
|
||
.await
|
||
.map_err(|_| ApiError::Internal)?;
|
||
if exists.is_none() {
|
||
return Err(ApiError::NotFound);
|
||
}
|
||
|
||
// Reap every resource the mission provisioned BEFORE the DB delete, so
|
||
// nothing is left hanging. Runtime-side steps are best-effort (Postgres
|
||
// is authoritative; the daemon config is a cache the fleet sweeper can
|
||
// reconcile) — a failure logs and continues rather than blocking delete.
|
||
reap_mission_resources(&state, id).await;
|
||
|
||
let deleted = cm_db::repo::missions::delete(&state.pool, id, ws).await?;
|
||
if deleted == 0 {
|
||
return Err(ApiError::NotFound);
|
||
}
|
||
Ok(Json(serde_json::json!({ "deleted": true })))
|
||
}
|
||
|
||
/// Tear down all resources a mission created: its per-mission runtime
|
||
/// container + workspace dir, every claw (ZeroClaw config, `.brain` files,
|
||
/// and all DB rows via `hard_purge`), the (permanent-lifecycle) teams, and
|
||
/// its topology runs. Called before the `missions` row is deleted so the
|
||
/// `mission_teams` junction is still resolvable. Best-effort throughout.
|
||
async fn reap_mission_resources(state: &AppState, mission_id: Uuid) {
|
||
// 1. Resolve the mission's teams, then their claws.
|
||
let team_ids: Vec<Uuid> =
|
||
sqlx::query_scalar("SELECT team_id FROM mission_teams WHERE mission_id = $1")
|
||
.bind(mission_id)
|
||
.fetch_all(&state.pool)
|
||
.await
|
||
.unwrap_or_default();
|
||
let claw_ids: Vec<Uuid> = if team_ids.is_empty() {
|
||
Vec::new()
|
||
} else {
|
||
sqlx::query_scalar(
|
||
"SELECT DISTINCT claw_id FROM team_members WHERE team_id = ANY($1)",
|
||
)
|
||
.bind(&team_ids)
|
||
.fetch_all(&state.pool)
|
||
.await
|
||
.unwrap_or_default()
|
||
};
|
||
|
||
// 2. Reap each claw: ZeroClaw config → sandbox container → .brain files →
|
||
// all DB rows. Shared with the batch-delete reaper so this path cannot
|
||
// drift back into skipping the container teardown.
|
||
let provisioner = crate::runtime_provision::RuntimeProvisioner::from_env();
|
||
for cid in &claw_ids {
|
||
let report = crate::routes::claws::purge_agent(
|
||
&state.pool,
|
||
&state.runtime,
|
||
provisioner.as_ref(),
|
||
cm_domain::AgentId::from(*cid),
|
||
)
|
||
.await;
|
||
if let Err(e) = report.counts {
|
||
eprintln!("missions::delete: hard_purge claw {cid} failed (continuing): {e}");
|
||
}
|
||
}
|
||
|
||
// 3. Delete the (permanent-lifecycle) teams — no mission FK cascades them.
|
||
// team_members cascades from teams.
|
||
if !team_ids.is_empty() {
|
||
if let Err(e) = sqlx::query("DELETE FROM teams WHERE id = ANY($1)")
|
||
.bind(&team_ids)
|
||
.execute(&state.pool)
|
||
.await
|
||
{
|
||
eprintln!("missions::delete: delete teams for {mission_id} failed (continuing): {e}");
|
||
}
|
||
}
|
||
|
||
// 4. Delete this mission's topology runs (else they linger with
|
||
// mission_id nulled by the cascade and accumulate forever).
|
||
if let Err(e) = sqlx::query("DELETE FROM topology_runs WHERE mission_id = $1")
|
||
.bind(mission_id)
|
||
.execute(&state.pool)
|
||
.await
|
||
{
|
||
eprintln!("missions::delete: delete topology_runs for {mission_id} failed (continuing): {e}");
|
||
}
|
||
|
||
// 5. Tear down the per-mission runtime container + its workspace dir.
|
||
if let Some(mp) = crate::mission_runtime::MissionRuntimeProvisioner::from_env() {
|
||
if let Err(e) = mp.teardown_container(mission_id).await {
|
||
eprintln!("missions::delete: teardown container for {mission_id} failed (continuing): {e}");
|
||
}
|
||
}
|
||
|
||
eprintln!(
|
||
"missions::delete: reaped {} claw(s), {} team(s) for mission {mission_id}",
|
||
claw_ids.len(),
|
||
team_ids.len()
|
||
);
|
||
}
|
||
|
||
#[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}/phases/{phase_id}/summary — the completion
|
||
/// card produced by `phase_summarizer` for a terminal-state phase.
|
||
/// Returns 404 while the phase is still running / hasn't been
|
||
/// summarized yet.
|
||
pub async fn get_phase_summary(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Path((id, phase_id)): Path<(Uuid, Uuid)>,
|
||
) -> Result<Json<Value>, ApiError> {
|
||
// Scope check.
|
||
let _ = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
use sqlx::Row;
|
||
let row = sqlx::query(
|
||
"SELECT kind, model, narrative, metrics, sources, artifacts,
|
||
tooling, next_actions, generated_at, error
|
||
FROM mission_phase_summaries
|
||
WHERE mission_id = $1 AND phase_id = $2",
|
||
)
|
||
.bind(id)
|
||
.bind(phase_id)
|
||
.fetch_optional(&state.pool)
|
||
.await?;
|
||
let Some(r) = row else {
|
||
return Err(ApiError::NotFound);
|
||
};
|
||
let generated_at: time::OffsetDateTime = r.get("generated_at");
|
||
let payload = serde_json::json!({
|
||
"kind": r.get::<String, _>("kind"),
|
||
"model": r.get::<String, _>("model"),
|
||
"narrative": r.get::<String, _>("narrative"),
|
||
"metrics": r.get::<Value, _>("metrics"),
|
||
"sources": r.get::<Value, _>("sources"),
|
||
"artifacts": r.get::<Value, _>("artifacts"),
|
||
"tooling": r.get::<Value, _>("tooling"),
|
||
"next_actions": r.get::<Value, _>("next_actions"),
|
||
"generated_at": generated_at
|
||
.format(&time::format_description::well_known::Rfc3339)
|
||
.unwrap_or_default(),
|
||
"error": r.get::<Option<String>, _>("error"),
|
||
});
|
||
Ok(Json(payload))
|
||
}
|
||
|
||
/// 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))
|
||
}
|
||
|
||
// ── Output reader ────────────────────────────────────────────────
|
||
//
|
||
// The mission Output tab is a document reader, not a log tail. The
|
||
// phase-card preview endpoint (`routes::topology::get_run_output`) caps
|
||
// every turn at 6,000 chars, which shows only ~11% of a typical research
|
||
// brief (they run 40–55kB) with no way to read the rest. These two routes
|
||
// are the reader's data source: one lists every document in the mission
|
||
// for the outline rail, the other returns one document in full.
|
||
|
||
/// One agent turn's output, as a readable document.
|
||
#[derive(Debug, Serialize)]
|
||
pub struct MissionDocument {
|
||
pub run_id: Uuid,
|
||
pub phase_id: Option<Uuid>,
|
||
/// Index into the run's `checkpoint.outputs` array.
|
||
pub index: usize,
|
||
/// Topology node id (`n0`) — stable within the run's graph.
|
||
pub node_id: String,
|
||
/// The node's role (`code_archeologist`), i.e. what this agent was.
|
||
pub role: String,
|
||
/// Human title: the document's first markdown heading when it has
|
||
/// one, else its first non-empty line.
|
||
pub title: String,
|
||
pub chars: usize,
|
||
pub run_status: String,
|
||
}
|
||
|
||
#[derive(Debug, Serialize)]
|
||
pub struct MissionDocumentsResponse {
|
||
pub documents: Vec<MissionDocument>,
|
||
}
|
||
|
||
/// Derive a display title from a document's own text: prefer the first
|
||
/// markdown ATX heading, else the first non-empty line. Both are trimmed
|
||
/// to keep the rail readable.
|
||
fn document_title(body: &str, fallback: &str) -> String {
|
||
const MAX: usize = 90;
|
||
let heading = body
|
||
.lines()
|
||
.map(str::trim)
|
||
.find(|l| l.starts_with('#'))
|
||
.map(|l| l.trim_start_matches('#').trim());
|
||
let line = heading.or_else(|| body.lines().map(str::trim).find(|l| !l.is_empty()));
|
||
match line {
|
||
Some(l) if !l.is_empty() => {
|
||
if l.chars().count() > MAX {
|
||
format!("{}…", l.chars().take(MAX).collect::<String>())
|
||
} else {
|
||
l.to_string()
|
||
}
|
||
}
|
||
_ => fallback.to_string(),
|
||
}
|
||
}
|
||
|
||
/// Map a run's graph node index → (node_id, role). The reader labels each
|
||
/// document by the agent that produced it; `checkpoint.outputs[i]`
|
||
/// corresponds to `graph.nodes[i]` (the worker appends one output per
|
||
/// step, in node order).
|
||
fn nodes_of(graph: Option<&Value>) -> Vec<(String, String)> {
|
||
graph
|
||
.and_then(|g| g.get("nodes"))
|
||
.and_then(|n| n.as_array())
|
||
.map(|arr| {
|
||
arr.iter()
|
||
.map(|n| {
|
||
(
|
||
n.get("id").and_then(|v| v.as_str()).unwrap_or("").to_string(),
|
||
n.get("role")
|
||
.and_then(|v| v.as_str())
|
||
.unwrap_or("agent")
|
||
.to_string(),
|
||
)
|
||
})
|
||
.collect()
|
||
})
|
||
.unwrap_or_default()
|
||
}
|
||
|
||
fn outputs_of(checkpoint: Option<&Value>) -> Vec<String> {
|
||
checkpoint
|
||
.and_then(|c| c.get("outputs"))
|
||
.and_then(|o| o.as_array())
|
||
.map(|arr| {
|
||
arr.iter()
|
||
.map(|v| match v {
|
||
Value::String(s) => s.clone(),
|
||
other => other.to_string(),
|
||
})
|
||
.collect()
|
||
})
|
||
.unwrap_or_default()
|
||
}
|
||
|
||
/// `GET /api/missions/{id}/documents` — every agent output in the mission,
|
||
/// oldest run first, as a flat list the reader groups by phase. Bodies are
|
||
/// NOT included; the rail only needs titles and sizes.
|
||
pub async fn list_documents(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Path(id): Path<Uuid>,
|
||
) -> Result<Json<MissionDocumentsResponse>, ApiError> {
|
||
let _ = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
let source =
|
||
cm_db::repo::topology_runs::documents_source_for_mission(&state.pool, id).await?;
|
||
|
||
let mut documents = Vec::new();
|
||
for (run_id, phase_id, run_status, graph, checkpoint) in source {
|
||
let nodes = nodes_of(graph.as_ref());
|
||
for (index, body) in outputs_of(checkpoint.as_ref()).into_iter().enumerate() {
|
||
let (node_id, role) = nodes
|
||
.get(index)
|
||
.cloned()
|
||
.unwrap_or_else(|| (format!("n{index}"), "agent".to_string()));
|
||
let fallback = format!("Turn {}", index + 1);
|
||
documents.push(MissionDocument {
|
||
run_id,
|
||
phase_id,
|
||
index,
|
||
node_id,
|
||
title: document_title(&body, &fallback),
|
||
role,
|
||
chars: body.chars().count(),
|
||
run_status: run_status.clone(),
|
||
});
|
||
}
|
||
}
|
||
Ok(Json(MissionDocumentsResponse { documents }))
|
||
}
|
||
|
||
#[derive(Debug, Serialize)]
|
||
pub struct MissionDocumentBody {
|
||
pub run_id: Uuid,
|
||
pub index: usize,
|
||
pub role: String,
|
||
pub title: String,
|
||
/// The complete output text — untruncated, which is the whole point.
|
||
pub body: String,
|
||
pub chars: usize,
|
||
}
|
||
|
||
/// `GET /api/missions/{id}/documents/{run_id}/{index}` — one document in
|
||
/// full. Separate from the list so opening the Output tab doesn't pull
|
||
/// every brief in the mission over the wire at once.
|
||
pub async fn get_document(
|
||
State(state): State<AppState>,
|
||
Authed(user): Authed,
|
||
Path((id, run_id, index)): Path<(Uuid, Uuid, usize)>,
|
||
) -> Result<Json<MissionDocumentBody>, ApiError> {
|
||
let _ = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
|
||
.await?
|
||
.ok_or(ApiError::NotFound)?;
|
||
// Scope the run to the mission as well, so a valid run id from another
|
||
// mission (or workspace) can't be read through this path.
|
||
let source =
|
||
cm_db::repo::topology_runs::documents_source_for_mission(&state.pool, id).await?;
|
||
let (_, _, _, graph, checkpoint) = source
|
||
.into_iter()
|
||
.find(|(rid, _, _, _, _)| *rid == run_id)
|
||
.ok_or(ApiError::NotFound)?;
|
||
|
||
let body = outputs_of(checkpoint.as_ref())
|
||
.into_iter()
|
||
.nth(index)
|
||
.ok_or(ApiError::NotFound)?;
|
||
let role = nodes_of(graph.as_ref())
|
||
.get(index)
|
||
.map(|(_, r)| r.clone())
|
||
.unwrap_or_else(|| "agent".to_string());
|
||
let fallback = format!("Turn {}", index + 1);
|
||
Ok(Json(MissionDocumentBody {
|
||
run_id,
|
||
index,
|
||
title: document_title(&body, &fallback),
|
||
role,
|
||
chars: body.chars().count(),
|
||
body,
|
||
}))
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
#[test]
|
||
fn title_prefers_first_markdown_heading() {
|
||
let body = "I'll start by exploring.\n\n# ClawHDF5 Research Report\n\ntext";
|
||
assert_eq!(document_title(body, "Turn 1"), "ClawHDF5 Research Report");
|
||
}
|
||
|
||
#[test]
|
||
fn title_falls_back_to_first_nonempty_line() {
|
||
let body = "\n\n Architecture notes for the io crate\nmore\n";
|
||
assert_eq!(
|
||
document_title(body, "Turn 1"),
|
||
"Architecture notes for the io crate"
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn title_falls_back_to_label_when_empty() {
|
||
assert_eq!(document_title(" \n\n", "Turn 3"), "Turn 3");
|
||
}
|
||
|
||
#[test]
|
||
fn title_is_truncated() {
|
||
let body = format!("# {}", "x".repeat(200));
|
||
let t = document_title(&body, "Turn 1");
|
||
assert!(t.ends_with('…'));
|
||
assert_eq!(t.chars().count(), 91);
|
||
}
|
||
|
||
#[test]
|
||
fn nodes_and_outputs_are_positionally_aligned() {
|
||
let graph = serde_json::json!({
|
||
"nodes": [
|
||
{"id": "n0", "role": "code_archeologist"},
|
||
{"id": "n1", "role": "architecture_mapper"}
|
||
]
|
||
});
|
||
let cp = serde_json::json!({ "outputs": ["first brief", "second brief"] });
|
||
let nodes = nodes_of(Some(&graph));
|
||
let outs = outputs_of(Some(&cp));
|
||
assert_eq!(nodes[1], ("n1".into(), "architecture_mapper".into()));
|
||
assert_eq!(outs[1], "second brief");
|
||
}
|
||
|
||
#[test]
|
||
fn missing_graph_or_checkpoint_yields_no_documents() {
|
||
assert!(nodes_of(None).is_empty());
|
||
assert!(outputs_of(None).is_empty());
|
||
assert!(outputs_of(Some(&serde_json::json!({}))).is_empty());
|
||
}
|
||
}
|