clawmates/crates/cm-db/tests/unreachable_phases.rs

//! A failed phase must not strand its mission at `running` forever.
//!
//! `start_pending_phases` launches a phase only when every lower-order phase is
//! `completed`, so once one fails the rest can never run. They sat `pending`,
//! and `close_finished_missions` requires no phase to be non-terminal — so the
//! mission never finished, and `mission_runtime`'s sweeper (which fires after a
//! terminal state) never reaped its container.
//!
//! Found by counting containers on gw-04, not by a test: one leaked runtime
//! container per failed multi-phase mission, accumulating for days. This is the
//! SQL that ends it, tested against a real database because the bug lived
//! entirely in the interaction between two queries' predicates.

use cm_domain::WorkspaceId;
use uuid::Uuid;

async fn workspace(pool: &sqlx::PgPool) -> WorkspaceId {
    let ws = cm_domain::Workspace {
        id: WorkspaceId::new(),
        name: "Unreachable".into(),
        plan: "team".into(),
    };
    cm_db::repo::workspaces::insert(pool, &ws).await.expect("workspace");
    ws.id
}

/// A running mission whose phase 0 failed and whose phases 1..n never started.
async fn stuck_mission(pool: &sqlx::PgPool, ws: WorkspaceId) -> Uuid {
    let id = Uuid::now_v7();
    sqlx::query(
        "INSERT INTO missions (id, workspace_id, title, template_kind, status, schedule, config)
         VALUES ($1, $2, 'stuck', 'research_and_code', 'running', '{}'::jsonb, '{}'::jsonb)",
    )
    .bind(id)
    .bind(ws.as_uuid())
    .execute(pool)
    .await
    .expect("mission");
    for (idx, status) in [(0, "failed"), (1, "pending"), (2, "pending")] {
        sqlx::query(
            "INSERT INTO mission_phases (id, mission_id, kind, order_idx, status, config)
             VALUES ($1, $2, 'coding', $3, $4, '{}'::jsonb)",
        )
        .bind(Uuid::now_v7())
        .bind(id)
        .bind(idx)
        .bind(status)
        .execute(pool)
        .await
        .expect("phase");
    }
    id
}

/// The sweep, as `phase_runner::skip_unreachable_phases` runs it.
async fn skip_unreachable(pool: &sqlx::PgPool) -> u64 {
    sqlx::query(
        "UPDATE mission_phases mp
            SET status = 'skipped', completed_at = now()
          WHERE mp.status = 'pending'
            AND EXISTS (SELECT 1 FROM missions m WHERE m.id = mp.mission_id AND m.status = 'running')
            AND EXISTS (
              SELECT 1 FROM mission_phases prior
               WHERE prior.mission_id = mp.mission_id
                 AND prior.order_idx < mp.order_idx
                 AND prior.status = 'failed'
            )",
    )
    .execute(pool)
    .await
    .expect("skip")
    .rows_affected()
}

async fn statuses(pool: &sqlx::PgPool, mission: Uuid) -> Vec<String> {
    sqlx::query_scalar("SELECT status FROM mission_phases WHERE mission_id = $1 ORDER BY order_idx")
        .bind(mission)
        .fetch_all(pool)
        .await
        .expect("statuses")
}

#[tokio::test]
async fn a_failed_phase_makes_the_later_ones_unreachable_not_pending_forever() {
    let pool = cm_testkit::test_pool().await;
    let ws = workspace(&pool).await;
    let m = stuck_mission(&pool, ws).await;

    assert_eq!(skip_unreachable(&pool).await, 2, "both later phases are unreachable");
    assert_eq!(
        statuses(&pool, m).await,
        vec!["failed", "skipped", "skipped"],
        "a phase that can never run must say so, or the mission never closes"
    );
    // Every phase is now terminal, which is what `close_finished_missions`
    // waits for — the container sweeper keys off the mission reaching that.
    let non_terminal: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM mission_phases
          WHERE mission_id = $1 AND status NOT IN ('completed','failed','skipped')",
    )
    .bind(m)
    .fetch_one(&pool)
    .await
    .unwrap();
    assert_eq!(non_terminal, 0);
}

/// A phase waiting AHEAD of the failure is untouched: order is what makes a
/// phase unreachable, and a failure later in the list says nothing about one
/// still queued before it.
#[tokio::test]
async fn a_phase_before_the_failure_is_left_alone() {
    let pool = cm_testkit::test_pool().await;
    let ws = workspace(&pool).await;
    let m = Uuid::now_v7();
    sqlx::query(
        "INSERT INTO missions (id, workspace_id, title, template_kind, status, schedule, config)
         VALUES ($1, $2, 'ordered', 'research_and_code', 'running', '{}'::jsonb, '{}'::jsonb)",
    )
    .bind(m)
    .bind(ws.as_uuid())
    .execute(&pool)
    .await
    .unwrap();
    for (idx, status) in [(0, "pending"), (1, "failed"), (2, "pending")] {
        sqlx::query(
            "INSERT INTO mission_phases (id, mission_id, kind, order_idx, status, config)
             VALUES ($1, $2, 'coding', $3, $4, '{}'::jsonb)",
        )
        .bind(Uuid::now_v7())
        .bind(m)
        .bind(idx)
        .bind(status)
        .execute(&pool)
        .await
        .unwrap();
    }
    skip_unreachable(&pool).await;
    assert_eq!(statuses(&pool, m).await, vec!["pending", "failed", "skipped"]);
}

/// A mission that is not running is not swept: a draft's phases are pending by
/// definition and must not be skipped out from under it.
#[tokio::test]
async fn only_a_running_missions_phases_are_skipped() {
    let pool = cm_testkit::test_pool().await;
    let ws = workspace(&pool).await;
    let m = stuck_mission(&pool, ws).await;
    sqlx::query("UPDATE missions SET status = 'draft' WHERE id = $1")
        .bind(m)
        .execute(&pool)
        .await
        .unwrap();
    assert_eq!(skip_unreachable(&pool).await, 0);
    assert_eq!(statuses(&pool, m).await, vec!["failed", "pending", "pending"]);
}