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, status: String, decided_by: Option, 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 { 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 { 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, 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 { 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 { 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, 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()) }