car_engine/

authz.rs

//! Authorization pipeline for tool execution.
//!
//! Provides a structured pre-execution decision path with typed decisions.
//! Each stage can Allow, Deny, or AskUser, and the pipeline short-circuits
//! on the first Deny.

use car_ir::{Action, ActionType, ToolSchema};
use car_policy::PolicyEngine;
use car_state::StateStore;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

/// The stage that produced an authorization decision.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AuthzStage {
    /// Tool existence check.
    ToolExists,
    /// Capability check (agent-level whitelist/blacklist).
    Capability,
    /// Permission mode / approval policy.
    Permission,
    /// Permanent restrictions (never bypassable).
    Restriction,
    /// Policy engine rules.
    Policy,
    /// Executor-level parameter validation.
    Validation,
}

/// The authorization decision.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AuthzDecision {
    /// Execution is allowed to proceed.
    Allow,
    /// Execution requires user approval before proceeding.
    AskUser,
    /// Execution is denied.
    Deny,
}

/// A complete authorization result from the pipeline.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AuthzResult {
    /// The final decision.
    pub decision: AuthzDecision,
    /// The stage that produced the decision (for Deny/AskUser, the stage that stopped it).
    pub stage: AuthzStage,
    /// Machine-readable reason code.
    pub reason_code: String,
    /// Human-readable explanation.
    pub explanation: String,
    /// Results from each stage that was evaluated.
    pub stage_results: Vec<StageResult>,
}

/// Result from a single authorization stage.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StageResult {
    pub stage: AuthzStage,
    pub decision: AuthzDecision,
    pub reason: String,
}

impl AuthzResult {
    pub fn allowed(stage: AuthzStage) -> Self {
        Self {
            decision: AuthzDecision::Allow,
            stage,
            reason_code: "allowed".to_string(),
            explanation: "All authorization checks passed".to_string(),
            stage_results: Vec::new(),
        }
    }

    pub fn denied(stage: AuthzStage, reason_code: &str, explanation: &str) -> Self {
        Self {
            decision: AuthzDecision::Deny,
            stage,
            reason_code: reason_code.to_string(),
            explanation: explanation.to_string(),
            stage_results: Vec::new(),
        }
    }

    pub fn ask_user(stage: AuthzStage, reason_code: &str, explanation: &str) -> Self {
        Self {
            decision: AuthzDecision::AskUser,
            stage,
            reason_code: reason_code.to_string(),
            explanation: explanation.to_string(),
            stage_results: Vec::new(),
        }
    }

    fn with_stages(mut self, stages: Vec<StageResult>) -> Self {
        self.stage_results = stages;
        self
    }
}

/// A permanent restriction that can never be bypassed.
pub struct Restriction {
    pub name: String,
    pub description: String,
    check: Box<dyn Fn(&Action) -> Option<String> + Send + Sync>,
}

impl Restriction {
    pub fn new<F>(name: &str, description: &str, check: F) -> Self
    where
        F: Fn(&Action) -> Option<String> + Send + Sync + 'static,
    {
        Self {
            name: name.to_string(),
            description: description.to_string(),
            check: Box::new(check),
        }
    }

    fn check(&self, action: &Action) -> Option<String> {
        (self.check)(action)
    }
}

/// Callback for permission mode decisions (allow/ask/deny).
/// Products implement this to integrate their approval UX.
#[async_trait::async_trait]
pub trait PermissionHandler: Send + Sync {
    /// Decide whether to allow, ask, or deny a tool call.
    async fn check(&self, tool_name: &str, action: &Action) -> AuthzDecision;
}

/// Default permission handler that allows everything.
pub struct AllowAllPermissions;

#[async_trait::async_trait]
impl PermissionHandler for AllowAllPermissions {
    async fn check(&self, _tool_name: &str, _action: &Action) -> AuthzDecision {
        AuthzDecision::Allow
    }
}

/// The authorization pipeline.
pub struct AuthzPipeline {
    restrictions: Vec<Restriction>,
    permission_handler: Box<dyn PermissionHandler>,
}

impl AuthzPipeline {
    pub fn new() -> Self {
        Self {
            restrictions: Vec::new(),
            permission_handler: Box::new(AllowAllPermissions),
        }
    }

    /// Add a permanent restriction.
    pub fn add_restriction(&mut self, restriction: Restriction) {
        self.restrictions.push(restriction);
    }

    /// Set the permission handler.
    pub fn set_permission_handler(&mut self, handler: Box<dyn PermissionHandler>) {
        self.permission_handler = handler;
    }

    /// Run the full authorization pipeline for an action.
    ///
    /// Stages (in order):
    /// 1. Tool exists
    /// 2. Capability allows it
    /// 3. Permission mode / approval
    /// 4. Permanent restrictions
    /// 5. Policy engine
    /// 6. Executor-level validation
    pub async fn authorize(
        &self,
        action: &Action,
        tools: &HashMap<String, ToolSchema>,
        capabilities: Option<&crate::capabilities::CapabilitySet>,
        policies: &PolicyEngine,
        state: &StateStore,
    ) -> AuthzResult {
        let mut stages = Vec::new();

        // Stage 1: Tool exists
        if let Some(tool_name) = &action.tool {
            if action.action_type == ActionType::ToolCall && !tools.contains_key(tool_name) {
                stages.push(StageResult {
                    stage: AuthzStage::ToolExists,
                    decision: AuthzDecision::Deny,
                    reason: format!("tool '{}' not registered", tool_name),
                });
                return AuthzResult::denied(
                    AuthzStage::ToolExists,
                    "tool_not_found",
                    &format!("Tool '{}' is not registered", tool_name),
                )
                .with_stages(stages);
            }
        }
        stages.push(StageResult {
            stage: AuthzStage::ToolExists,
            decision: AuthzDecision::Allow,
            reason: "tool registered".to_string(),
        });

        // Stage 2: Capability check
        if let Some(caps) = capabilities {
            if let Some(tool_name) = &action.tool {
                if !caps.tool_allowed(tool_name) {
                    stages.push(StageResult {
                        stage: AuthzStage::Capability,
                        decision: AuthzDecision::Deny,
                        reason: format!("tool '{}' not in capability set", tool_name),
                    });
                    return AuthzResult::denied(
                        AuthzStage::Capability,
                        "capability_denied",
                        &format!("Tool '{}' denied by capability set", tool_name),
                    )
                    .with_stages(stages);
                }
            }
        }
        stages.push(StageResult {
            stage: AuthzStage::Capability,
            decision: AuthzDecision::Allow,
            reason: "capability check passed".to_string(),
        });

        // Stage 3: Permission mode
        if let Some(tool_name) = &action.tool {
            let perm = self.permission_handler.check(tool_name, action).await;
            stages.push(StageResult {
                stage: AuthzStage::Permission,
                decision: perm,
                reason: format!("permission handler returned {:?}", perm),
            });
            if perm == AuthzDecision::Deny {
                return AuthzResult::denied(
                    AuthzStage::Permission,
                    "permission_denied",
                    &format!("Permission denied for tool '{}'", tool_name),
                )
                .with_stages(stages);
            }
            if perm == AuthzDecision::AskUser {
                return AuthzResult::ask_user(
                    AuthzStage::Permission,
                    "approval_required",
                    &format!("Tool '{}' requires user approval", tool_name),
                )
                .with_stages(stages);
            }
        } else {
            stages.push(StageResult {
                stage: AuthzStage::Permission,
                decision: AuthzDecision::Allow,
                reason: "no tool name, skipped".to_string(),
            });
        }

        // Stage 4: Permanent restrictions
        for restriction in &self.restrictions {
            if let Some(reason) = restriction.check(action) {
                stages.push(StageResult {
                    stage: AuthzStage::Restriction,
                    decision: AuthzDecision::Deny,
                    reason: reason.clone(),
                });
                return AuthzResult::denied(
                    AuthzStage::Restriction,
                    &format!("restriction_{}", restriction.name),
                    &format!("Permanent restriction '{}': {}", restriction.name, reason),
                )
                .with_stages(stages);
            }
        }
        stages.push(StageResult {
            stage: AuthzStage::Restriction,
            decision: AuthzDecision::Allow,
            reason: "all restrictions passed".to_string(),
        });

        // Stage 5: Policy engine
        let violations = policies.check(action, state);
        if !violations.is_empty() {
            let reasons: Vec<String> = violations
                .iter()
                .map(|v| format!("{}: {}", v.policy_name, v.reason))
                .collect();
            stages.push(StageResult {
                stage: AuthzStage::Policy,
                decision: AuthzDecision::Deny,
                reason: reasons.join("; "),
            });
            return AuthzResult::denied(
                AuthzStage::Policy,
                "policy_violation",
                &format!("Policy violations: {}", reasons.join("; ")),
            )
            .with_stages(stages);
        }
        stages.push(StageResult {
            stage: AuthzStage::Policy,
            decision: AuthzDecision::Allow,
            reason: "all policies passed".to_string(),
        });

        // Stage 6: Validation (deferred to caller — we just mark it as passed here)
        stages.push(StageResult {
            stage: AuthzStage::Validation,
            decision: AuthzDecision::Allow,
            reason: "validation deferred".to_string(),
        });

        AuthzResult::allowed(AuthzStage::Validation).with_stages(stages)
    }
}

impl Default for AuthzPipeline {
    fn default() -> Self {
        Self::new()
    }
}

/// Bridges a [`car_policy::PermissionGate`] into the authorization
/// pipeline's permission stage.
///
/// This is where the permission-tier model (survey §3.4.3, §5.2.5) meets
/// the existing pipeline: the gate classifies each action's risk tier,
/// compares it to the session's granted standing authority, and consults
/// the durable approval ledger. Its decision maps onto the pipeline's
/// existing vocabulary — `Allow` proceeds, `NeedsApproval` becomes
/// `AskUser` (autonomy suspended pending a human decision), and a prior
/// rejection becomes `Deny`. Every decision is audited to the event log
/// as a `PermissionDecision` event so the tier reasoning is inspectable
/// rather than implicit.
///
/// # Two axes on one event
///
/// The audited event carries `reversibility` alongside `required_tier`,
/// from `car_policy::classify_reversibility` — the independent answer to
/// *can this be undone?* The gate itself does not consult it and its
/// decision does not depend on it; the field is recorded because the two
/// questions used to be fused inside `PermissionTier` and an audit trail
/// that reports only the ladder cannot tell a `git push` (recoverable) from
/// a charged card (not) — both arrive as `full_access` / `needs_approval`.
/// Splitting them at the point of record is what lets a later gate, a
/// reviewer, or a post-hoc analysis distinguish the two without re-deriving
/// the classification from a tool name months later. See
/// `docs/proposals/shepherd-substrate-adoption.md`.
pub struct TierPermissionHandler {
    gate: std::sync::Arc<tokio::sync::RwLock<car_policy::PermissionGate>>,
    log: Option<std::sync::Arc<tokio::sync::Mutex<car_eventlog::EventLog>>>,
}

impl TierPermissionHandler {
    pub fn new(gate: std::sync::Arc<tokio::sync::RwLock<car_policy::PermissionGate>>) -> Self {
        Self { gate, log: None }
    }

    /// Audit each gate decision to this event log as a `PermissionDecision`.
    pub fn with_event_log(
        mut self,
        log: std::sync::Arc<tokio::sync::Mutex<car_eventlog::EventLog>>,
    ) -> Self {
        self.log = Some(log);
        self
    }

    /// Record a durable human-in-the-loop decision through the gate and
    /// emit it to the event log as an `ApprovalRecorded` event — the
    /// auditable state transition §5.2.5 calls for ("who approved/rejected
    /// what, when, on what evidence"). `approve=false` records a rejection.
    /// Errs when the ledger journal write fails — the decision was NOT
    /// recorded, and no `ApprovalRecorded` event is emitted (review A7).
    pub async fn record_approval(
        &self,
        action: &Action,
        approve: bool,
        reviewer: &str,
        reason: &str,
        evidence: Option<String>,
    ) -> std::io::Result<car_policy::ApprovalRecord> {
        let record = {
            let mut gate = self.gate.write().await;
            if approve {
                gate.approve(action, reviewer, reason, evidence)?
            } else {
                gate.reject(action, reviewer, reason, evidence)?
            }
        };
        if let Some(log) = &self.log {
            let mut data = HashMap::new();
            data.insert("fingerprint".into(), record.fingerprint.clone().into());
            data.insert(
                "approval".into(),
                match record.decision {
                    car_policy::ApprovalDecision::Approved => "approved",
                    car_policy::ApprovalDecision::Rejected => "rejected",
                }
                .into(),
            );
            data.insert("required_tier".into(), record.required_tier.as_str().into());
            data.insert("reviewer".into(), record.reviewer.clone().into());
            data.insert("reason".into(), record.reason.clone().into());
            if let Some(ev) = &record.evidence {
                data.insert("evidence".into(), ev.clone().into());
            }
            log.lock().await.append(
                car_eventlog::EventKind::ApprovalRecorded,
                Some(&action.id),
                None,
                data,
            );
        }
        Ok(record)
    }

    /// Map a gate decision onto the pipeline's decision vocabulary.
    fn map_decision(decision: &car_policy::GateDecision) -> AuthzDecision {
        match decision {
            car_policy::GateDecision::Allow { .. } => AuthzDecision::Allow,
            car_policy::GateDecision::NeedsApproval { .. } => AuthzDecision::AskUser,
            car_policy::GateDecision::Deny { .. } => AuthzDecision::Deny,
        }
    }

    /// Flatten a gate decision into structured event data for the audit
    /// trail.
    ///
    /// `reversibility` is on every variant, including `allow`: an action that
    /// sailed through the gate still has a rollback contract, and an audit
    /// trail that only records the contract of the actions it *stopped* is
    /// missing exactly the rows an incident review reads first. It is derived
    /// from the action rather than the decision because the two axes are
    /// independent — the gate's verdict carries no information about whether
    /// the effect can be undone.
    fn decision_data(
        reversibility: car_ir::Reversibility,
        decision: &car_policy::GateDecision,
    ) -> HashMap<String, serde_json::Value> {
        use car_policy::GateDecision::*;
        let mut data = HashMap::new();
        data.insert("reversibility".into(), reversibility.as_str().into());
        match decision {
            Allow { required, granted } => {
                data.insert("gate_decision".into(), "allow".into());
                data.insert("required_tier".into(), required.as_str().into());
                data.insert("granted_tier".into(), granted.as_str().into());
            }
            NeedsApproval {
                required,
                granted,
                fingerprint,
                reason,
            } => {
                data.insert("gate_decision".into(), "needs_approval".into());
                data.insert("required_tier".into(), required.as_str().into());
                data.insert("granted_tier".into(), granted.as_str().into());
                data.insert("fingerprint".into(), fingerprint.clone().into());
                data.insert("reason".into(), reason.clone().into());
            }
            Deny {
                required,
                fingerprint,
                reason,
            } => {
                data.insert("gate_decision".into(), "deny".into());
                data.insert("required_tier".into(), required.as_str().into());
                data.insert("fingerprint".into(), fingerprint.clone().into());
                data.insert("reason".into(), reason.clone().into());
            }
        }
        data
    }
}

#[async_trait::async_trait]
impl PermissionHandler for TierPermissionHandler {
    async fn check(&self, _tool_name: &str, action: &Action) -> AuthzDecision {
        // Both axes from one flatten of the parameter payload — this runs per
        // action on the execution path, where the payload can be a whole
        // document or diff (Parslee-ai/car#856).
        let axes = {
            let gate = self.gate.read().await;
            gate.evaluate_axes(action, None, None)
        };
        let decision = axes.decision;
        if let Some(log) = &self.log {
            let data = Self::decision_data(axes.reversibility, &decision);
            log.lock().await.append(
                car_eventlog::EventKind::PermissionDecision,
                Some(&action.id),
                None,
                data,
            );
        }
        Self::map_decision(&decision)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use car_ir::{Action, ActionType, ToolSchema};

    fn test_action(tool: &str) -> Action {
        {
            let mut a = Action::new(ActionType::ToolCall);
            a.id = "test-1".to_string();
            a.tool = Some(tool.to_string());
            a
        }
    }

    fn test_tools() -> HashMap<String, ToolSchema> {
        let mut m = HashMap::new();
        m.insert(
            "read".to_string(),
            ToolSchema {
                name: "read".to_string(),
                source: car_ir::ToolSourceKind::UserDefined,
                description: "Read a file".to_string(),
                parameters: serde_json::json!({"type": "object"}),
                returns: None,
                idempotent: true,
                cache_ttl_secs: None,
                rate_limit: None,
            },
        );
        m
    }

    #[tokio::test]
    async fn test_allow_registered_tool() {
        let pipeline = AuthzPipeline::new();
        let tools = test_tools();
        let policies = PolicyEngine::new();
        let state = StateStore::new();

        let result = pipeline
            .authorize(&test_action("read"), &tools, None, &policies, &state)
            .await;
        assert_eq!(result.decision, AuthzDecision::Allow);
        assert_eq!(result.stage_results.len(), 6);
    }

    #[tokio::test]
    async fn test_deny_unregistered_tool() {
        let pipeline = AuthzPipeline::new();
        let tools = test_tools();
        let policies = PolicyEngine::new();
        let state = StateStore::new();

        let result = pipeline
            .authorize(&test_action("delete"), &tools, None, &policies, &state)
            .await;
        assert_eq!(result.decision, AuthzDecision::Deny);
        assert_eq!(result.stage, AuthzStage::ToolExists);
        assert_eq!(result.reason_code, "tool_not_found");
    }

    #[tokio::test]
    async fn test_capability_denial() {
        let pipeline = AuthzPipeline::new();
        let tools = test_tools();
        let policies = PolicyEngine::new();
        let state = StateStore::new();
        let mut caps = crate::capabilities::CapabilitySet::default();
        caps.denied_tools.insert("read".to_string());

        let result = pipeline
            .authorize(&test_action("read"), &tools, Some(&caps), &policies, &state)
            .await;
        assert_eq!(result.decision, AuthzDecision::Deny);
        assert_eq!(result.stage, AuthzStage::Capability);
    }

    #[tokio::test]
    async fn test_restriction() {
        let mut pipeline = AuthzPipeline::new();
        pipeline.add_restriction(Restriction::new("no_read", "Never allow read", |action| {
            if action.tool.as_deref() == Some("read") {
                Some("reads are restricted".to_string())
            } else {
                None
            }
        }));
        let tools = test_tools();
        let policies = PolicyEngine::new();
        let state = StateStore::new();

        let result = pipeline
            .authorize(&test_action("read"), &tools, None, &policies, &state)
            .await;
        assert_eq!(result.decision, AuthzDecision::Deny);
        assert_eq!(result.stage, AuthzStage::Restriction);
    }

    #[tokio::test]
    async fn test_policy_violation() {
        let pipeline = AuthzPipeline::new();
        let tools = test_tools();
        let state = StateStore::new();
        let mut policies = PolicyEngine::new();
        policies.register(
            "deny_all",
            Box::new(|_action: &Action, _state: &StateStore| Some("denied by test".to_string())),
            "test policy",
        );

        let result = pipeline
            .authorize(&test_action("read"), &tools, None, &policies, &state)
            .await;
        assert_eq!(result.decision, AuthzDecision::Deny);
        assert_eq!(result.stage, AuthzStage::Policy);
    }

    #[tokio::test]
    async fn test_ask_user_permission() {
        struct AskPermissions;
        #[async_trait::async_trait]
        impl PermissionHandler for AskPermissions {
            async fn check(&self, _tool_name: &str, _action: &Action) -> AuthzDecision {
                AuthzDecision::AskUser
            }
        }

        let mut pipeline = AuthzPipeline::new();
        pipeline.set_permission_handler(Box::new(AskPermissions));
        let tools = test_tools();
        let policies = PolicyEngine::new();
        let state = StateStore::new();

        let result = pipeline
            .authorize(&test_action("read"), &tools, None, &policies, &state)
            .await;
        assert_eq!(result.decision, AuthzDecision::AskUser);
        assert_eq!(result.stage, AuthzStage::Permission);
        assert_eq!(result.reason_code, "approval_required");
    }

    #[tokio::test]
    async fn test_stage_results_trace() {
        let pipeline = AuthzPipeline::new();
        let tools = test_tools();
        let policies = PolicyEngine::new();
        let state = StateStore::new();

        let result = pipeline
            .authorize(&test_action("read"), &tools, None, &policies, &state)
            .await;
        // All 6 stages should be present when everything passes
        let stage_names: Vec<AuthzStage> = result.stage_results.iter().map(|s| s.stage).collect();
        assert_eq!(
            stage_names,
            vec![
                AuthzStage::ToolExists,
                AuthzStage::Capability,
                AuthzStage::Permission,
                AuthzStage::Restriction,
                AuthzStage::Policy,
                AuthzStage::Validation,
            ]
        );
    }

    #[tokio::test]
    async fn test_short_circuit_on_deny() {
        let pipeline = AuthzPipeline::new();
        let tools = test_tools();
        let policies = PolicyEngine::new();
        let state = StateStore::new();

        // Unregistered tool should short-circuit at stage 1
        let result = pipeline
            .authorize(&test_action("nonexistent"), &tools, None, &policies, &state)
            .await;
        assert_eq!(result.stage_results.len(), 1);
        assert_eq!(result.stage_results[0].stage, AuthzStage::ToolExists);
    }

    #[tokio::test]
    async fn test_serde_roundtrip() {
        let result = AuthzResult::denied(AuthzStage::Policy, "policy_violation", "Test violation");
        let json = serde_json::to_string(&result).unwrap();
        let roundtripped: AuthzResult = serde_json::from_str(&json).unwrap();
        assert_eq!(roundtripped.decision, AuthzDecision::Deny);
        assert_eq!(roundtripped.stage, AuthzStage::Policy);
        assert_eq!(roundtripped.reason_code, "policy_violation");
    }

    // --- TierPermissionHandler bridge (permission tiers → pipeline) ---

    use std::sync::Arc;
    use tokio::sync::RwLock;

    fn deploy_action() -> Action {
        let mut a = test_action("deploy_service");
        a.id = "deploy-1".to_string();
        a
    }

    #[tokio::test]
    async fn tier_handler_asks_user_for_full_access() {
        // A FullAccess action under a SandboxEdit grant must escalate to
        // the user, surfacing as AuthzDecision::AskUser through the
        // pipeline's permission stage.
        let gate = Arc::new(RwLock::new(car_policy::PermissionGate::new(
            car_policy::PermissionTier::SandboxEdit,
        )));
        let handler = TierPermissionHandler::new(gate);
        assert_eq!(
            handler.check("deploy_service", &deploy_action()).await,
            AuthzDecision::AskUser
        );
    }

    #[tokio::test]
    async fn tier_handler_allows_after_approval_and_audits() {
        let gate = Arc::new(RwLock::new(car_policy::PermissionGate::new(
            car_policy::PermissionTier::SandboxEdit,
        )));
        let log = Arc::new(tokio::sync::Mutex::new(car_eventlog::EventLog::new()));
        let handler = TierPermissionHandler::new(gate.clone()).with_event_log(log.clone());
        let action = deploy_action();

        // First pass: escalation, audited.
        assert_eq!(
            handler.check("deploy_service", &action).await,
            AuthzDecision::AskUser
        );
        // Human approves the operation through the handler (durable ledger
        // entry + ApprovalRecorded audit event).
        handler
            .record_approval(&action, true, "matt", "reviewed", Some("diff".into()))
            .await
            .unwrap();
        // Now the same operation is allowed.
        assert_eq!(
            handler.check("deploy_service", &action).await,
            AuthzDecision::Allow
        );

        let log = log.lock().await;
        // Both evaluations were audited as PermissionDecision events.
        let decisions: Vec<_> = log
            .events()
            .iter()
            .filter(|e| e.kind == car_eventlog::EventKind::PermissionDecision)
            .collect();
        assert_eq!(decisions.len(), 2);
        assert_eq!(
            decisions[0].data.get("gate_decision").unwrap(),
            "needs_approval"
        );
        assert_eq!(decisions[1].data.get("gate_decision").unwrap(), "allow");
        // The second axis rides on BOTH rows, the allow included: an action
        // that passed the gate still has a rollback contract, and that is the
        // row an incident review reads first.
        for d in &decisions {
            assert_eq!(d.data.get("reversibility").unwrap(), "compensable");
        }
        // The approval itself was audited as a durable state transition.
        let approvals: Vec<_> = log
            .events()
            .iter()
            .filter(|e| e.kind == car_eventlog::EventKind::ApprovalRecorded)
            .collect();
        assert_eq!(approvals.len(), 1);
        assert_eq!(approvals[0].data.get("approval").unwrap(), "approved");
        assert_eq!(approvals[0].data.get("reviewer").unwrap(), "matt");
    }

    #[tokio::test]
    async fn tier_handler_denies_after_rejection() {
        let gate = Arc::new(RwLock::new(car_policy::PermissionGate::new(
            car_policy::PermissionTier::FullAccess,
        )));
        let handler = TierPermissionHandler::new(gate.clone());
        let action = deploy_action();
        gate.write()
            .await
            .reject(&action, "matt", "not authorized", None)
            .unwrap();
        assert_eq!(
            handler.check("deploy_service", &action).await,
            AuthzDecision::Deny
        );
    }

    #[tokio::test]
    async fn permission_decision_events_carry_both_axes() {
        // A deploy and an outbound email produce audit rows that are IDENTICAL
        // on the authority ladder — full_access, escalated to a human — and
        // differ completely in whether the effect can be taken back. Without
        // `reversibility` on the event nothing downstream can tell them apart,
        // which is the conflation the second axis removes.
        let gate = Arc::new(RwLock::new(car_policy::PermissionGate::new(
            car_policy::PermissionTier::FullAccess,
        )));
        let log = Arc::new(tokio::sync::Mutex::new(car_eventlog::EventLog::new()));
        let handler = TierPermissionHandler::new(gate).with_event_log(log.clone());

        let deploy = deploy_action();
        let mut email = test_action("send_email");
        email.id = "email-1".to_string();

        assert_eq!(
            handler.check("deploy_service", &deploy).await,
            AuthzDecision::AskUser
        );
        assert_eq!(
            handler.check("send_email", &email).await,
            AuthzDecision::AskUser
        );

        let log = log.lock().await;
        let rows: Vec<_> = log
            .events()
            .iter()
            .filter(|e| e.kind == car_eventlog::EventKind::PermissionDecision)
            .collect();
        assert_eq!(rows.len(), 2);
        for row in &rows {
            assert_eq!(row.data.get("gate_decision").unwrap(), "needs_approval");
            assert_eq!(row.data.get("required_tier").unwrap(), "full_access");
        }
        // ...and this is what the ladder alone could not have said.
        assert_eq!(rows[0].data.get("reversibility").unwrap(), "compensable");
        assert_eq!(rows[1].data.get("reversibility").unwrap(), "irreversible");
    }
}