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::{loops, topology_runs, users, workspaces};
use cm_domain::{Role, User, UserId, 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());
}

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

    // A loop needs a valid `created_by` user in the same workspace.
    let user = User {
        id: UserId::new(),
        workspace_id: ws,
        email: "[email protected]".into(),
        role: Role::Owner,
        display_name: "Hop".into(),
        created_at: time::OffsetDateTime::UNIX_EPOCH,
    };
    users::insert(&pool, &user).await.unwrap();

    let graph = json!({"kind": "pipeline", "nodes": [{"id":"n1","role":"drafter"}], "edges": []});
    let triggers = json!({});
    let repeat = json!({"kind": "infinite"});
    let loop_id = loops::create(
        &pool,
        loops::NewLoop {
            workspace_id: ws.as_uuid(),
            title: "L",
            description: "d",
            graph: &graph,
            task_template: "task",
            triggers: &triggers,
            repeat_policy: &repeat,
            enabled: true,
            next_fire_at: None,
            webhook_token: None,
            webhook_signing_key: None,
            created_by: user.id.as_uuid(),
        },
    )
    .await
    .unwrap();

    // Three iterations of the target loop, plus a naked run + another loop's
    // iteration that should be filtered out.
    let it1 = loops::enqueue_iteration(&pool, loop_id, ws.as_uuid(), "t1", &graph, 1, None)
        .await
        .unwrap();
    let it2 = loops::enqueue_iteration(&pool, loop_id, ws.as_uuid(), "t2", &graph, 2, Some(it1))
        .await
        .unwrap();
    let it3 = loops::enqueue_iteration(&pool, loop_id, ws.as_uuid(), "t3", &graph, 3, Some(it2))
        .await
        .unwrap();
    topology_runs::enqueue_run(&pool, Uuid::now_v7(), ws, "naked", &graph)
        .await
        .unwrap();
    let other_loop = loops::create(
        &pool,
        loops::NewLoop {
            workspace_id: ws.as_uuid(),
            title: "L2",
            description: "d2",
            graph: &graph,
            task_template: "task2",
            triggers: &triggers,
            repeat_policy: &repeat,
            enabled: true,
            next_fire_at: None,
            webhook_token: None,
            webhook_signing_key: None,
            created_by: user.id.as_uuid(),
        },
    )
    .await
    .unwrap();
    let _other_it = loops::enqueue_iteration(
        &pool,
        other_loop,
        ws.as_uuid(),
        "other",
        &graph,
        1,
        None,
    )
    .await
    .unwrap();

    let rows = topology_runs::list_by_loop(&pool, ws, loop_id, 20)
        .await
        .unwrap();

    assert_eq!(rows.len(), 3, "only the target loop's iterations");
    // Newest iteration first.
    assert_eq!(rows[0].iteration, Some(3));
    assert_eq!(rows[0].id, it3);
    assert_eq!(rows[1].iteration, Some(2));
    assert_eq!(rows[1].id, it2);
    assert_eq!(rows[2].iteration, Some(1));
    assert_eq!(rows[2].id, it1);
    // Each iteration is still a durable run — finished_at is None until completion.
    assert!(rows.iter().all(|r| r.finished_at.is_none()));
    assert!(rows.iter().all(|r| r.kind == "run"));

    // Wrong workspace: nothing.
    let other_ws = seed_workspace(&pool).await;
    let cross = topology_runs::list_by_loop(&pool, other_ws, loop_id, 20)
        .await
        .unwrap();
    assert!(cross.is_empty(), "loops are scoped by workspace");
}