Three things, all visible on the agents page.
**The roster looked like it was multiplying.** The sidebar flattened
orgs → companies → teams → agents, which renders a claw once per TEAM it
belongs to. Claws are reused across missions now, so a crew of five that had
run five missions appeared as twenty-five rows of the same five people. The
data was right and the view was lying. `GET /api/workforce` returns the roster
grouped by mission, and the tree renders each mission as a collapsible group,
so the repetition means something: the same colleague under each mission they
staffed. Claws on no mission come back under "Not on a mission" rather than
vanishing. The root now counts DISTINCT people, not rows.
**Agents were named after their jobs.** A team came back as planner, coder,
tester, reviewer, committer — the UI showed the same word twice (name on top,
role beneath) and the roster read as a stack of job tickets. New claws get a
given name from a deliberately wide pool (Amara, Vijay, Tomasz, Meredith…),
unique against the workspace roster AND within the team being minted. The role
is untouched in `job_title`, which is what the mission machinery binds on:
team_members.role_slot and the topology node carry the slot, so nothing
downstream keys off the display name. A reused claw keeps the name it had.
**Two latent reap bugs found while investigating a leak that was not one.**
Containers of completed missions are removed by `spawn_sweeper` after a
30-minute grace, and it works — an earlier report of leaking containers was me
reading that deliberate grace as a bug. But:
- the sweeper cleared the runtime binding even when teardown FAILED, and it
selects on `runtime_endpoint IS NOT NULL`. One transient docker error would
therefore hide a surviving container from the only thing that would retry
it, permanently. It now asks docker whether the container actually
survived: gone means clear, still there means keep the binding and retry —
which closes the orphan path without reintroducing the infinite retry the
original comment was guarding against.
- `set_runtime_binding` discarded rows_affected, so a mismatched workspace
updated nothing and returned Ok. The binding is how the sweeper finds a
container; a silent no-op there leaks one with no record of anything wrong.
Co-Authored-By: Claude Opus 5 <[email protected]>
859 lines
30 KiB
Rust
859 lines
30 KiB
Rust
//! Missions — the unified top-level workflow entity introduced in
|
|
//! Slice 1 of the research/loops consolidation.
|
|
//!
|
|
//! A mission composes one or more [`MissionPhase`]s. Existing
|
|
//! `research_topics` and `loops` rows are shadow-migrated into
|
|
//! `missions` + `mission_phases` by migration 0047 and by continued
|
|
//! writes from the compat shims (Slice 2). Old tables stay live and
|
|
//! writable until Slice 9's big-bang cutover.
|
|
|
|
use serde::{Deserialize, Serialize};
|
|
use serde_json::Value;
|
|
use sqlx::PgPool;
|
|
use time::OffsetDateTime;
|
|
use uuid::Uuid;
|
|
|
|
use crate::DbError;
|
|
|
|
// ── Types ────────────────────────────────────────────────────────
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct Mission {
|
|
pub id: Uuid,
|
|
pub workspace_id: Uuid,
|
|
pub title: String,
|
|
pub template_kind: String,
|
|
pub team_id: Option<Uuid>,
|
|
pub team_template_id: Option<Uuid>,
|
|
pub repo_id: Option<Uuid>,
|
|
pub schedule: Value,
|
|
pub status: String,
|
|
pub description: Option<String>,
|
|
pub config: Value,
|
|
/// 'zeroclaw' (default, headless) | 'local_herdr' (attended, fleet node)
|
|
pub runtime_kind: String,
|
|
/// FK → nodes(id); only relevant when runtime_kind = 'local_herdr'
|
|
pub target_node_id: Option<Uuid>,
|
|
/// Which per-CLI rootfs a `microvm` mission boots. NULL = the node's default
|
|
/// image. Read by placement (a node must HOLD this image) and by the executor
|
|
/// (it is passed to `vm_create`).
|
|
pub backend: Option<String>,
|
|
/// Per-mission ZeroClaw runtime container name (C3 workspace isolation).
|
|
/// Null until `mission_runtime::ensure_container` provisions it.
|
|
pub runtime_container_name: Option<String>,
|
|
/// Gateway URL the topology_worker dials for this mission's runs.
|
|
pub runtime_endpoint: Option<String>,
|
|
/// One-time pairing code captured from the fresh gateway's startup
|
|
/// log. topology_worker uses it to lazy-pair with this specific
|
|
/// mission runtime instead of the shared-runtime env token.
|
|
pub runtime_pairing_code: Option<String>,
|
|
#[serde(with = "time::serde::rfc3339")]
|
|
pub created_at: OffsetDateTime,
|
|
#[serde(with = "time::serde::rfc3339")]
|
|
pub updated_at: OffsetDateTime,
|
|
#[serde(with = "time::serde::rfc3339::option")]
|
|
pub completed_at: Option<OffsetDateTime>,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct MissionPhase {
|
|
pub id: Uuid,
|
|
pub mission_id: Uuid,
|
|
pub kind: String,
|
|
pub order_idx: i32,
|
|
pub status: String,
|
|
pub config: Value,
|
|
/// Completion condition. `None` = the phase completes as soon as its runs
|
|
/// finish, with no evaluation (the pre-conditions behaviour).
|
|
pub done_when: Option<String>,
|
|
/// Upper bound on passes; 1 means run once.
|
|
pub max_iterations: i32,
|
|
/// Which pass the phase is on, 0-based.
|
|
pub iteration: i32,
|
|
#[serde(with = "time::serde::rfc3339::option")]
|
|
pub started_at: Option<OffsetDateTime>,
|
|
#[serde(with = "time::serde::rfc3339::option")]
|
|
pub completed_at: Option<OffsetDateTime>,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct MissionTask {
|
|
pub id: Uuid,
|
|
pub mission_id: Uuid,
|
|
pub phase_id: Uuid,
|
|
pub external_id: Option<String>,
|
|
pub title: String,
|
|
pub assigned_agent_id: Option<Uuid>,
|
|
pub status: String,
|
|
pub run_id: Option<Uuid>,
|
|
pub artifact_paths: Vec<String>,
|
|
#[serde(with = "time::serde::rfc3339")]
|
|
pub created_at: OffsetDateTime,
|
|
#[serde(with = "time::serde::rfc3339")]
|
|
pub updated_at: OffsetDateTime,
|
|
#[serde(with = "time::serde::rfc3339::option")]
|
|
pub completed_at: Option<OffsetDateTime>,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct MissionArtifact {
|
|
pub id: Uuid,
|
|
pub mission_id: Uuid,
|
|
pub phase_id: Option<Uuid>,
|
|
pub path: String,
|
|
pub kind: String,
|
|
pub mime: Option<String>,
|
|
pub generated_by_run: Option<Uuid>,
|
|
pub rendered_pdf_path: Option<String>,
|
|
pub render_pdf_status: String,
|
|
pub render_pdf_error: Option<String>,
|
|
pub title: Option<String>,
|
|
pub metadata: Value,
|
|
#[serde(with = "time::serde::rfc3339")]
|
|
pub created_at: OffsetDateTime,
|
|
#[serde(with = "time::serde::rfc3339")]
|
|
pub updated_at: OffsetDateTime,
|
|
}
|
|
|
|
/// Convenience payload for creating a mission from the wizard.
|
|
/// Callers pass `phases` describing the desired plan; the create call
|
|
/// inserts the mission row + the corresponding mission_phase rows in
|
|
/// one transaction so partial creates never leak.
|
|
#[derive(Debug, Clone)]
|
|
pub struct NewMission<'a> {
|
|
pub workspace_id: Uuid,
|
|
pub title: &'a str,
|
|
pub template_kind: &'a str,
|
|
pub team_id: Option<Uuid>,
|
|
pub team_template_id: Option<Uuid>,
|
|
pub repo_id: Option<Uuid>,
|
|
pub schedule: Value,
|
|
pub description: Option<&'a str>,
|
|
pub config: Value,
|
|
/// Defaults to 'zeroclaw' when None.
|
|
pub runtime_kind: Option<&'a str>,
|
|
pub target_node_id: Option<Uuid>,
|
|
/// Which per-CLI image a `microvm` mission boots. NULL = the node's default
|
|
/// rootfs. Deliberately unconstrained in the schema: which images exist is a
|
|
/// property of the NODES, not of the database.
|
|
pub backend: Option<&'a str>,
|
|
/// Independent validator for this mission's verdicts. `None` = deployment
|
|
/// default; `Some("")` = explicitly none. See migration 0068.
|
|
pub validator_model: Option<&'a str>,
|
|
/// `"claude_code"` to ask for an agent team; `None` = solo. See 0069.
|
|
pub team_engine: Option<&'a str>,
|
|
pub phases: Vec<NewMissionPhase>,
|
|
}
|
|
|
|
#[derive(Debug, Clone)]
|
|
pub struct NewMissionPhase {
|
|
pub kind: String,
|
|
pub order_idx: i32,
|
|
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.
|
|
/// Returns the newly-minted mission id.
|
|
pub async fn insert(pool: &PgPool, m: NewMission<'_>) -> Result<Uuid, DbError> {
|
|
let mission_id = Uuid::now_v7();
|
|
let mut tx = pool.begin().await?;
|
|
|
|
sqlx::query(
|
|
"INSERT INTO missions
|
|
(id, workspace_id, title, template_kind, team_id,
|
|
team_template_id, repo_id, schedule, status, description, config,
|
|
runtime_kind, target_node_id, backend, validator_model, team_engine)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,'draft',$9,$10,
|
|
COALESCE($11,'zeroclaw'),$12,$13,$14,$15)",
|
|
)
|
|
.bind(mission_id)
|
|
.bind(m.workspace_id)
|
|
.bind(m.title)
|
|
.bind(m.template_kind)
|
|
.bind(m.team_id)
|
|
.bind(m.team_template_id)
|
|
.bind(m.repo_id)
|
|
.bind(&m.schedule)
|
|
.bind(m.description)
|
|
.bind(&m.config)
|
|
.bind(m.runtime_kind)
|
|
.bind(m.target_node_id)
|
|
.bind(m.backend)
|
|
.bind(m.validator_model)
|
|
.bind(m.team_engine)
|
|
.execute(&mut *tx)
|
|
.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, done_when, max_iterations)
|
|
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?;
|
|
}
|
|
|
|
tx.commit().await?;
|
|
Ok(mission_id)
|
|
}
|
|
|
|
/// Workspace-scoped fetch — returns None when the id belongs to a
|
|
/// different workspace so callers can't leak cross-workspace metadata.
|
|
pub async fn get(pool: &PgPool, id: Uuid, workspace_id: Uuid) -> Result<Option<Mission>, DbError> {
|
|
use sqlx::Row;
|
|
let row = sqlx::query(
|
|
"SELECT id, workspace_id, title, template_kind, team_id,
|
|
team_template_id, repo_id, schedule, status,
|
|
description, config, runtime_kind, target_node_id, backend,
|
|
runtime_container_name, runtime_endpoint, runtime_pairing_code,
|
|
created_at, updated_at, completed_at
|
|
FROM missions WHERE id = $1 AND workspace_id = $2",
|
|
)
|
|
.bind(id)
|
|
.bind(workspace_id)
|
|
.fetch_optional(pool)
|
|
.await?;
|
|
Ok(row.map(|r| Mission {
|
|
id: r.get("id"),
|
|
workspace_id: r.get("workspace_id"),
|
|
title: r.get("title"),
|
|
template_kind: r.get("template_kind"),
|
|
team_id: r.get("team_id"),
|
|
team_template_id: r.get("team_template_id"),
|
|
repo_id: r.get("repo_id"),
|
|
schedule: r.get("schedule"),
|
|
status: r.get("status"),
|
|
description: r.get("description"),
|
|
config: r.get("config"),
|
|
runtime_kind: r.get("runtime_kind"),
|
|
target_node_id: r.get("target_node_id"),
|
|
backend: r.get("backend"),
|
|
runtime_container_name: r.get("runtime_container_name"),
|
|
runtime_endpoint: r.get("runtime_endpoint"),
|
|
runtime_pairing_code: r.get("runtime_pairing_code"),
|
|
created_at: r.get("created_at"),
|
|
updated_at: r.get("updated_at"),
|
|
completed_at: r.get("completed_at"),
|
|
}))
|
|
}
|
|
|
|
/// List missions for a workspace, newest-first.
|
|
pub async fn list_by_workspace(
|
|
pool: &PgPool,
|
|
workspace_id: Uuid,
|
|
limit: i64,
|
|
) -> Result<Vec<Mission>, DbError> {
|
|
use sqlx::Row;
|
|
let rows = sqlx::query(
|
|
"SELECT id, workspace_id, title, template_kind, team_id,
|
|
team_template_id, repo_id, schedule, status,
|
|
description, config, runtime_kind, target_node_id, backend,
|
|
runtime_container_name, runtime_endpoint, runtime_pairing_code,
|
|
created_at, updated_at, completed_at
|
|
FROM missions WHERE workspace_id = $1
|
|
ORDER BY created_at DESC LIMIT $2",
|
|
)
|
|
.bind(workspace_id)
|
|
.bind(limit)
|
|
.fetch_all(pool)
|
|
.await?;
|
|
Ok(rows
|
|
.into_iter()
|
|
.map(|r| Mission {
|
|
id: r.get("id"),
|
|
workspace_id: r.get("workspace_id"),
|
|
title: r.get("title"),
|
|
template_kind: r.get("template_kind"),
|
|
team_id: r.get("team_id"),
|
|
team_template_id: r.get("team_template_id"),
|
|
repo_id: r.get("repo_id"),
|
|
schedule: r.get("schedule"),
|
|
status: r.get("status"),
|
|
description: r.get("description"),
|
|
config: r.get("config"),
|
|
runtime_kind: r.get("runtime_kind"),
|
|
target_node_id: r.get("target_node_id"),
|
|
backend: r.get("backend"),
|
|
runtime_container_name: r.get("runtime_container_name"),
|
|
runtime_endpoint: r.get("runtime_endpoint"),
|
|
runtime_pairing_code: r.get("runtime_pairing_code"),
|
|
created_at: r.get("created_at"),
|
|
updated_at: r.get("updated_at"),
|
|
completed_at: r.get("completed_at"),
|
|
})
|
|
.collect())
|
|
}
|
|
|
|
pub async fn set_description(
|
|
pool: &PgPool,
|
|
id: Uuid,
|
|
workspace_id: Uuid,
|
|
description: &str,
|
|
) -> Result<(), DbError> {
|
|
sqlx::query(
|
|
"UPDATE missions
|
|
SET description = $3, updated_at = now()
|
|
WHERE id = $1 AND workspace_id = $2",
|
|
)
|
|
.bind(id)
|
|
.bind(workspace_id)
|
|
.bind(description)
|
|
.execute(pool)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Patch title + description in one shot. Either field `None` = leave
|
|
/// as-is (uses COALESCE so partial edits don't clobber the other).
|
|
pub async fn update_meta(
|
|
pool: &PgPool,
|
|
id: Uuid,
|
|
workspace_id: Uuid,
|
|
title: Option<&str>,
|
|
description: Option<&str>,
|
|
) -> Result<(), DbError> {
|
|
sqlx::query(
|
|
"UPDATE missions
|
|
SET title = COALESCE($3, title),
|
|
description = COALESCE($4, description),
|
|
updated_at = now()
|
|
WHERE id = $1 AND workspace_id = $2",
|
|
)
|
|
.bind(id)
|
|
.bind(workspace_id)
|
|
.bind(title)
|
|
.bind(description)
|
|
.execute(pool)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Bind a mission to its per-mission runtime container + endpoint.
|
|
/// Called by `mission_runtime::ensure_container` after the docker
|
|
/// container is running. Null endpoint clears the binding (used by
|
|
/// the teardown sweeper).
|
|
pub async fn set_runtime_binding(
|
|
pool: &PgPool,
|
|
id: Uuid,
|
|
workspace_id: Uuid,
|
|
container_name: Option<&str>,
|
|
endpoint: Option<&str>,
|
|
pairing_code: Option<&str>,
|
|
) -> Result<(), DbError> {
|
|
let r = sqlx::query(
|
|
"UPDATE missions
|
|
SET runtime_container_name = $3,
|
|
runtime_endpoint = $4,
|
|
runtime_pairing_code = $5,
|
|
updated_at = now()
|
|
WHERE id = $1 AND workspace_id = $2",
|
|
)
|
|
.bind(id)
|
|
.bind(workspace_id)
|
|
.bind(container_name)
|
|
.bind(endpoint)
|
|
.bind(pairing_code)
|
|
.execute(pool)
|
|
.await?;
|
|
// A `WHERE id = $1 AND workspace_id = $2` that matches nothing is not an
|
|
// error to sqlx — it updates zero rows and returns Ok. That made a
|
|
// mismatched workspace indistinguishable from a successful bind, and the
|
|
// binding is what the sweeper uses to find a mission's container: a silent
|
|
// no-op here leaks a container with no record that anything went wrong.
|
|
// Callers log this rather than aborting, which is the point — it becomes
|
|
// visible instead of invisible.
|
|
if r.rows_affected() == 0 {
|
|
return Err(DbError::NotFound);
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Hard-delete a mission. Cascades via FKs on mission_phases /
|
|
/// mission_tasks / mission_artifacts / benchmark_snapshots (all
|
|
/// declared ON DELETE CASCADE in 0047).
|
|
pub async fn delete(pool: &PgPool, id: Uuid, workspace_id: Uuid) -> Result<u64, DbError> {
|
|
let r = sqlx::query("DELETE FROM missions WHERE id = $1 AND workspace_id = $2")
|
|
.bind(id)
|
|
.bind(workspace_id)
|
|
.execute(pool)
|
|
.await?;
|
|
Ok(r.rows_affected())
|
|
}
|
|
|
|
pub async fn set_status(
|
|
pool: &PgPool,
|
|
id: Uuid,
|
|
workspace_id: Uuid,
|
|
status: &str,
|
|
) -> Result<(), DbError> {
|
|
let completed = matches!(status, "completed" | "failed" | "cancelled");
|
|
sqlx::query(
|
|
"UPDATE missions
|
|
SET status = $3,
|
|
updated_at = now(),
|
|
completed_at = CASE WHEN $4 THEN now() ELSE completed_at END
|
|
WHERE id = $1 AND workspace_id = $2",
|
|
)
|
|
.bind(id)
|
|
.bind(workspace_id)
|
|
.bind(status)
|
|
.bind(completed)
|
|
.execute(pool)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
// ── Phases ───────────────────────────────────────────────────────
|
|
|
|
pub async fn phases_for(pool: &PgPool, mission_id: Uuid) -> Result<Vec<MissionPhase>, DbError> {
|
|
use sqlx::Row;
|
|
let rows = sqlx::query(
|
|
"SELECT id, mission_id, kind, order_idx, status, config,
|
|
done_when, max_iterations, iteration,
|
|
started_at, completed_at
|
|
FROM mission_phases WHERE mission_id = $1
|
|
ORDER BY order_idx ASC",
|
|
)
|
|
.bind(mission_id)
|
|
.fetch_all(pool)
|
|
.await?;
|
|
Ok(rows
|
|
.into_iter()
|
|
.map(|r| MissionPhase {
|
|
id: r.get("id"),
|
|
mission_id: r.get("mission_id"),
|
|
kind: r.get("kind"),
|
|
order_idx: r.get("order_idx"),
|
|
status: r.get("status"),
|
|
config: r.get("config"),
|
|
done_when: r.get("done_when"),
|
|
max_iterations: r.get("max_iterations"),
|
|
iteration: r.get("iteration"),
|
|
started_at: r.get("started_at"),
|
|
completed_at: r.get("completed_at"),
|
|
})
|
|
.collect())
|
|
}
|
|
|
|
pub async fn set_phase_status(pool: &PgPool, phase_id: Uuid, status: &str) -> Result<(), DbError> {
|
|
let is_start = status == "running";
|
|
let is_end = matches!(status, "completed" | "failed" | "skipped");
|
|
sqlx::query(
|
|
"UPDATE mission_phases
|
|
SET status = $2,
|
|
started_at = CASE WHEN $3 AND started_at IS NULL THEN now() ELSE started_at END,
|
|
completed_at = CASE WHEN $4 THEN now() ELSE completed_at END
|
|
WHERE id = $1",
|
|
)
|
|
.bind(phase_id)
|
|
.bind(status)
|
|
.bind(is_start)
|
|
.bind(is_end)
|
|
.execute(pool)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
// ── Tasks ────────────────────────────────────────────────────────
|
|
|
|
#[derive(Debug, Clone)]
|
|
pub struct UpsertTask<'a> {
|
|
pub mission_id: Uuid,
|
|
pub phase_id: Uuid,
|
|
pub external_id: &'a str,
|
|
pub title: &'a str,
|
|
pub assigned_agent_id: Option<Uuid>,
|
|
pub status: &'a str,
|
|
pub run_id: Option<Uuid>,
|
|
}
|
|
|
|
/// UPSERT a task keyed on (phase_id, external_id). Used by the
|
|
/// task-card parser (Slice 5) — same INT-XX marker across iterations
|
|
/// updates one row instead of creating duplicates.
|
|
pub async fn upsert_task(pool: &PgPool, t: UpsertTask<'_>) -> Result<Uuid, DbError> {
|
|
use sqlx::Row;
|
|
let is_completion = t.status == "complete";
|
|
let row = sqlx::query(
|
|
"INSERT INTO mission_tasks
|
|
(id, mission_id, phase_id, external_id, title,
|
|
assigned_agent_id, status, run_id, completed_at)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,
|
|
CASE WHEN $9 THEN now() ELSE NULL END)
|
|
ON CONFLICT (phase_id, external_id)
|
|
DO UPDATE SET
|
|
title = EXCLUDED.title,
|
|
assigned_agent_id = COALESCE(EXCLUDED.assigned_agent_id, mission_tasks.assigned_agent_id),
|
|
status = EXCLUDED.status,
|
|
run_id = COALESCE(EXCLUDED.run_id, mission_tasks.run_id),
|
|
updated_at = now(),
|
|
completed_at = CASE WHEN $9 AND mission_tasks.completed_at IS NULL
|
|
THEN now()
|
|
ELSE mission_tasks.completed_at END
|
|
RETURNING id",
|
|
)
|
|
.bind(Uuid::now_v7())
|
|
.bind(t.mission_id)
|
|
.bind(t.phase_id)
|
|
.bind(t.external_id)
|
|
.bind(t.title)
|
|
.bind(t.assigned_agent_id)
|
|
.bind(t.status)
|
|
.bind(t.run_id)
|
|
.bind(is_completion)
|
|
.fetch_one(pool)
|
|
.await?;
|
|
Ok(row.get("id"))
|
|
}
|
|
|
|
pub async fn tasks_for(pool: &PgPool, mission_id: Uuid) -> Result<Vec<MissionTask>, DbError> {
|
|
use sqlx::Row;
|
|
let rows = sqlx::query(
|
|
"SELECT id, mission_id, phase_id, external_id, title,
|
|
assigned_agent_id, status, run_id, artifact_paths,
|
|
created_at, updated_at, completed_at
|
|
FROM mission_tasks WHERE mission_id = $1
|
|
ORDER BY created_at ASC",
|
|
)
|
|
.bind(mission_id)
|
|
.fetch_all(pool)
|
|
.await?;
|
|
Ok(rows
|
|
.into_iter()
|
|
.map(|r| MissionTask {
|
|
id: r.get("id"),
|
|
mission_id: r.get("mission_id"),
|
|
phase_id: r.get("phase_id"),
|
|
external_id: r.get("external_id"),
|
|
title: r.get("title"),
|
|
assigned_agent_id: r.get("assigned_agent_id"),
|
|
status: r.get("status"),
|
|
run_id: r.get("run_id"),
|
|
artifact_paths: r.get("artifact_paths"),
|
|
created_at: r.get("created_at"),
|
|
updated_at: r.get("updated_at"),
|
|
completed_at: r.get("completed_at"),
|
|
})
|
|
.collect())
|
|
}
|
|
|
|
// ── Artifacts ────────────────────────────────────────────────────
|
|
|
|
#[derive(Debug, Clone)]
|
|
pub struct RegisterArtifact<'a> {
|
|
pub mission_id: Uuid,
|
|
pub phase_id: Option<Uuid>,
|
|
pub path: &'a str,
|
|
pub kind: &'a str,
|
|
pub mime: Option<&'a str>,
|
|
pub title: Option<&'a str>,
|
|
pub generated_by_run: Option<Uuid>,
|
|
pub render_pdf: bool,
|
|
/// Free-form facts about the artifact (diffstat, branch, gate verdict).
|
|
/// The column has existed since 0047 and was never written — an artifact
|
|
/// with no metadata is a path and a kind, which is not enough for a UI to
|
|
/// say anything useful about it.
|
|
pub metadata: Option<serde_json::Value>,
|
|
}
|
|
|
|
/// Register an artifact discovered on disk (or produced inline).
|
|
/// Idempotent on (mission_id, path).
|
|
pub async fn register_artifact(pool: &PgPool, a: RegisterArtifact<'_>) -> Result<Uuid, DbError> {
|
|
use sqlx::Row;
|
|
let render_status = if a.render_pdf { "pending" } else { "skip" };
|
|
let row = sqlx::query(
|
|
"INSERT INTO mission_artifacts
|
|
(id, mission_id, phase_id, path, kind, mime, title,
|
|
generated_by_run, render_pdf_status, metadata)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,COALESCE($10, '{}'::jsonb))
|
|
ON CONFLICT (mission_id, path) DO UPDATE
|
|
SET kind = EXCLUDED.kind,
|
|
mime = COALESCE(EXCLUDED.mime, mission_artifacts.mime),
|
|
title = COALESCE(EXCLUDED.title, mission_artifacts.title),
|
|
generated_by_run = COALESCE(EXCLUDED.generated_by_run,
|
|
mission_artifacts.generated_by_run),
|
|
metadata = COALESCE(EXCLUDED.metadata, mission_artifacts.metadata),
|
|
updated_at = now()
|
|
RETURNING id",
|
|
)
|
|
.bind(Uuid::now_v7())
|
|
.bind(a.mission_id)
|
|
.bind(a.phase_id)
|
|
.bind(a.path)
|
|
.bind(a.kind)
|
|
.bind(a.mime)
|
|
.bind(a.title)
|
|
.bind(a.generated_by_run)
|
|
.bind(render_status)
|
|
.bind(a.metadata.as_ref())
|
|
.fetch_one(pool)
|
|
.await?;
|
|
Ok(row.get("id"))
|
|
}
|
|
|
|
pub async fn artifacts_for(
|
|
pool: &PgPool,
|
|
mission_id: Uuid,
|
|
) -> Result<Vec<MissionArtifact>, DbError> {
|
|
use sqlx::Row;
|
|
let rows = sqlx::query(
|
|
"SELECT id, mission_id, phase_id, path, kind, mime,
|
|
generated_by_run, rendered_pdf_path, render_pdf_status,
|
|
render_pdf_error, title, metadata, created_at, updated_at
|
|
FROM mission_artifacts WHERE mission_id = $1
|
|
ORDER BY created_at DESC",
|
|
)
|
|
.bind(mission_id)
|
|
.fetch_all(pool)
|
|
.await?;
|
|
Ok(rows
|
|
.into_iter()
|
|
.map(|r| MissionArtifact {
|
|
id: r.get("id"),
|
|
mission_id: r.get("mission_id"),
|
|
phase_id: r.get("phase_id"),
|
|
path: r.get("path"),
|
|
kind: r.get("kind"),
|
|
mime: r.get("mime"),
|
|
generated_by_run: r.get("generated_by_run"),
|
|
rendered_pdf_path: r.get("rendered_pdf_path"),
|
|
render_pdf_status: r.get("render_pdf_status"),
|
|
render_pdf_error: r.get("render_pdf_error"),
|
|
title: r.get("title"),
|
|
metadata: r.get("metadata"),
|
|
created_at: r.get("created_at"),
|
|
updated_at: r.get("updated_at"),
|
|
})
|
|
.collect())
|
|
}
|
|
|
|
/// PDF renderer worker (Slice 6) picks a pending artifact off the
|
|
/// queue. Skips the FOR UPDATE dance until we see concurrent renderers
|
|
/// — one renderer per process is fine for the first cut.
|
|
pub async fn next_pdf_pending(pool: &PgPool, limit: i64) -> Result<Vec<MissionArtifact>, DbError> {
|
|
use sqlx::Row;
|
|
let rows = sqlx::query(
|
|
"SELECT id, mission_id, phase_id, path, kind, mime,
|
|
generated_by_run, rendered_pdf_path, render_pdf_status,
|
|
render_pdf_error, title, metadata, created_at, updated_at
|
|
FROM mission_artifacts
|
|
WHERE render_pdf_status = 'pending'
|
|
ORDER BY created_at ASC
|
|
LIMIT $1",
|
|
)
|
|
.bind(limit)
|
|
.fetch_all(pool)
|
|
.await?;
|
|
Ok(rows
|
|
.into_iter()
|
|
.map(|r| MissionArtifact {
|
|
id: r.get("id"),
|
|
mission_id: r.get("mission_id"),
|
|
phase_id: r.get("phase_id"),
|
|
path: r.get("path"),
|
|
kind: r.get("kind"),
|
|
mime: r.get("mime"),
|
|
generated_by_run: r.get("generated_by_run"),
|
|
rendered_pdf_path: r.get("rendered_pdf_path"),
|
|
render_pdf_status: r.get("render_pdf_status"),
|
|
render_pdf_error: r.get("render_pdf_error"),
|
|
title: r.get("title"),
|
|
metadata: r.get("metadata"),
|
|
created_at: r.get("created_at"),
|
|
updated_at: r.get("updated_at"),
|
|
})
|
|
.collect())
|
|
}
|
|
|
|
pub async fn set_pdf_result(
|
|
pool: &PgPool,
|
|
id: Uuid,
|
|
rendered_pdf_path: Option<&str>,
|
|
error: Option<&str>,
|
|
) -> Result<(), DbError> {
|
|
let status = if error.is_some() { "failed" } else { "done" };
|
|
sqlx::query(
|
|
"UPDATE mission_artifacts
|
|
SET rendered_pdf_path = COALESCE($2, rendered_pdf_path),
|
|
render_pdf_status = $3,
|
|
render_pdf_error = $4,
|
|
updated_at = now()
|
|
WHERE id = $1",
|
|
)
|
|
.bind(id)
|
|
.bind(rendered_pdf_path)
|
|
.bind(status)
|
|
.bind(error)
|
|
.execute(pool)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
// ── Benchmark snapshots (Slice 7) ────────────────────────────────
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct BenchmarkSnapshot {
|
|
pub id: Uuid,
|
|
pub mission_id: Uuid,
|
|
pub phase_id: Uuid,
|
|
pub iteration: i32,
|
|
pub before_metrics: Option<Value>,
|
|
pub after_metrics: Option<Value>,
|
|
pub delta: Option<Value>,
|
|
pub driver: Option<String>,
|
|
#[serde(with = "time::serde::rfc3339")]
|
|
pub created_at: OffsetDateTime,
|
|
}
|
|
|
|
#[derive(Debug, Clone)]
|
|
pub struct UpsertBenchmarkSnapshot<'a> {
|
|
pub mission_id: Uuid,
|
|
pub phase_id: Uuid,
|
|
pub iteration: i32,
|
|
pub before_metrics: Option<&'a Value>,
|
|
pub after_metrics: Option<&'a Value>,
|
|
pub delta: Option<&'a Value>,
|
|
pub driver: Option<&'a str>,
|
|
}
|
|
|
|
/// UPSERT a snapshot keyed on (phase_id, iteration). Baseline pass
|
|
/// uses iteration=0 with before_metrics only; each post-iteration
|
|
/// call updates the same row with after_metrics + delta so the pair
|
|
/// stays coherent for the canvas's side-by-side render.
|
|
pub async fn upsert_benchmark_snapshot(
|
|
pool: &PgPool,
|
|
s: UpsertBenchmarkSnapshot<'_>,
|
|
) -> Result<Uuid, DbError> {
|
|
use sqlx::Row;
|
|
let row = sqlx::query(
|
|
"INSERT INTO benchmark_snapshots
|
|
(id, mission_id, phase_id, iteration,
|
|
before_metrics, after_metrics, delta, driver)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
|
|
ON CONFLICT (phase_id, iteration) DO UPDATE SET
|
|
-- Preserve non-NULL prior values so a baseline call
|
|
-- doesn't wipe the after_metrics from a previous run.
|
|
before_metrics = COALESCE(EXCLUDED.before_metrics, benchmark_snapshots.before_metrics),
|
|
after_metrics = COALESCE(EXCLUDED.after_metrics, benchmark_snapshots.after_metrics),
|
|
delta = COALESCE(EXCLUDED.delta, benchmark_snapshots.delta),
|
|
driver = COALESCE(EXCLUDED.driver, benchmark_snapshots.driver)
|
|
RETURNING id",
|
|
)
|
|
.bind(Uuid::now_v7())
|
|
.bind(s.mission_id)
|
|
.bind(s.phase_id)
|
|
.bind(s.iteration)
|
|
.bind(s.before_metrics)
|
|
.bind(s.after_metrics)
|
|
.bind(s.delta)
|
|
.bind(s.driver)
|
|
.fetch_one(pool)
|
|
.await?;
|
|
Ok(row.get("id"))
|
|
}
|
|
|
|
pub async fn benchmark_snapshots_for(
|
|
pool: &PgPool,
|
|
mission_id: Uuid,
|
|
) -> Result<Vec<BenchmarkSnapshot>, DbError> {
|
|
use sqlx::Row;
|
|
let rows = sqlx::query(
|
|
"SELECT id, mission_id, phase_id, iteration,
|
|
before_metrics, after_metrics, delta, driver, created_at
|
|
FROM benchmark_snapshots
|
|
WHERE mission_id = $1
|
|
ORDER BY phase_id, iteration",
|
|
)
|
|
.bind(mission_id)
|
|
.fetch_all(pool)
|
|
.await?;
|
|
Ok(rows
|
|
.into_iter()
|
|
.map(|r| BenchmarkSnapshot {
|
|
id: r.get("id"),
|
|
mission_id: r.get("mission_id"),
|
|
phase_id: r.get("phase_id"),
|
|
iteration: r.get("iteration"),
|
|
before_metrics: r.get("before_metrics"),
|
|
after_metrics: r.get("after_metrics"),
|
|
delta: r.get("delta"),
|
|
driver: r.get("driver"),
|
|
created_at: r.get("created_at"),
|
|
})
|
|
.collect())
|
|
}
|
|
|
|
/// Phase progress for a set of missions, for the missions list cards.
|
|
/// Returns `(mission_id, total, done, running_phase_kind)`.
|
|
///
|
|
/// A status dot alone doesn't tell you where a mission actually is; this
|
|
/// is what lets a card say "Coding · 1/2" instead of just "running".
|
|
pub async fn phase_progress(
|
|
pool: &PgPool,
|
|
mission_ids: &[Uuid],
|
|
) -> Result<Vec<(Uuid, i64, i64, Option<String>)>, DbError> {
|
|
use sqlx::Row;
|
|
if mission_ids.is_empty() {
|
|
return Ok(Vec::new());
|
|
}
|
|
let rows = sqlx::query(
|
|
"SELECT mission_id,
|
|
count(*) AS total,
|
|
count(*) FILTER (WHERE status IN ('completed','skipped')) AS done,
|
|
(array_agg(kind ORDER BY order_idx)
|
|
FILTER (WHERE status = 'running'))[1] AS running_kind
|
|
FROM mission_phases
|
|
WHERE mission_id = ANY($1)
|
|
GROUP BY mission_id",
|
|
)
|
|
.bind(mission_ids)
|
|
.fetch_all(pool)
|
|
.await?;
|
|
Ok(rows
|
|
.into_iter()
|
|
.map(|r| {
|
|
(
|
|
r.get("mission_id"),
|
|
r.get("total"),
|
|
r.get("done"),
|
|
r.get("running_kind"),
|
|
)
|
|
})
|
|
.collect())
|
|
}
|