clawmates/crates/cm-api/tests/phase_conditions.rs

//! Coverage for goal conditions and phase iteration (migration 0061).
//!
//! These tests exercise the SQL directly rather than the sweep loop, because
//! the part that is easy to get wrong is the *iteration scoping*: "are this
//! phase's runs all finished?" must ask about the CURRENT pass. Without that,
//! pass 1's completed rows satisfy pass 2 the instant it is enqueued and the
//! phase completes without doing any work.
//!
//! What this locks in:
//!   * A phase with no `done_when` still goes running -> completed on terminal
//!     runs (the regression guard: existing missions are unaffected).
//!   * A phase with `done_when` goes running -> evaluating instead.
//!   * A failed run fails the phase outright, condition or not.
//!   * Pass 2 is not satisfied by pass 1's completed runs.
//!   * `mission_phase_evaluations` is unique per (phase, iteration) and
//!     upserts.

use cm_db::repo::workspaces;
use cm_domain::{Workspace, WorkspaceId};
use sqlx::Row;
use uuid::Uuid;

async fn seed_workspace(pool: &sqlx::PgPool) -> WorkspaceId {
    let ws = Workspace {
        id: WorkspaceId::new(),
        name: "Phase Conditions Test".into(),
        plan: "team".into(),
    };
    workspaces::insert(pool, &ws).await.unwrap();
    ws.id
}

async fn seed_mission(pool: &sqlx::PgPool, ws: WorkspaceId) -> Uuid {
    let id = Uuid::now_v7();
    sqlx::query(
        "INSERT INTO missions (id, workspace_id, title, template_kind, status)
         VALUES ($1, $2, 'test mission', 'research_only', 'running')",
    )
    .bind(id)
    .bind(ws.as_uuid())
    .execute(pool)
    .await
    .unwrap();
    id
}

/// A phase in `running`, optionally carrying a completion condition.
async fn seed_phase(
    pool: &sqlx::PgPool,
    mission_id: Uuid,
    done_when: Option<&str>,
    max_iterations: i32,
    iteration: i32,
) -> Uuid {
    let id = Uuid::now_v7();
    sqlx::query(
        "INSERT INTO mission_phases
             (id, mission_id, kind, order_idx, status, done_when, max_iterations, iteration)
         VALUES ($1, $2, 'research', 0, 'running', $3, $4, $5)",
    )
    .bind(id)
    .bind(mission_id)
    .bind(done_when)
    .bind(max_iterations)
    .bind(iteration)
    .execute(pool)
    .await
    .unwrap();
    id
}

async fn seed_run(
    pool: &sqlx::PgPool,
    ws: WorkspaceId,
    mission_id: Uuid,
    phase_id: Uuid,
    status: &str,
    iteration: i32,
) {
    sqlx::query(
        "INSERT INTO topology_runs
             (id, workspace_id, task, kind, status, tier, mission_id, mission_phase_id, iteration)
         VALUES ($1, $2, 'task', 'run', $3, 'team', $4, $5, $6)",
    )
    .bind(Uuid::now_v7())
    .bind(ws.as_uuid())
    .bind(status)
    .bind(mission_id)
    .bind(phase_id)
    .bind(iteration)
    .execute(pool)
    .await
    .unwrap();
}

/// The exact statement `phase_runner::close_finished_phases` runs.
async fn close_finished_phases(pool: &sqlx::PgPool) {
    sqlx::query(
        "UPDATE mission_phases mp
            SET status =
              CASE
                WHEN EXISTS (
                  SELECT 1 FROM topology_runs r
                  WHERE r.mission_phase_id = mp.id
                    AND r.iteration = mp.iteration
                    AND r.status = 'failed'
                ) THEN 'failed'
                WHEN mp.done_when IS NOT NULL AND btrim(mp.done_when) <> '' THEN 'evaluating'
                ELSE 'completed'
              END,
              completed_at =
                CASE
                  WHEN mp.done_when IS NOT NULL AND btrim(mp.done_when) <> ''
                       AND NOT EXISTS (
                         SELECT 1 FROM topology_runs r
                         WHERE r.mission_phase_id = mp.id
                           AND r.iteration = mp.iteration
                           AND r.status = 'failed'
                       )
                  THEN NULL ELSE now()
                END
          WHERE mp.status = 'running'
            AND EXISTS (
              SELECT 1 FROM topology_runs r
              WHERE r.mission_phase_id = mp.id AND r.iteration = mp.iteration
            )
            AND NOT EXISTS (
              SELECT 1 FROM topology_runs r
              WHERE r.mission_phase_id = mp.id
                AND r.iteration = mp.iteration
                AND r.status NOT IN ('completed', 'failed', 'cancelled')
            )",
    )
    .execute(pool)
    .await
    .unwrap();
}

async fn phase_status(pool: &sqlx::PgPool, phase_id: Uuid) -> String {
    sqlx::query("SELECT status FROM mission_phases WHERE id = $1")
        .bind(phase_id)
        .fetch_one(pool)
        .await
        .unwrap()
        .get::<String, _>("status")
}

/// The regression guard. A mission that never opts into a condition must
/// behave exactly as it did before conditions existed.
#[tokio::test]
async fn phase_without_condition_completes_as_before() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let mission = seed_mission(&pool, ws).await;
    let phase = seed_phase(&pool, mission, None, 1, 0).await;
    seed_run(&pool, ws, mission, phase, "completed", 0).await;

    close_finished_phases(&pool).await;
    assert_eq!(phase_status(&pool, phase).await, "completed");
}

#[tokio::test]
async fn phase_with_condition_goes_to_evaluating() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let mission = seed_mission(&pool, ws).await;
    let phase = seed_phase(&pool, mission, Some("a brief exists"), 3, 0).await;
    seed_run(&pool, ws, mission, phase, "completed", 0).await;

    close_finished_phases(&pool).await;
    assert_eq!(
        phase_status(&pool, phase).await,
        "evaluating",
        "a phase with a condition must be judged before it can complete"
    );
    // completed_at must stay NULL while the phase is still being judged.
    let completed_at: Option<time::OffsetDateTime> =
        sqlx::query("SELECT completed_at FROM mission_phases WHERE id = $1")
            .bind(phase)
            .fetch_one(&pool)
            .await
            .unwrap()
            .get("completed_at");
    assert!(completed_at.is_none(), "not finished, so not timestamped");
}

/// A blank condition is not a condition — otherwise a UI that sends "" would
/// silently park every phase in `evaluating` forever.
#[tokio::test]
async fn blank_condition_is_treated_as_none() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let mission = seed_mission(&pool, ws).await;
    let phase = seed_phase(&pool, mission, Some("   "), 3, 0).await;
    seed_run(&pool, ws, mission, phase, "completed", 0).await;

    close_finished_phases(&pool).await;
    assert_eq!(phase_status(&pool, phase).await, "completed");
}

#[tokio::test]
async fn failed_run_fails_the_phase_even_with_a_condition() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let mission = seed_mission(&pool, ws).await;
    let phase = seed_phase(&pool, mission, Some("a brief exists"), 3, 0).await;
    seed_run(&pool, ws, mission, phase, "failed", 0).await;

    close_finished_phases(&pool).await;
    assert_eq!(
        phase_status(&pool, phase).await,
        "failed",
        "there is nothing to evaluate when the work itself failed"
    );
}

/// The subtle one. On pass 2 the phase has `iteration = 1`, but pass 1's
/// completed run is still in the table. Without scoping the check to the
/// current iteration, that stale row satisfies "all runs finished" and the
/// phase completes having done no work on this pass.
#[tokio::test]
async fn second_pass_is_not_satisfied_by_first_pass_runs() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let mission = seed_mission(&pool, ws).await;
    // Phase is on pass 2 (iteration=1) and running.
    let phase = seed_phase(&pool, mission, Some("a brief exists"), 3, 1).await;
    // Pass 1 left a completed run behind.
    seed_run(&pool, ws, mission, phase, "completed", 0).await;
    // Pass 2's run is still queued.
    seed_run(&pool, ws, mission, phase, "queued", 1).await;

    close_finished_phases(&pool).await;
    assert_eq!(
        phase_status(&pool, phase).await,
        "running",
        "pass 1's completed run must not close out pass 2"
    );

    // Finish pass 2 for real.
    sqlx::query(
        "UPDATE topology_runs SET status = 'completed'
          WHERE mission_phase_id = $1 AND iteration = 1",
    )
    .bind(phase)
    .execute(&pool)
    .await
    .unwrap();
    close_finished_phases(&pool).await;
    assert_eq!(phase_status(&pool, phase).await, "evaluating");
}

/// A phase whose current pass has enqueued nothing yet must not be closed by
/// an earlier pass's rows either.
#[tokio::test]
async fn phase_with_no_runs_this_pass_stays_running() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let mission = seed_mission(&pool, ws).await;
    let phase = seed_phase(&pool, mission, None, 3, 1).await;
    seed_run(&pool, ws, mission, phase, "completed", 0).await;

    close_finished_phases(&pool).await;
    assert_eq!(phase_status(&pool, phase).await, "running");
}

#[tokio::test]
async fn evaluations_are_unique_per_iteration_and_upsert() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let mission = seed_mission(&pool, ws).await;
    let phase = seed_phase(&pool, mission, Some("done"), 3, 0).await;

    let first = cm_api::evaluator::Verdict {
        met: false,
        reason: "no brief yet".into(),
        model: "runtime:coordinator".into(),
        error: None,
    };
    cm_api::evaluator::record(&pool, mission, phase, 0, &first)
        .await
        .unwrap();
    // Same iteration again — upsert, not a duplicate row or a constraint error.
    let second = cm_api::evaluator::Verdict {
        met: true,
        reason: "brief written".into(),
        model: "runtime:coordinator".into(),
        error: None,
    };
    cm_api::evaluator::record(&pool, mission, phase, 0, &second)
        .await
        .unwrap();

    let count: i64 = sqlx::query("SELECT count(*) AS n FROM mission_phase_evaluations WHERE phase_id = $1")
        .bind(phase)
        .fetch_one(&pool)
        .await
        .unwrap()
        .get("n");
    assert_eq!(count, 1, "one row per (phase, iteration)");

    let latest = cm_api::evaluator::latest(&pool, phase).await.unwrap();
    assert_eq!(latest, Some((0, true, "brief written".into())));
}

/// `latest` must return the newest pass, which is what feeds guidance into the
/// next attempt.
#[tokio::test]
async fn latest_returns_the_most_recent_iteration() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let mission = seed_mission(&pool, ws).await;
    let phase = seed_phase(&pool, mission, Some("done"), 3, 0).await;

    for (i, reason) in [(0, "first"), (1, "second"), (2, "third")] {
        cm_api::evaluator::record(
            &pool,
            mission,
            phase,
            i,
            &cm_api::evaluator::Verdict {
                met: false,
                reason: reason.into(),
                model: "m".into(),
                error: None,
            },
        )
        .await
        .unwrap();
    }
    let latest = cm_api::evaluator::latest(&pool, phase).await.unwrap();
    assert_eq!(latest, Some((2, false, "third".into())));
}