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

//! The durable topology-job lifecycle at the persistence layer: enqueue → claim
//! (CAS) → checkpoint → complete, plus the stale-run resume sweep. This is the
//! foundation that lets long-horizon topology runs survive worker restarts.

use cm_db::repo::{topology_runs, workspaces};
use cm_domain::{Workspace, WorkspaceId};
use serde_json::json;
use uuid::Uuid;

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

#[tokio::test]
async fn durable_run_lifecycle_enqueue_claim_checkpoint_complete() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;

    let id = Uuid::now_v7();
    let graph = json!({"kind": "pipeline", "nodes": [{"id":"n1","role":"drafter"}], "edges": []});
    topology_runs::enqueue_run(&pool, id, ws, "write a haiku", &graph)
        .await
        .unwrap();

    // Status starts queued, no result yet.
    let st = topology_runs::status(&pool, id, ws).await.unwrap();
    assert_eq!(st.status, "queued");
    assert_eq!(st.kind, "run");
    assert!(st.result.is_none());

    // Claim flips it to running and returns the job + its graph.
    let claimed = topology_runs::claim_next_queued(&pool)
        .await
        .unwrap()
        .expect("a queued job to claim");
    assert_eq!(claimed.id, id);
    assert!(claimed.graph.is_some());
    assert!(claimed.checkpoint.is_none());

    // A second claim finds nothing (the job is no longer queued).
    assert!(topology_runs::claim_next_queued(&pool).await.unwrap().is_none());

    // Checkpoint mid-run progress.
    let progress = json!({"completed": 1, "outputs": ["draft"], "records": [], "totals": {}});
    topology_runs::checkpoint(&pool, id, &progress, 1).await.unwrap();
    let st = topology_runs::status(&pool, id, ws).await.unwrap();
    assert_eq!(st.status, "running");
    assert_eq!(st.last_event_id, 1);
    assert!(st.checkpoint.is_some());

    // Complete with a result blob.
    let result = json!({"final_output": "a haiku", "totals": {"turns": 1}});
    topology_runs::complete(&pool, id, &result).await.unwrap();
    let st = topology_runs::status(&pool, id, ws).await.unwrap();
    assert_eq!(st.status, "completed");
    assert_eq!(st.result.unwrap()["final_output"], "a haiku");
}

#[tokio::test]
async fn stale_running_jobs_are_requeued_for_resume() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;

    let id = Uuid::now_v7();
    let graph = json!({"kind": "pipeline", "nodes": [{"id":"n1","role":"drafter"}], "edges": []});
    topology_runs::enqueue_run(&pool, id, ws, "task", &graph)
        .await
        .unwrap();
    // Claim it → running.
    topology_runs::claim_next_queued(&pool).await.unwrap().unwrap();

    // Not stale yet (just claimed) → sweep is a no-op.
    assert_eq!(topology_runs::requeue_stale(&pool, 60.0).await.unwrap(), 0);

    // With a zero threshold the running job counts as stale and is requeued,
    // so the next claim picks it up again (resume from checkpoint).
    assert_eq!(topology_runs::requeue_stale(&pool, 0.0).await.unwrap(), 1);
    let st = topology_runs::status(&pool, id, ws).await.unwrap();
    assert_eq!(st.status, "queued");
    assert!(topology_runs::claim_next_queued(&pool).await.unwrap().is_some());
}

#[tokio::test]
async fn cancel_transitions_only_active_runs() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;

    let id = Uuid::now_v7();
    let graph = json!({"kind": "pipeline", "nodes": [{"id":"n1","role":"drafter"}], "edges": []});
    topology_runs::enqueue_run(&pool, id, ws, "task", &graph)
        .await
        .unwrap();

    // A queued run cancels; the worker sees the new status (no workspace scope).
    assert!(topology_runs::cancel(&pool, id, ws).await.unwrap());
    assert_eq!(
        topology_runs::current_status(&pool, id).await.unwrap().as_deref(),
        Some("cancelled")
    );
    assert_eq!(topology_runs::status(&pool, id, ws).await.unwrap().status, "cancelled");

    // Already terminal → cannot cancel again; wrong workspace → no-op.
    assert!(!topology_runs::cancel(&pool, id, ws).await.unwrap());
    let other = seed_workspace(&pool).await;
    let id2 = Uuid::now_v7();
    topology_runs::enqueue_run(&pool, id2, ws, "t", &graph).await.unwrap();
    assert!(!topology_runs::cancel(&pool, id2, other).await.unwrap());
}