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

//! The sweepers must leave SELF-DRIVEN runs alone.
//!
//! A `tier='microvm'` or `tier='session'` run is inserted directly as `running` by
//! `phase_runner` and owned start-to-finish by its own `tokio::spawn`. Nothing
//! touches its `updated_at` or `checkpoint` while it is in flight, because there
//! is no per-step loop to hook.
//!
//! Both sweepers were written when every `running` row was a `cm_orchestrator` job
//! that checkpointed after each step, and neither filtered on tier. The result,
//! measured in production on 2026-08-06: `requeue_stale` declared a healthy microVM
//! run stale at 180 seconds, the worker claimed it, failed to deserialize its graph
//! placeholder, and killed the phase with "missing or invalid graph" — while the
//! agent went on working and its VM was orphaned for over an hour.
//!
//! **Every microVM mission that appeared to work did so by finishing inside three
//! minutes.** The end-to-end harness runs a 90-second mission, so it cannot see
//! this class at all — which is why the guard lives here, against the real SQL, and
//! costs milliseconds instead of eight minutes.

use cm_db::repo::topology_runs;
use cm_domain::WorkspaceId;
use uuid::Uuid;

/// Insert a run that is `running` and has looked idle for a long time — exactly
/// the shape a long agent turn presents.
async fn stale_running_run(pool: &sqlx::PgPool, ws: WorkspaceId, tier: &str) -> Uuid {
    let id = Uuid::now_v7();
    sqlx::query(
        "INSERT INTO topology_runs
             (id, workspace_id, task, kind, status, graph, tier,
              created_at, updated_at)
         VALUES ($1, $2, 'long turn', 'run', 'running', $3, $4,
                 now() - interval '30 minutes', now() - interval '30 minutes')",
    )
    .bind(id)
    .bind(ws.as_uuid())
    // The placeholder a self-driven run carries: no `kind`, so `TopologyGraph`
    // cannot parse it. That is what turned a requeue into a hard failure.
    .bind(serde_json::json!({ "nodes": [], "edges": [], "executor": tier }))
    .bind(tier)
    // `mission_id` is left NULL: it has an FK to `missions`, and `requeue_stale`
    // does not look at it. The reaper DOES filter on `mission_id IS NOT NULL` —
    // which is precisely what used to be mistaken for "orchestrator-driven" — and
    // it now shares the same tier allowlist, asserted below.
    .execute(pool)
    .await
    .expect("insert run");
    id
}

async fn status_of(pool: &sqlx::PgPool, id: Uuid) -> String {
    sqlx::query_scalar::<_, String>("SELECT status FROM topology_runs WHERE id = $1")
        .bind(id)
        .fetch_one(pool)
        .await
        .expect("read status")
}

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

/// The bug, in one assertion: 30 minutes idle and it must still be `running`.
#[tokio::test]
async fn requeue_stale_leaves_self_driven_runs_alone() {
    let pool = cm_testkit::test_pool().await;
    let ws = workspace(&pool).await;

    let microvm = stale_running_run(&pool, ws, "microvm").await;
    let session = stale_running_run(&pool, ws, "session").await;

    let moved = topology_runs::requeue_stale(&pool, 180.0)
        .await
        .expect("requeue");

    assert_eq!(
        status_of(&pool, microvm).await,
        "running",
        "a microvm run was requeued out from under a live VM ({moved} rows moved)"
    );
    assert_eq!(
        status_of(&pool, session).await,
        "running",
        "a session run was requeued out from under a live agent"
    );
}

/// And a worker-driven run in the same state MUST still be requeued, or the fix
/// would have been "stop sweeping" rather than "sweep the right rows".
#[tokio::test]
async fn requeue_stale_still_rescues_worker_driven_runs() {
    let pool = cm_testkit::test_pool().await;
    let ws = workspace(&pool).await;
    let team = stale_running_run(&pool, ws, "team").await;

    topology_runs::requeue_stale(&pool, 180.0).await.expect("requeue");

    assert_eq!(
        status_of(&pool, team).await,
        "queued",
        "a genuinely stalled team run must still be recovered"
    );
}

/// Defence in depth: even handed a queued self-driven row, the worker must not
/// adopt a job it cannot execute. Claiming one is what produced the
/// "missing or invalid graph" failure on a run that was perfectly healthy.
#[tokio::test]
async fn the_worker_will_not_claim_a_self_driven_run() {
    let pool = cm_testkit::test_pool().await;
    let ws = workspace(&pool).await;

    let id = Uuid::now_v7();
    topology_runs::enqueue_run_tier(
        &pool,
        id,
        ws,
        "should never be claimed",
        &serde_json::json!({ "nodes": [], "edges": [], "executor": "microvm" }),
        "microvm",
    )
    .await
    .expect("enqueue");

    let claimed = topology_runs::claim_next_queued(&pool).await.expect("claim");
    assert!(
        claimed.is_none(),
        "the worker claimed a microvm run: {:?}",
        claimed.map(|c| c.tier)
    );
    assert_eq!(status_of(&pool, id).await, "queued", "and it must be left as it was");
}

/// The composed tier is the mirror image of the two above and must not be
/// mistaken for them: it runs VMs, but the WORKER drives its graph, so being
/// claimed and requeued is exactly what gives it checkpointing and resume.
#[tokio::test]
async fn the_worker_claims_and_rescues_a_composed_run() {
    let pool = cm_testkit::test_pool().await;
    let ws = workspace(&pool).await;

    let id = Uuid::now_v7();
    topology_runs::enqueue_run_tier(
        &pool,
        id,
        ws,
        "compose the engines",
        // A real graph, unlike the self-driven placeholder: the worker plans it.
        &serde_json::json!({
            "kind": "pipeline",
            "nodes": [{ "id": "a", "role": "worker", "attrs": {} }],
            "edges": []
        }),
        "microvm_graph",
    )
    .await
    .expect("enqueue");

    let claimed = topology_runs::claim_next_queued(&pool)
        .await
        .expect("claim")
        .expect("a composed run must be claimable, or it never runs at all");
    assert_eq!(claimed.tier, "microvm_graph");
    assert_eq!(claimed.id, id);

    // And a composed run whose worker died must come back: its checkpoint is
    // what makes resume possible, and requeue is what triggers it.
    sqlx::query("UPDATE topology_runs SET updated_at = now() - interval '30 minutes' WHERE id = $1")
        .bind(id)
        .execute(&pool)
        .await
        .expect("age the run");
    topology_runs::requeue_stale(&pool, 180.0).await.expect("requeue");
    assert_eq!(
        status_of(&pool, id).await,
        "queued",
        "a composed run orphaned by a dead worker must be recovered"
    );
}

/// The allowlist is the single place this policy lives, so assert its membership
/// directly — a new self-driven tier added without touching it would otherwise be
/// exposed exactly as microvm was.
#[test]
fn the_allowlist_names_only_worker_driven_tiers() {
    for driven in ["team", "swarm", "company", "org"] {
        assert!(
            topology_runs::WORKER_DRIVEN_TIERS.contains(&driven),
            "{driven} is driven by the worker and must be sweepable"
        );
    }
    for self_driven in ["microvm", "session"] {
        assert!(
            !topology_runs::WORKER_DRIVEN_TIERS.contains(&self_driven),
            "{self_driven} owns its own lifecycle; sweeping it kills live work"
        );
    }
    // `microvm_graph` is worker-driven but NOT reapable: one of its steps is a
    // whole agent session in a VM, so "no step records in 15 minutes" is what a
    // healthy composed run looks like, and reaping it would orphan a live VM —
    // #54 in a different tier.
    assert!(topology_runs::WORKER_DRIVEN_TIERS.contains(&"microvm_graph"));
    assert!(!topology_runs::REAPABLE_TIERS.contains(&"microvm_graph"));
    for reapable in topology_runs::REAPABLE_TIERS {
        assert!(
            topology_runs::WORKER_DRIVEN_TIERS.contains(reapable),
            "{reapable} is reaped but never driven"
        );
    }
}