clawmates/crates/cm-runtime/tests/soak.rs

//! P6 security soak (spec §17 exit): under concurrent load with racing
//! decisions and duplicated resume attempts, ZERO unaudited gated
//! executions — every outbox row maps to exactly one approved decision,
//! every grant is consumed at most once, and nothing executes twice.

use std::sync::Arc;
use std::time::Duration;

use cm_domain::{
    AccessPolicy, Agent, AgentId, AgentStatus, Role, RunState, User, UserId, Workspace, WorkspaceId,
};
use cm_llm::ScriptedProvider;
use cm_runtime::{RunEventBody, Runtime, RuntimeConfig};
use cm_safety::{approvals, Decision, ResumeReady};

const SCENARIOS: &str = r#"
[[scenario]]
marker = "[[scenario:gated-email]]"

[[scenario.turns]]
events = [
    { type = "tool_use", name = "email.send", input = { to = "[email protected]", subject = "Soak", body = "Load test." } },
]

[[scenario.turns]]
events = [
    { type = "text", text = "Done." },
]
"#;

const RUNS: usize = 12;

#[tokio::test]
async fn concurrent_gated_runs_never_execute_unaudited_or_twice() {
    let pool = cm_testkit::test_pool().await;
    let ws = Workspace {
        id: WorkspaceId::new(),
        name: "Acme".into(),
        plan: "team".into(),
    };
    cm_db::repo::workspaces::insert(&pool, &ws).await.unwrap();
    let owner = User {
        id: UserId::new(),
        workspace_id: ws.id,
        email: format!("{}@acme.test", UserId::new()),
        role: Role::Owner,
        display_name: "Owner".into(),
        created_at: time::OffsetDateTime::UNIX_EPOCH,
    };
    cm_db::repo::users::insert(&pool, &owner).await.unwrap();
    let agent = Agent {
        id: AgentId::new(),
        workspace_id: ws.id,
        name: "Scout".into(),
        job_title: "Analyst".into(),
        system_prompt: String::new(),
        avatar: String::new(),
        accent: String::new(),
        wallpaper: String::new(),
        managed_by: owner.id,
        status: AgentStatus::Online,
    };
    cm_db::repo::agents::insert(&pool, &agent, &AccessPolicy::default())
        .await
        .unwrap();
    cm_db::repo::credits::add_lot(&pool, ws.id, 10_000, "soak")
        .await
        .unwrap();

    let rt = Runtime::new(
        pool.clone(),
        Arc::new(ScriptedProvider::from_toml(SCENARIOS).unwrap()),
        RuntimeConfig::basic("scripted", 1024),
    );
    // The durable sweeper races our explicit resume calls on purpose.
    rt.spawn_resume_sweeper(Duration::from_millis(25));

    // Launch all runs concurrently, each in its own session.
    let mut run_ids = Vec::new();
    let mut handles = Vec::new();
    for i in 0..RUNS {
        let session = cm_db::repo::sessions::create(&pool, agent.id, ws.id, &format!("Soak {i}"))
            .await
            .unwrap();
        let rt = rt.clone();
        handles.push(tokio::spawn(async move {
            let started = rt
                .send_message(session.id, "send it [[scenario:gated-email]]")
                .await
                .unwrap();
            let mut rx = started.events;
            while let Ok(envelope) = rx.recv().await {
                if matches!(
                    envelope.event,
                    RunEventBody::RunSuspended { .. } | RunEventBody::Error { .. }
                ) {
                    break;
                }
            }
            started.run_id
        }));
    }
    for handle in handles {
        run_ids.push(handle.await.unwrap());
    }

    // All suspended with pending approvals.
    let pending = approvals::list_pending(&pool, ws.id).await.unwrap();
    assert_eq!(pending.len(), RUNS);
    let outbox_before = sqlx::query_scalar::<_, i64>("SELECT count(*) FROM outbox")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(outbox_before, 0, "nothing may execute while pending");

    // Decide concurrently: even-indexed approve, odd-indexed reject — and
    // every decision is attempted TWICE concurrently (double-click storm),
    // plus an explicit duplicate resume racing the sweeper.
    let mut deciders = Vec::new();
    for (i, approval) in pending.iter().enumerate() {
        let decision = if i % 2 == 0 {
            Decision::Approve
        } else {
            Decision::Reject
        };
        for _ in 0..2 {
            let pool = pool.clone();
            let rt = rt.clone();
            let id = approval.id;
            let run_id = approval.run_id;
            let user = owner.id;
            deciders.push(tokio::spawn(async move {
                let decided = approvals::decide(&pool, id, user, decision).await;
                // Exactly one of the two concurrent attempts wins.
                let _ = rt
                    .resume_run(ResumeReady {
                        run_id,
                        approval_id: id,
                        approved: decision == Decision::Approve,
                    })
                    .await;
                decided.is_ok()
            }));
        }
    }
    let mut wins = 0;
    for decider in deciders {
        if decider.await.unwrap() {
            wins += 1;
        }
    }
    assert_eq!(wins, RUNS, "each approval decided exactly once");

    // Every run reaches Completed.
    for run_id in &run_ids {
        let mut done = false;
        for _ in 0..200 {
            let run = cm_db::repo::runs::get(&pool, *run_id).await.unwrap();
            if run.state == RunState::Completed {
                done = true;
                break;
            }
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
        assert!(done, "run {run_id} never completed");
    }

    // THE invariant: executions == approvals approved, exactly.
    let approved = RUNS / 2;
    let outbox = sqlx::query_scalar::<_, i64>("SELECT count(*) FROM outbox")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(
        outbox as usize, approved,
        "execute once per approval, never more"
    );

    // Every approved grant consumed exactly once; rejected ones have none.
    let consumed =
        sqlx::query_scalar::<_, i64>("SELECT count(*) FROM execution_grants WHERE consumed = true")
            .fetch_one(&pool)
            .await
            .unwrap();
    let total_grants = sqlx::query_scalar::<_, i64>("SELECT count(*) FROM execution_grants")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(consumed as usize, approved);
    assert_eq!(total_grants as usize, approved);

    // Every decision audited.
    let audited = sqlx::query_scalar::<_, i64>(
        "SELECT count(*) FROM audit_log WHERE event_type IN ('approval.approved', 'approval.rejected')",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert_eq!(audited as usize, RUNS);

    // No approval left pending.
    assert!(approvals::list_pending(&pool, ws.id)
        .await
        .unwrap()
        .is_empty());
}