slice 1: missions data model + migration

Introduce the unified `missions` tier that will replace the current
research_topics + loops split. This slice ships the data model +
backfill + skeleton REST surface; the old wizards keep working in
parallel until Slice 9's big-bang cutover.

Migration 0047 adds:
  - missions              (top-level workflow: template_kind + team +
                           schedule + status + config)
  - mission_phases        (ordered {research|coding|benchmark|
                           security_scan} phases per mission)
  - mission_tasks         (typed units of work, e.g. INT-XX cards,
                           UPSERT-keyed on (phase_id, external_id))
  - mission_artifacts     (MD/PDF/benchmark/security/diff files with
                           a pending queue for the PDF renderer worker)
  - benchmark_snapshots   (before/after pairs per iteration)

Backfill copies existing research_topics + loops rows into the new
tables as one-shot missions with the appropriate template_kind, so
Slice 2's UI can render the full history immediately.

New Rust surface:
  - cm_domain: MissionId, MissionPhaseId, MissionTaskId, MissionArtifactId
  - cm_db::repo::missions: Mission/MissionPhase/MissionTask/
    MissionArtifact structs + insert (txn-wrapped)/get/list/set_status/
    phases_for/set_phase_status/upsert_task/tasks_for/register_artifact/
    artifacts_for/next_pdf_pending/set_pdf_result
  - cm_api::routes::missions: skeleton list/create/get/set_status
    routes registered at /api/missions/*

Follow-up slices layer richer behavior (template dispatch, phase
execution, task parsing, artifact rendering) on this foundation.

Co-Authored-By: Claude Opus 4.7 <[email protected]>

crates/cm-api/src/lib.rs

@@ -401,6 +401,17 @@ pub fn router(state: AppState) -> Router {
             axum::routing::patch(routes::orgs::rename_org),
         )
         .route("/api/orgs/{id}/run", post(routes::orgs::run_org))
+        // Missions — Slice 1 skeleton. Runs in parallel with the
+        // legacy research/loops routes until Slice 9's cutover.
+        .route(
+            "/api/missions",
+            get(routes::missions::list).post(routes::missions::create),
+        )
+        .route("/api/missions/{id}", get(routes::missions::get))
+        .route(
+            "/api/missions/{id}/status",
+            axum::routing::patch(routes::missions::set_status),
+        )
         .route(
             "/api/research",
             get(routes::research::list_topics).post(routes::research::create_topic),

crates/cm-api/src/routes/missions.rs

@@ -0,0 +1,165 @@
+//! `/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::{
+    Json,
+    extract::{Path, Query, State},
+};
+use cm_db::repo::missions::{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>,
+}
+
+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>,
+}
+
+// ── 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);
+    }
+    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,
+        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?;
+    Ok(Json(MissionDetail {
+        mission,
+        phases,
+        tasks,
+        artifacts,
+    }))
+}
+
+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);
+    }
+    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))
+}

crates/cm-api/src/routes/mod.rs

@@ -13,6 +13,7 @@ pub mod gateway;
 pub mod health;
 pub mod identity;
 pub mod loops;
+pub mod missions;
 pub mod nodes;
 pub mod oauth;
 pub mod orgs;

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

@@ -0,0 +1,545 @@
+//! 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,
+    #[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,
+    #[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,
+    pub phases: Vec<NewMissionPhase>,
+}
+
+#[derive(Debug, Clone)]
+pub struct NewMissionPhase {
+    pub kind: String,
+    pub order_idx: i32,
+    pub config: Value,
+}
+
+// ── 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)
+         VALUES ($1,$2,$3,$4,$5,$6,$7,$8,'draft',$9,$10)",
+    )
+    .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)
+    .execute(&mut *tx)
+    .await?;
+
+    for p in &m.phases {
+        sqlx::query(
+            "INSERT INTO mission_phases
+               (id, mission_id, kind, order_idx, status, config)
+             VALUES ($1,$2,$3,$4,'pending',$5)",
+        )
+        .bind(Uuid::now_v7())
+        .bind(mission_id)
+        .bind(&p.kind)
+        .bind(p.order_idx)
+        .bind(&p.config)
+        .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, 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"),
+        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, 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"),
+            created_at: r.get("created_at"),
+            updated_at: r.get("updated_at"),
+            completed_at: r.get("completed_at"),
+        })
+        .collect())
+}
+
+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,
+                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"),
+            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,
+}
+
+/// 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)
+         VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
+         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),
+                 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)
+    .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(())
+}

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

@@ -10,6 +10,7 @@ pub mod fleet_beszel;
 pub mod fleet_tailscale;
 pub mod loops;
 pub mod messages;
+pub mod missions;
 pub mod node_metrics;
 pub mod node_rules;
 pub mod node_tools;

crates/cm-domain/src/ids.rs

@@ -73,3 +73,24 @@ define_id!(
     /// A connected fleet node (a user's local-hardware host running the daemon).
     NodeId
 );
+define_id!(
+    /// A user-driven workflow composed of one or more mission_phases.
+    /// Unifies research_topics + loops behind a single tier (Slice 1
+    /// of the missions unification).
+    MissionId
+);
+define_id!(
+    /// A single phase within a mission (research | coding | benchmark |
+    /// security_scan). Ordered via `order_idx` on the row.
+    MissionPhaseId
+);
+define_id!(
+    /// A typed unit of work within a phase — e.g. an INT-XX item, a
+    /// CVE finding, a research outcome iteration.
+    MissionTaskId
+);
+define_id!(
+    /// A file artifact produced by a mission phase — MD, PDF,
+    /// benchmark result, security report, code diff.
+    MissionArtifactId
+);

crates/cm-domain/src/lib.rs

@@ -18,6 +18,9 @@ pub use chat::{
 };
 pub use entities::{Agent, AgentStatus, FileDrive, FileNode, User, Workspace};
 pub use gated::GatedCategory;
-pub use ids::{AgentId, MessageId, NodeId, SessionId, UserId, WorkspaceId};
+pub use ids::{
+    AgentId, MessageId, MissionArtifactId, MissionId, MissionPhaseId, MissionTaskId, NodeId,
+    SessionId, UserId, WorkspaceId,
+};
 pub use role::Role;
 pub use session_key::{SessionKey, SessionKeyError};

migrations/0047_missions.sql

@@ -0,0 +1,292 @@
+-- Slice 1 of the missions unification (see design notes).
+--
+-- A `mission` is a single top-level user-driven workflow. It replaces
+-- the current two-headed split between `research_topics` and `loops`:
+-- both of those become special cases of a mission composed of phases.
+--
+-- This migration ONLY adds the new tables + backfills existing rows.
+-- Old tables (research_topics, loops, research_topic_agents,
+-- research_outcomes, ...) stay live and writable until Slice 9's
+-- big-bang cutover deletes them. Reads route through the compat
+-- shims added in Slice 2; the old wizards keep working unchanged.
+--
+-- Discriminators — kept as TEXT (not enums) so new templates + phase
+-- kinds ship as PRs without schema migrations.
+--   missions.template_kind ∈
+--     {research_only, research_and_code, security_hardening,
+--      refactor, benchmark, custom}
+--   mission_phases.kind ∈
+--     {research, coding, benchmark, security_scan}
+--   mission_tasks.status ∈
+--     {created, working, validating, complete, failed}
+--   mission_artifacts.kind ∈
+--     {md, pdf, benchmark_result, security_report, code_diff, index}
+
+CREATE TABLE missions (
+    id                UUID PRIMARY KEY,
+    workspace_id      UUID NOT NULL,
+    title             TEXT NOT NULL,
+    template_kind     TEXT NOT NULL,
+    -- team_id is nullable so a mission can start without a bound team
+    -- (auto-provision materializes one during phase startup).
+    team_id           UUID REFERENCES teams(id) ON DELETE SET NULL,
+    -- team_template_id captures template lineage — level-up promotions
+    -- diff against the template version the team was minted from.
+    team_template_id  UUID,
+    repo_id           UUID REFERENCES repos(id) ON DELETE SET NULL,
+    -- schedule JSONB carries the trigger config (cron | one_shot |
+    -- on_event). Kept as JSONB so we can grow the schedule surface
+    -- without a table alter each time.
+    schedule          JSONB NOT NULL DEFAULT '{"kind":"one_shot"}'::jsonb,
+    -- Mission-level status. Rolls up phase statuses per lifecycle rules
+    -- enforced in cm-api::routes::missions.
+    -- draft → running → (completed | failed | cancelled)
+    status            TEXT NOT NULL DEFAULT 'draft',
+    -- Human-facing description + optional target subject. Both are
+    -- shown in the canvas overview and passed to the LLM as context.
+    description       TEXT,
+    -- Free-form config for the specific workflow template (task
+    -- template overrides, benchmark commands, LLM overrides for PDF
+    -- rendering, etc.). Every template contributes its own keys.
+    config            JSONB NOT NULL DEFAULT '{}'::jsonb,
+    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    completed_at      TIMESTAMPTZ
+);
+CREATE INDEX missions_workspace_idx
+    ON missions (workspace_id, created_at DESC);
+CREATE INDEX missions_team_idx
+    ON missions (team_id) WHERE team_id IS NOT NULL;
+CREATE INDEX missions_status_idx
+    ON missions (status, updated_at DESC);
+
+-- A phase is one segment of a mission's plan. Ordering by order_idx.
+-- Multiple phases of the same kind are allowed (e.g. two coding
+-- passes bracketing a benchmark phase).
+CREATE TABLE mission_phases (
+    id                UUID PRIMARY KEY,
+    mission_id        UUID NOT NULL REFERENCES missions(id) ON DELETE CASCADE,
+    kind              TEXT NOT NULL,
+    order_idx         INT  NOT NULL,
+    status            TEXT NOT NULL DEFAULT 'pending',
+    -- Phase-scoped config (e.g. commit_policy, loop bounds, benchmark
+    -- cmd) merged over the template's phase spec at run time.
+    config            JSONB NOT NULL DEFAULT '{}'::jsonb,
+    started_at        TIMESTAMPTZ,
+    completed_at      TIMESTAMPTZ,
+    UNIQUE (mission_id, order_idx)
+);
+CREATE INDEX mission_phases_mission_idx
+    ON mission_phases (mission_id, order_idx);
+CREATE INDEX mission_phases_status_idx
+    ON mission_phases (status) WHERE status IN ('pending', 'running');
+
+-- A task is one addressable unit inside a phase. For coding phases
+-- that consume an integrations artifact, tasks correspond 1:1 with
+-- INT-XX items parsed from run output (Slice 5).
+-- Research phases materialize one task per "outcome iteration".
+-- Security phases materialize one task per finding.
+CREATE TABLE mission_tasks (
+    id                UUID PRIMARY KEY,
+    mission_id        UUID NOT NULL REFERENCES missions(id) ON DELETE CASCADE,
+    phase_id          UUID NOT NULL REFERENCES mission_phases(id) ON DELETE CASCADE,
+    -- external_id: e.g. "INT-05", "CVE-2024-1234", "OUTCOME-v3".
+    -- Nullable so ad-hoc tasks work without a parseable marker.
+    external_id       TEXT,
+    title             TEXT NOT NULL,
+    -- Assigned agent (nullable — some tasks are team-wide).
+    assigned_agent_id UUID REFERENCES agents(id) ON DELETE SET NULL,
+    status            TEXT NOT NULL DEFAULT 'created',
+    -- The topology run that produced this task's most recent state
+    -- update. Follows the run's lifecycle for observability.
+    run_id            UUID REFERENCES topology_runs(id) ON DELETE SET NULL,
+    -- Paths to the artifacts this task produced (relative to the
+    -- mission's artifact root). One task can have many artifacts;
+    -- kept as an array for cheap lookup without a join table.
+    artifact_paths    TEXT[] NOT NULL DEFAULT '{}',
+    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    completed_at      TIMESTAMPTZ
+);
+CREATE INDEX mission_tasks_mission_idx
+    ON mission_tasks (mission_id, created_at);
+CREATE INDEX mission_tasks_phase_idx
+    ON mission_tasks (phase_id, status);
+CREATE INDEX mission_tasks_agent_idx
+    ON mission_tasks (assigned_agent_id) WHERE assigned_agent_id IS NOT NULL;
+-- Uniqueness on (phase_id, external_id) so the task-card parser (Slice
+-- 5) can UPSERT on marker match without duplicating rows.
+CREATE UNIQUE INDEX mission_tasks_external_uniq
+    ON mission_tasks (phase_id, external_id)
+    WHERE external_id IS NOT NULL;
+
+-- Artifacts land on the filesystem under a per-mission root
+-- (see cm-api::missions::artifact_root). We index them here for
+-- typed discovery from the UI without walking the filesystem.
+CREATE TABLE mission_artifacts (
+    id                    UUID PRIMARY KEY,
+    mission_id            UUID NOT NULL REFERENCES missions(id) ON DELETE CASCADE,
+    phase_id              UUID REFERENCES mission_phases(id) ON DELETE SET NULL,
+    -- Repo-relative path from the mission root, e.g.
+    -- "research/v3/spec.md" or "benchmarks/before.json".
+    path                  TEXT NOT NULL,
+    kind                  TEXT NOT NULL,
+    mime                  TEXT,
+    -- Optional back-pointer to the run that produced this artifact.
+    generated_by_run      UUID REFERENCES topology_runs(id) ON DELETE SET NULL,
+    -- Sidecar for the PDF renderer (Slice 6). NULL until rendered.
+    rendered_pdf_path     TEXT,
+    render_pdf_status     TEXT NOT NULL DEFAULT 'skip',
+    -- ↑ skip | pending | rendering | done | failed
+    render_pdf_error      TEXT,
+    -- Human-shown title; falls back to filename when NULL.
+    title                 TEXT,
+    -- Free-form metadata (word count, sha256, source model, etc.).
+    metadata              JSONB NOT NULL DEFAULT '{}'::jsonb,
+    created_at            TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at            TIMESTAMPTZ NOT NULL DEFAULT now(),
+    UNIQUE (mission_id, path)
+);
+CREATE INDEX mission_artifacts_mission_idx
+    ON mission_artifacts (mission_id, created_at DESC);
+-- PDF renderer picks up its queue via this partial index.
+CREATE INDEX mission_artifacts_pdf_queue_idx
+    ON mission_artifacts (created_at)
+    WHERE render_pdf_status = 'pending';
+
+-- Benchmark snapshots — one row per benchmark phase per iteration.
+-- Stored as JSONB pairs so the shape can evolve per template
+-- (criterion, cargo bench, k6, wrk, custom scripts) without alters.
+CREATE TABLE benchmark_snapshots (
+    id                UUID PRIMARY KEY,
+    mission_id        UUID NOT NULL REFERENCES missions(id) ON DELETE CASCADE,
+    phase_id          UUID NOT NULL REFERENCES mission_phases(id) ON DELETE CASCADE,
+    -- iteration = 0 for the pre-coding baseline; ≥1 for post-iteration
+    -- snapshots. Uniqueness enforced so re-runs overwrite in place.
+    iteration         INT  NOT NULL,
+    before_metrics    JSONB,
+    after_metrics     JSONB,
+    delta             JSONB,
+    -- Optional descriptor of the benchmark command / driver.
+    driver            TEXT,
+    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    UNIQUE (phase_id, iteration)
+);
+CREATE INDEX benchmark_snapshots_mission_idx
+    ON benchmark_snapshots (mission_id, created_at DESC);
+
+-- ── Backfill ─────────────────────────────────────────────────────
+--
+-- Every existing research_topic becomes a mission with a single
+-- research phase. Every existing loop with source_research_topic_id
+-- becomes a research_and_code mission with two phases pointing at the
+-- same underlying topic + loop rows through the compat shim.
+--
+-- Standalone loops (no source topic) become refactor missions with a
+-- single coding phase.
+--
+-- This is a one-shot copy — old rows are NOT deleted (Slice 9). The
+-- compat shims read from `missions` first, fall through to the
+-- legacy tables when nothing landed, so both surfaces stay coherent
+-- during the transition.
+
+-- 1. Research-only backfill.
+INSERT INTO missions (
+    id, workspace_id, title, template_kind, team_id, repo_id,
+    schedule, status, description, config, created_at, updated_at
+)
+SELECT
+    id,
+    workspace_id,
+    title,
+    'research_only',
+    team_id,
+    repo_id,
+    '{"kind":"one_shot"}'::jsonb,
+    CASE
+        WHEN status IN ('publishing','complete','archived') THEN 'completed'
+        WHEN status = 'processing'                          THEN 'running'
+        WHEN status = 'failed'                              THEN 'failed'
+        ELSE 'draft'
+    END,
+    description,
+    jsonb_build_object(
+        'legacy_topic_id', id::text,
+        'outcome_kind',    outcome_kind,
+        'topology_kind',   topology_kind
+    ),
+    created_at,
+    updated_at
+FROM research_topics
+WHERE NOT EXISTS (SELECT 1 FROM missions m WHERE m.id = research_topics.id);
+
+INSERT INTO mission_phases (id, mission_id, kind, order_idx, status, config)
+SELECT
+    gen_random_uuid(),
+    id,
+    'research',
+    0,
+    CASE
+        WHEN status IN ('publishing','complete','archived') THEN 'completed'
+        WHEN status = 'processing'                          THEN 'running'
+        WHEN status = 'failed'                              THEN 'failed'
+        ELSE 'pending'
+    END,
+    jsonb_build_object('legacy_topic_id', id::text)
+FROM research_topics
+WHERE NOT EXISTS (
+    SELECT 1 FROM mission_phases mp
+    WHERE mp.mission_id = research_topics.id AND mp.kind = 'research'
+);
+
+-- 2. Loop backfill — one mission per loop, template_kind chosen by
+-- whether it's paired to a research topic.
+INSERT INTO missions (
+    id, workspace_id, title, template_kind, team_id, repo_id,
+    schedule, status, description, config, created_at, updated_at
+)
+SELECT
+    l.id,
+    l.workspace_id,
+    l.title,
+    CASE
+        WHEN l.source_research_topic_id IS NOT NULL THEN 'research_and_code'
+        ELSE 'refactor'
+    END,
+    l.team_id,
+    NULL,
+    jsonb_build_object(
+        'kind',     CASE WHEN l.enabled THEN 'cron' ELSE 'one_shot' END,
+        'triggers', l.triggers
+    ),
+    'running',
+    l.description,
+    jsonb_build_object(
+        'legacy_loop_id',        l.id::text,
+        'source_research_topic', l.source_research_topic_id::text,
+        'loop_kind',             l.kind
+    ),
+    l.created_at,
+    l.updated_at
+FROM loops l
+WHERE NOT EXISTS (SELECT 1 FROM missions m WHERE m.id = l.id);
+
+-- 2a. Research-then-code loops get two phases (research phase already
+-- exists as the paired topic mission; we add a coding phase to this
+-- mission that points at the same loop_id via config).
+INSERT INTO mission_phases (id, mission_id, kind, order_idx, status, config)
+SELECT
+    gen_random_uuid(),
+    l.id,
+    'coding',
+    0,
+    'pending',
+    jsonb_build_object(
+        'legacy_loop_id',              l.id::text,
+        'source_research_topic_id',    l.source_research_topic_id::text
+    )
+FROM loops l
+WHERE NOT EXISTS (
+    SELECT 1 FROM mission_phases mp
+    WHERE mp.mission_id = l.id AND mp.kind = 'coding'
+);