clawmates/crates/cm-safety/src/approvals.rs

use cm_domain::{AgentId, UserId, WorkspaceId};
use sqlx::PgPool;
use uuid::Uuid;

use crate::{Approval, ApprovalStatus, Decision, NewApproval, ResumeReady, SafetyError};

#[allow(clippy::too_many_arguments)]
fn map_row(
    id: Uuid,
    workspace_id: Uuid,
    run_id: Uuid,
    session_key: String,
    action_type: String,
    category: String,
    payload: serde_json::Value,
    preview: serde_json::Value,
    requested_by_agent: Uuid,
    taint_sources: Vec<String>,
    status: String,
    decided_by: Option<Uuid>,
    created_at: time::OffsetDateTime,
) -> Approval {
    Approval {
        id,
        workspace_id: WorkspaceId::from(workspace_id),
        run_id,
        session_key,
        action_type,
        category: category.parse().expect("category CHECK constraint"),
        payload,
        preview,
        requested_by_agent: AgentId::from(requested_by_agent),
        taint_sources,
        status: status.parse().expect("status CHECK constraint"),
        decided_by: decided_by.map(UserId::from),
        created_at,
    }
}

macro_rules! approval_from {
    ($row:expr) => {
        map_row(
            $row.id,
            $row.workspace_id,
            $row.run_id,
            $row.session_key,
            $row.action_type,
            $row.category,
            $row.payload,
            $row.preview,
            $row.requested_by_agent,
            $row.taint_sources,
            $row.status,
            $row.decided_by,
            $row.created_at,
        )
    };
}

pub async fn create(pool: &PgPool, new: NewApproval) -> Result<Approval, SafetyError> {
    let id = Uuid::now_v7();
    let row = sqlx::query!(
        r#"INSERT INTO approvals
            (id, workspace_id, run_id, session_key, action_type, category,
             payload, preview, requested_by_agent, taint_sources, status,
             expires_at)
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, 'pending', $11)
         RETURNING id, workspace_id, run_id, session_key, action_type,
                   category, payload, preview, requested_by_agent,
                   taint_sources, status, decided_by, created_at"#,
        id,
        new.workspace_id.as_uuid(),
        new.run_id,
        new.session_key,
        new.action_type,
        new.category.as_str(),
        new.payload,
        new.preview,
        new.requested_by_agent.as_uuid(),
        &new.taint_sources,
        new.expires_at,
    )
    .fetch_one(pool)
    .await?;
    Ok(approval_from!(row))
}

pub async fn get(pool: &PgPool, id: Uuid) -> Result<Approval, SafetyError> {
    let row = sqlx::query!(
        r#"SELECT id, workspace_id, run_id, session_key, action_type,
                  category, payload, preview, requested_by_agent,
                  taint_sources, status, decided_by, created_at
           FROM approvals WHERE id = $1"#,
        id,
    )
    .fetch_optional(pool)
    .await?
    .ok_or(SafetyError::NotFound)?;
    Ok(approval_from!(row))
}

/// The approval queue (§10), oldest first so reviews happen in order.
pub async fn list_pending(
    pool: &PgPool,
    workspace_id: WorkspaceId,
) -> Result<Vec<Approval>, SafetyError> {
    let rows = sqlx::query!(
        r#"SELECT id, workspace_id, run_id, session_key, action_type,
                  category, payload, preview, requested_by_agent,
                  taint_sources, status, decided_by, created_at
           FROM approvals
           WHERE workspace_id = $1 AND status = 'pending'
           ORDER BY created_at"#,
        workspace_id.as_uuid(),
    )
    .fetch_all(pool)
    .await?;
    Ok(rows.into_iter().map(|row| approval_from!(row)).collect())
}

/// Idempotent decision: a compare-and-swap from `pending`, the audit row,
/// and (on approval) the single-use grant — one transaction. Losing the
/// race surfaces as `AlreadyDecided`, never a double execution.
pub async fn decide(
    pool: &PgPool,
    id: Uuid,
    decided_by: UserId,
    decision: Decision,
) -> Result<Approval, SafetyError> {
    let status = match decision {
        Decision::Approve => ApprovalStatus::Approved,
        Decision::Reject => ApprovalStatus::Rejected,
    };
    let mut tx = pool.begin().await?;
    let row = sqlx::query!(
        r#"UPDATE approvals
           SET status = $2, decided_by = $3, decided_at = now()
           WHERE id = $1 AND status = 'pending'
           RETURNING id, workspace_id, run_id, session_key, action_type,
                     category, payload, preview, requested_by_agent,
                     taint_sources, status, decided_by, created_at"#,
        id,
        status.as_str(),
        decided_by.as_uuid(),
    )
    .fetch_optional(&mut *tx)
    .await?;
    let Some(row) = row else {
        // Distinguish missing from already-decided for honest API errors.
        let exists = sqlx::query_scalar!("SELECT 1 AS x FROM approvals WHERE id = $1", id)
            .fetch_optional(pool)
            .await?
            .is_some();
        return Err(if exists {
            SafetyError::AlreadyDecided
        } else {
            SafetyError::NotFound
        });
    };
    let approval = approval_from!(row);

    sqlx::query!(
        "INSERT INTO audit_log
            (workspace_id, actor_kind, actor_id, event_type, subject_type,
             subject_id, detail)
         VALUES ($1, 'user', $2, $3, 'approval', $4, $5)",
        approval.workspace_id.as_uuid(),
        decided_by.as_uuid(),
        match decision {
            Decision::Approve => "approval.approved",
            Decision::Reject => "approval.rejected",
        },
        approval.id.to_string(),
        serde_json::json!({
            "action_type": approval.action_type,
            "category": approval.category,
        }),
    )
    .execute(&mut *tx)
    .await?;

    if decision == Decision::Approve {
        sqlx::query!(
            "INSERT INTO execution_grants (id, approval_id, nonce)
             VALUES ($1, $2, $3)",
            Uuid::now_v7(),
            approval.id,
            Uuid::now_v7().to_string(),
        )
        .execute(&mut *tx)
        .await?;
    }
    tx.commit().await?;
    Ok(approval)
}

/// Expires overdue pending approvals; their runs fail on resume sweep.
pub async fn sweep_expired(pool: &PgPool) -> Result<u64, SafetyError> {
    let result = sqlx::query!(
        "UPDATE approvals SET status = 'expired'
         WHERE status = 'pending' AND expires_at IS NOT NULL AND expires_at < now()",
    )
    .execute(pool)
    .await?;
    Ok(result.rows_affected())
}

/// Decided approvals whose suspended runs still await resumption — the
/// durable work queue for the resume sweeper.
pub async fn decided_unresumed(pool: &PgPool) -> Result<Vec<ResumeReady>, SafetyError> {
    let rows = sqlx::query!(
        r#"SELECT a.run_id, a.id AS approval_id, a.status
           FROM approvals a
           JOIN agent_runs r ON r.id = a.run_id
           WHERE a.status IN ('approved', 'rejected')
             AND r.state = 'awaiting_approval'
           ORDER BY a.decided_at"#,
    )
    .fetch_all(pool)
    .await?;
    Ok(rows
        .into_iter()
        .map(|row| ResumeReady {
            run_id: row.run_id,
            approval_id: row.approval_id,
            approved: row.status == "approved",
        })
        .collect())
}