car-server-core 0.49.0

//! Session permission-tier enforcement at proposal admission (Parslee-ai/car#890).
//!
//! The daemon has had the whole permission machinery for a while: a
//! per-session [`car_policy::PermissionGate`] carrying the granted standing
//! tier and the risk classifier, a daemon-wide durable
//! [`car_policy::ApprovalLedger`], and the `permission.evaluate` /
//! `permission.approve` / `permission.reject` RPCs that read and write them.
//! What it did not have was a *call site*: `proposal.submit` dispatched
//! straight to the executor, so an action the daemon's own
//! `permission.evaluate` called `needs_approval` executed anyway — on the same
//! session, in the same breath — and its state write persisted. The published
//! contract (`docs/websocket-protocol.md`, the `car_policy::permission` module
//! doc) said such an action "cannot run autonomously". It could.
//!
//! [`PermissionAdmissionGate`] is that call site. It is an
//! [`AdmissionGate`], so it runs on every proposal admitted by a session
//! runtime, *before any action executes*, alongside
//! `car_engine::StaticVerificationGate` and [`crate::supervision::SupervisionGate`].
//!
//! # One verdict, one evaluator
//!
//! The gate calls `PermissionGate::evaluate_axes(action, None, Some(ledger))`
//! — the **identical call** `handle_permission_evaluate` makes through
//! `evaluate_actions`. That is load-bearing, not incidental: an enforcement
//! point that classified actions with its own fresh `RiskClassifier` would be
//! free to disagree with the advisory RPC an operator is looking at, which
//! replaces one inconsistency with a worse one. The gate holds the *same*
//! `Arc<RwLock<PermissionGate>>` the `permission.*` handlers hold, so a
//! `permission.set_tier` on the session is visible here on the very next
//! submit, and a custom classifier rule installed on the session applies to
//! both surfaces by construction.
//!
//! # Why the gate reads the shared ledger itself
//!
//! Escalations resolve through the daemon-wide
//! [`crate::session::ServerState::approval_ledger`], not through the
//! executor's own. `Runtime::approval_ledger` is `None` on daemon sessions and
//! must stay that way — a second ledger instance would let the approver's
//! decision land somewhere the runner never reads. So the loop closes here
//! instead: once `permission.approve` records a decision for an action's
//! fingerprint, `evaluate_axes` returns `Allow` on the next submit and this
//! gate produces **no escalation at all**, which is what admits the proposal.
//!
//! The consequence for an *unapproved* escalation is deliberate. The executor
//! resolves `GateOutcome::NeedsApproval` against its own (absent) ledger, finds
//! nothing, and fails closed with the escalation's reason and fingerprint on
//! every rejected action. That is the intended UX: the caller is told the
//! proposal needs a human, and told exactly which fingerprints to approve.
//!
//! # Which fingerprint an operator approves
//!
//! [`GateOutcome::NeedsApproval`] carries one fingerprint, and a proposal can
//! escalate several actions, so the escalation's own fingerprint is a
//! deterministic composite (sorted + deduped per-action fingerprints under a
//! `permission:` prefix) — the same shape `car_engine::SkillCeilingGate` uses
//! for its multi-action case. **It is an identity, not an approval target**:
//! nothing looks it up. The approvable fingerprints are the per-action ones,
//! and the `reason` names every one of them next to its required and granted
//! tier so `permission.approve { fingerprint, required_tier }` can be driven
//! straight off the error text.

use std::collections::HashSet;
use std::sync::Arc;

use car_engine::admission::{AdmissionGate, GateContext, GateOutcome};
use car_ir::ActionProposal;
use tokio::sync::RwLock;

/// Enforces the session's granted permission tier on every submitted proposal.
///
/// Both fields are shared handles, never owned copies:
/// - `gate` is the session's own [`car_policy::PermissionGate`], so the tier
///   the `permission.*` RPCs report is the tier enforced here.
/// - `ledger` is the daemon-wide approval ledger, so a decision recorded on
///   the host connection clears an escalation raised on an agent connection,
///   and survives a restart.
pub struct PermissionAdmissionGate {
    gate: Arc<RwLock<car_policy::PermissionGate>>,
    ledger: Arc<RwLock<car_policy::ApprovalLedger>>,
}

impl PermissionAdmissionGate {
    /// Build a gate over a session's permission gate and the shared daemon
    /// approval ledger. Pass the very same `Arc`s the `permission.*` handlers
    /// use — a second `PermissionGate` instance is the bug class this exists
    /// to close.
    pub fn new(
        gate: Arc<RwLock<car_policy::PermissionGate>>,
        ledger: Arc<RwLock<car_policy::ApprovalLedger>>,
    ) -> Self {
        Self { gate, ledger }
    }
}

#[async_trait::async_trait]
impl AdmissionGate for PermissionAdmissionGate {
    fn name(&self) -> &str {
        "permission"
    }

    async fn check(&self, proposal: &ActionProposal, _ctx: &GateContext<'_>) -> GateOutcome {
        // Lock order is gate-then-ledger, matching every `permission.*`
        // handler (evaluate/pending read both in that order; approve/reject
        // reads the gate then writes the ledger), so admission cannot
        // deadlock against a concurrent approval on the same connection.
        let gate = self.gate.read().await;
        let ledger = self.ledger.read().await;

        let mut denied: HashSet<String> = HashSet::new();
        let mut deny_notes: Vec<String> = Vec::new();
        let mut escalate: HashSet<String> = HashSet::new();
        let mut escalation_notes: Vec<String> = Vec::new();
        let mut fingerprints: Vec<String> = Vec::new();

        for action in &proposal.actions {
            // The identical call `permission.evaluate` makes — same gate, same
            // classifier, same ledger. See the module doc for why.
            match gate.evaluate_axes(action, None, Some(&ledger)).decision {
                car_policy::GateDecision::Allow { .. } => {}
                car_policy::GateDecision::Deny {
                    required,
                    fingerprint,
                    reason,
                } => {
                    denied.insert(action.id.clone());
                    deny_notes.push(format!(
                        "action '{}' requires {} and was {} (fingerprint: {fingerprint})",
                        action.id,
                        required.as_str(),
                        reason,
                    ));
                }
                car_policy::GateDecision::NeedsApproval {
                    required,
                    granted,
                    fingerprint,
                    reason,
                } => {
                    escalate.insert(action.id.clone());
                    escalation_notes.push(format!(
                        "action '{}' requires {} but the session is granted {} — {reason} \
                         (approve fingerprint: {fingerprint})",
                        action.id,
                        required.as_str(),
                        granted.as_str(),
                    ));
                    fingerprints.push(fingerprint);
                }
            }
        }

        // Fail-closed: a standing operator rejection is not resolvable by an
        // approval elsewhere in the same proposal, so a Deny anywhere sinks the
        // whole thing regardless of what else escalated. (The engine's own
        // aggregation takes the same stance — a hard reject is never
        // approval-resolvable.)
        if !denied.is_empty() {
            return GateOutcome::Reject {
                blocked: denied,
                reason: format!(
                    "operator previously rejected this operation: {}",
                    deny_notes.join("; ")
                ),
            };
        }

        if escalate.is_empty() {
            return GateOutcome::Allow;
        }

        // Sorted + deduped so the same action set in a different proposal order
        // yields the same escalation identity.
        fingerprints.sort();
        fingerprints.dedup();
        GateOutcome::NeedsApproval {
            actions: escalate,
            fingerprint: format!("permission:{}", fingerprints.join(",")),
            reason: format!(
                "action(s) exceed the session's granted permission tier and require \
                 human approval: {}",
                escalation_notes.join("; ")
            ),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use car_ir::{Action, ActionType};
    use car_policy::{
        action_fingerprint, ApprovalDecision, ApprovalLedger, PermissionGate, PermissionTier,
    };
    use std::collections::HashMap;

    fn state_write(id: &str, key: &str) -> Action {
        let mut a = Action::new(ActionType::StateWrite);
        a.id = id.to_string();
        a.parameters
            .insert("key".to_string(), serde_json::Value::from(key));
        a.parameters
            .insert("value".to_string(), serde_json::Value::from("v"));
        a.max_retries = 0;
        a
    }

    /// "deploy" is in the classifier's full-access keyword set.
    fn full_access_action(id: &str) -> Action {
        let mut a = Action::new(ActionType::ToolCall);
        a.id = id.to_string();
        a.tool = Some("deploy_service".to_string());
        a.max_retries = 0;
        a
    }

    fn proposal(actions: Vec<Action>) -> ActionProposal {
        ActionProposal {
            id: "p".to_string(),
            source: "test".to_string(),
            actions,
            timestamp: chrono::Utc::now(),
            context: HashMap::new(),
        }
    }

    fn build(
        granted: PermissionTier,
        ledger: ApprovalLedger,
    ) -> (
        PermissionAdmissionGate,
        Arc<RwLock<car_policy::PermissionGate>>,
    ) {
        let gate = Arc::new(RwLock::new(PermissionGate::new(granted)));
        let ledger = Arc::new(RwLock::new(ledger));
        (PermissionAdmissionGate::new(gate.clone(), ledger), gate)
    }

    fn ctx<'a>(
        state: &'a HashMap<String, serde_json::Value>,
        versions: &'a HashMap<String, u64>,
    ) -> GateContext<'a> {
        GateContext {
            session_id: None,
            scope: None,
            state,
            versions,
        }
    }

    #[tokio::test]
    async fn action_within_the_grant_is_allowed() {
        let (gate, _) = build(PermissionTier::SandboxEdit, ApprovalLedger::new());
        let (s, v) = (HashMap::new(), HashMap::new());
        let p = proposal(vec![state_write("a", "k")]);
        assert!(matches!(
            gate.check(&p, &ctx(&s, &v)).await,
            GateOutcome::Allow
        ));
    }

    /// The reported defect: a state_write on a `read_only` session. The gate
    /// escalates it instead of letting it through, and names the action's own
    /// fingerprint so the operator can approve exactly that operation.
    #[tokio::test]
    async fn action_above_the_grant_escalates() {
        let (gate, _) = build(PermissionTier::ReadOnly, ApprovalLedger::new());
        let (s, v) = (HashMap::new(), HashMap::new());
        let action = state_write("a", "k");
        let expected_fp = action_fingerprint(&action);
        let p = proposal(vec![action]);
        match gate.check(&p, &ctx(&s, &v)).await {
            GateOutcome::NeedsApproval {
                actions,
                fingerprint,
                reason,
            } => {
                assert!(actions.contains("a"));
                assert!(fingerprint.starts_with("permission:"));
                assert!(fingerprint.contains(&expected_fp));
                // Required + granted tier and the approvable fingerprint are
                // all in the reason — it is what reaches the caller.
                assert!(reason.contains("sandbox_edit"), "{reason}");
                assert!(reason.contains("read_only"), "{reason}");
                assert!(reason.contains(&expected_fp), "{reason}");
            }
            other => panic!("expected escalation, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn a_recorded_rejection_denies() {
        let action = state_write("a", "k");
        let mut ledger = ApprovalLedger::new();
        ledger
            .record_decision(
                &action_fingerprint(&action),
                PermissionTier::SandboxEdit,
                ApprovalDecision::Rejected,
                "operator",
                "not this one",
                None,
            )
            .expect("in-memory ledger cannot fail");
        // Granted tier would otherwise cover it — the standing rejection wins.
        let (gate, _) = build(PermissionTier::FullAccess, ledger);
        let (s, v) = (HashMap::new(), HashMap::new());
        let p = proposal(vec![action]);
        match gate.check(&p, &ctx(&s, &v)).await {
            GateOutcome::Reject { blocked, reason } => {
                assert!(blocked.contains("a"));
                assert!(reason.contains("previously rejected"), "{reason}");
            }
            other => panic!("expected reject, got {other:?}"),
        }
    }

    /// The approve → re-submit loop: a recorded approval makes the very same
    /// proposal admissible, with no escalation raised at all.
    #[tokio::test]
    async fn a_recorded_approval_allows() {
        let action = state_write("a", "k");
        let mut ledger = ApprovalLedger::new();
        ledger
            .record_decision(
                &action_fingerprint(&action),
                PermissionTier::SandboxEdit,
                ApprovalDecision::Approved,
                "operator",
                "reviewed",
                None,
            )
            .expect("in-memory ledger cannot fail");
        let (gate, _) = build(PermissionTier::ReadOnly, ledger);
        let (s, v) = (HashMap::new(), HashMap::new());
        let p = proposal(vec![action]);
        assert!(matches!(
            gate.check(&p, &ctx(&s, &v)).await,
            GateOutcome::Allow
        ));
    }

    /// Only the offending actions are named — but the whole proposal is held,
    /// because partial execution of a plan whose later steps need a human is
    /// the outcome admission exists to prevent.
    #[tokio::test]
    async fn multi_action_escalation_names_only_the_offenders() {
        let (gate, _) = build(PermissionTier::SandboxEdit, ApprovalLedger::new());
        let (s, v) = (HashMap::new(), HashMap::new());
        let ok = state_write("a1", "k");
        let bad1 = full_access_action("a2");
        let bad2 = full_access_action("a3");
        let fp2 = action_fingerprint(&bad2);
        let p = proposal(vec![ok, bad1, bad2]);
        match gate.check(&p, &ctx(&s, &v)).await {
            GateOutcome::NeedsApproval {
                actions,
                fingerprint,
                ..
            } => {
                assert!(!actions.contains("a1"), "the in-grant write is not blamed");
                assert!(actions.contains("a2"));
                assert!(actions.contains("a3"));
                // a2 and a3 are the same operation, so they share a
                // fingerprint — the composite dedupes it.
                assert_eq!(fingerprint, format!("permission:{fp2}"));
            }
            other => panic!("expected escalation, got {other:?}"),
        }
    }

    /// A live `permission.set_tier` is visible on the next check, because the
    /// gate holds the session's gate rather than a snapshot of its tier.
    #[tokio::test]
    async fn tier_changes_take_effect_on_the_shared_gate() {
        let (gate, shared) = build(PermissionTier::ReadOnly, ApprovalLedger::new());
        let (s, v) = (HashMap::new(), HashMap::new());
        let p = proposal(vec![state_write("a", "k")]);
        assert!(matches!(
            gate.check(&p, &ctx(&s, &v)).await,
            GateOutcome::NeedsApproval { .. }
        ));
        shared
            .write()
            .await
            .set_granted_tier(PermissionTier::SandboxEdit);
        assert!(matches!(
            gate.check(&p, &ctx(&s, &v)).await,
            GateOutcome::Allow
        ));
    }
}