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::{teams, topology_runs, users, workspaces};
use cm_domain::{
    AccessPolicy, Agent, AgentId, AgentStatus, Role, User, UserId, Workspace, WorkspaceId,
};
use serde_json::json;
use uuid::Uuid;

async fn seed_user(pool: &sqlx::PgPool, ws: WorkspaceId, email: &str) -> UserId {
    let user = User {
        id: UserId::new(),
        workspace_id: ws,
        email: email.into(),
        role: Role::Owner,
        display_name: "Owner".into(),
        created_at: time::OffsetDateTime::UNIX_EPOCH,
    };
    users::insert(pool, &user).await.unwrap();
    user.id
}

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());
}

async fn seed_team(
    pool: &sqlx::PgPool,
    ws: WorkspaceId,
    user_id: UserId,
    lifecycle: &str,
    claw_count: usize,
) -> (Uuid, Vec<Uuid>) {
    let team_id = Uuid::now_v7();
    let graph = json!({"kind": "hub_spoke", "nodes": [], "edges": []});
    teams::insert_team_with_lifecycle(pool, team_id, ws, "T", "hub_spoke", &graph, lifecycle)
        .await
        .unwrap();
    let mut claws = Vec::with_capacity(claw_count);
    for i in 0..claw_count {
        let agent = Agent {
            id: AgentId::new(),
            workspace_id: ws,
            name: format!("claw{i}"),
            job_title: "worker".into(),
            system_prompt: String::new(),
            avatar: String::new(),
            accent: String::new(),
            wallpaper: String::new(),
            managed_by: user_id,
            status: AgentStatus::Online,
        };
        cm_db::repo::agents::insert(pool, &agent, &AccessPolicy::default())
            .await
            .unwrap();
        teams::add_member(
            pool,
            team_id,
            &format!("n{i}"),
            agent.id.as_uuid(),
            "worker",
        )
        .await
        .unwrap();
        claws.push(agent.id.as_uuid());
    }
    (team_id, claws)
}

#[tokio::test]
async fn check_ephemeral_teardown_returns_claws_when_no_siblings_left() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let user_id = seed_user(&pool, ws, "[email protected]").await;
    let (team_id, claws) = seed_team(&pool, ws, user_id, "ephemeral", 3).await;

    let graph = json!({"kind": "hub_spoke", "nodes": [], "edges": []});
    let run_id = Uuid::now_v7();
    topology_runs::enqueue_run_for_team(&pool, run_id, ws, "task", &graph, team_id)
        .await
        .unwrap();
    topology_runs::complete(&pool, run_id, &json!({}))
        .await
        .unwrap();

    let teardown = topology_runs::check_ephemeral_teardown(&pool, run_id)
        .await
        .unwrap()
        .expect("ephemeral team, no siblings — should return teardown");
    assert_eq!(teardown.team_id, team_id);
    assert_eq!(teardown.workspace_id, ws.as_uuid());
    let mut got = teardown.claw_ids.clone();
    let mut want = claws.clone();
    got.sort();
    want.sort();
    assert_eq!(got, want, "returns every bound claw");
}

#[tokio::test]
async fn check_ephemeral_teardown_holds_when_siblings_in_flight() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let user_id = seed_user(&pool, ws, "[email protected]").await;
    let (team_id, _) = seed_team(&pool, ws, user_id, "ephemeral", 2).await;

    let graph = json!({"kind": "hub_spoke", "nodes": [], "edges": []});
    let done = Uuid::now_v7();
    let still_queued = Uuid::now_v7();
    topology_runs::enqueue_run_for_team(&pool, done, ws, "first", &graph, team_id)
        .await
        .unwrap();
    topology_runs::enqueue_run_for_team(&pool, still_queued, ws, "second", &graph, team_id)
        .await
        .unwrap();
    topology_runs::complete(&pool, done, &json!({}))
        .await
        .unwrap();

    let teardown = topology_runs::check_ephemeral_teardown(&pool, done)
        .await
        .unwrap();
    assert!(teardown.is_none(), "sibling still queued → hold teardown");
}

#[tokio::test]
async fn check_ephemeral_teardown_ignores_permanent_teams() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let user_id = seed_user(&pool, ws, "[email protected]").await;
    let (team_id, _) = seed_team(&pool, ws, user_id, "permanent", 1).await;

    let graph = json!({"kind": "hub_spoke", "nodes": [], "edges": []});
    let run_id = Uuid::now_v7();
    topology_runs::enqueue_run_for_team(&pool, run_id, ws, "task", &graph, team_id)
        .await
        .unwrap();
    topology_runs::complete(&pool, run_id, &json!({}))
        .await
        .unwrap();

    let teardown = topology_runs::check_ephemeral_teardown(&pool, run_id)
        .await
        .unwrap();
    assert!(teardown.is_none(), "permanent teams are never torn down");
}