car_server_core/

supervision.rs

//! The admission gate, published as an out-of-process subscription.
//!
//! CAR already blocks before execution: [`car_engine::AdmissionGate`] runs on
//! every admitted proposal, conjunctively and fail-closed. What it could not do
//! is let an **external** process read that stream and decide. Every gate was
//! in-process — a Rust trait impl compiled into the daemon — so supervising a
//! third-party agent meant forking CAR, not connecting to it.
//!
//! This module closes that. [`SupervisionGate`] is an `AdmissionGate` that
//! publishes each proposal as a trimmed [`SupervisionIntent`] to subscribed WS
//! clients and parks until one of them decides, or the timeout fires. The
//! supervisor is any process that can hold a WebSocket — a model-driven meta
//! agent, a rules engine, a human console.
//!
//! ## What this is NOT — read before extending
//!
//! Shepherd (arXiv 2605.10913, Appendix E) supervises with a three-verb
//! vocabulary: `inject` (append one user message, session and tool trail
//! intact), `handoff` (abort the session, restart on the same identity),
//! `discard` (handoff plus roll the scope back). **None of the three are here**,
//! and the reason is not oversight:
//!
//! - `inject` mutates a live conversation. This gate sits at *proposal
//!   admission* in the engine, which has no conversation to append to. Its
//!   natural home is the assistant loop, a different seam.
//! - `handoff` needs session abort-and-restart as a primitive. CAR has no such
//!   primitive today.
//! - `discard` is `handoff` plus scope rollback, which is proposal item 4 —
//!   measured on 2026-08-09 and found constrained on macOS (`clonefile` is flat
//!   in bytes but linear in file count, so restore is only viable for a bounded,
//!   artifact-free working set).
//!
//! So the vocabulary here is the one the admission seam can actually honour:
//! allow, deny, escalate-to-human. [`SupervisionDecision`] is
//! `#[non_exhaustive]` so the Shepherd verbs can join it without breaking
//! callers — but a verb that cannot be routed is worse than an absent one, and
//! this crate does not ship promises it cannot keep.
//!
//! ## Two properties that make it safe to leave registered
//!
//! **Inert when nobody is subscribed.** `AdmissionGate`'s contract says
//! implementations "must be cheap and side-effect-free: they run on the hot path
//! of every admitted proposal". A network round trip is neither. So the gate
//! checks its subscriber set first and returns `Allow` without allocating an
//! intent when no supervisor matches. The cost is a read-lock on an empty map.
//!
//! **Fail-closed on timeout.** A supervisor that dies mid-decision must not
//! become an open door. An unanswered intent is a `Reject`, consistent with the
//! rest of admission, where an unresolved escalation blocks rather than passes.
//!
//! ## Why the payload is trimmed
//!
//! Shepherd's cost finding is that supervision is affordable *because it is
//! batched and trimmed* — ~150 bytes per tool call, keeping a supervisor prompt
//! under 3K tokens with 10 agents in flight, not because the meta-model is
//! cheap. An intent therefore carries action identity, type, tool, and
//! reversibility, plus a **digest** of the parameters rather than the parameters
//! themselves. A supervisor that needs the full text asks for it out of band;
//! the default path stays small enough to put in a prompt.
//!
//! The batching half is [`SupervisionRegistry::pending`], surfaced as
//! `supervision.pending`: every intent currently parked, so a supervisor can
//! make one model call across all of them and then issue N decisions. Delivery
//! stays one notification per intent — a proposal blocks on *its own* verdict,
//! and coalescing deliveries would only add latency to the first intent in a
//! tick without reducing the number of decisions owed.

use std::collections::{HashMap, HashSet};
use std::sync::Arc;

use car_engine::admission::{AdmissionGate, GateContext, GateOutcome};
use car_ir::{Action, ActionProposal};
use car_server_types::host::EventSubscriber;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tokio::sync::{Mutex, Notify};

/// How long an intent waits for a decision before failing closed.
pub const DEFAULT_DECISION_TIMEOUT_MS: u64 = 30_000;

/// Cap on how many intents may be parked at once. A supervisor that stops
/// answering must not let the pending table grow without bound; past this,
/// new intents are rejected outright rather than queued behind a dead
/// supervisor.
pub const MAX_PENDING_INTENTS: usize = 256;

/// One action, trimmed for a supervisor prompt. See the module docs on why the
/// parameters are digested rather than carried.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct IntentAction {
    pub id: String,
    /// `car_ir::ActionType`'s serde label (`tool_call`, `state_write`, …).
    pub action_type: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub tool: Option<String>,
    /// `reversible` / `compensable` / `irreversible` — the axis that answers
    /// "can this be undone?", which is what a supervisor deciding whether to
    /// intervene actually needs. Distinct from *who may authorize it*.
    pub reversibility: String,
    /// Parameter key names only, sorted. Names are cheap and often decisive
    /// (`command` vs `path`); values can be megabytes.
    #[serde(default)]
    pub parameter_keys: Vec<String>,
    /// Stable digest of the full parameter map, so a supervisor can correlate
    /// or cache a decision without the daemon shipping the values.
    pub parameters_digest: String,
}

impl IntentAction {
    fn from_action(action: &Action) -> Self {
        let mut parameter_keys: Vec<String> = action.parameters.keys().cloned().collect();
        parameter_keys.sort();
        Self {
            id: action.id.clone(),
            action_type: action_type_label(action),
            tool: action.tool.clone(),
            reversibility: reversibility_label(action),
            parameters_digest: digest_parameters(&action.parameters),
            parameter_keys,
        }
    }
}

/// Serde label for the action type, taken from its own serialization so this
/// never drifts from the wire form the rest of the system uses.
fn action_type_label(action: &Action) -> String {
    match serde_json::to_value(&action.action_type) {
        Ok(Value::String(s)) => s,
        Ok(other) => other.to_string(),
        Err(_) => "unknown".to_string(),
    }
}

fn reversibility_label(action: &Action) -> String {
    match serde_json::to_value(action.reversibility) {
        Ok(Value::String(s)) => s,
        _ => "irreversible".to_string(),
    }
}

/// A stable, order-independent digest of a parameter map.
///
/// Deliberately not a cryptographic commitment — it exists so a supervisor can
/// say "I have seen this exact call before", not to prove anything to anyone.
/// Order independence matters because `HashMap` iteration order is not stable
/// across runs, and a digest that changed run to run would make every cached
/// decision miss.
fn digest_parameters(parameters: &HashMap<String, Value>) -> String {
    let mut entries: Vec<(&String, String)> = parameters
        .iter()
        .map(|(k, v)| (k, serde_json::to_string(v).unwrap_or_default()))
        .collect();
    entries.sort_by(|a, b| a.0.cmp(b.0));

    // FNV-1a, 64-bit. Small, dependency-free, and adequate for a correlation
    // key. If this ever needs to resist an adversary, it is the wrong function.
    let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
    for (key, value) in entries {
        for byte in key.as_bytes().iter().chain(b"=").chain(value.as_bytes()) {
            hash ^= *byte as u64;
            hash = hash.wrapping_mul(0x1000_0000_01b3);
        }
    }
    format!("{hash:016x}")
}

/// A proposal awaiting a supervisor's verdict.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct SupervisionIntent {
    /// Identity a decision is issued against.
    pub id: String,
    pub proposal_id: String,
    /// Where the proposal came from — `ActionProposal::source`.
    pub source: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub session_id: Option<String>,
    /// Caller/tenant identity, when the runtime is scoped.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub scope: Option<String>,
    pub actions: Vec<IntentAction>,
    /// The least-reversible contract across the batch — the batch's own
    /// rollback contract, so a supervisor can triage on one field.
    pub reversibility: String,
    pub created_at: DateTime<Utc>,
}

/// What a supervisor decides about an intent.
///
/// `#[non_exhaustive]`: Shepherd's `inject`/`handoff`/`discard` may join this
/// once the seams they need exist. See the module docs for why they are absent
/// rather than stubbed.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum SupervisionDecision {
    /// Proceed. Other gates still apply — one supervisor's allow is not a
    /// bypass of the rest of admission.
    Allow,
    /// Refuse. Nothing in the proposal runs.
    Deny { reason: String },
    /// Hand the question to the human approval ledger rather than answering it.
    /// Lets a supervisor be conservative about what it is unsure of without
    /// either blocking the work or waving it through.
    Escalate { reason: String },
}

/// Which proposals a supervisor wants to see. An empty filter means all of
/// them.
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
pub struct SupervisionFilter {
    /// Only intents whose proposal touches one of these tools.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub tools: Option<Vec<String>>,
    /// Only intents from these sessions.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub sessions: Option<Vec<String>>,
    /// Only intents at or below this rollback contract — e.g. `compensable`
    /// also matches `irreversible`, because a supervisor asking to see the
    /// risky ones means "this risky and worse".
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub min_reversibility: Option<String>,
}

impl SupervisionFilter {
    fn matches(&self, intent: &SupervisionIntent) -> bool {
        if let Some(sessions) = &self.sessions {
            match &intent.session_id {
                Some(id) if sessions.iter().any(|s| s == id) => {}
                _ => return false,
            }
        }
        if let Some(tools) = &self.tools {
            let hit = intent
                .actions
                .iter()
                .filter_map(|a| a.tool.as_ref())
                .any(|t| tools.iter().any(|w| w == t));
            if !hit {
                return false;
            }
        }
        if let Some(min) = &self.min_reversibility {
            if severity(&intent.reversibility) < severity(min) {
                return false;
            }
        }
        true
    }
}

/// Ordering on the rollback contract: reversible < compensable < irreversible.
/// Matches `car_ir::Reversibility`'s own severity ordering. An unrecognised
/// label sorts as the most severe, so a future variant is over-reported to
/// supervisors rather than silently filtered out of view.
fn severity(label: &str) -> u8 {
    match label {
        "reversible" => 0,
        "compensable" => 1,
        _ => 2,
    }
}

struct Supervisor {
    filter: SupervisionFilter,
    channel: Arc<dyn EventSubscriber>,
}

struct PendingIntent {
    intent: SupervisionIntent,
    decision: Option<SupervisionDecision>,
    notify: Arc<Notify>,
}

/// Subscribers, parked intents, and the decisions that release them.
pub struct SupervisionRegistry {
    supervisors: Mutex<HashMap<String, Supervisor>>,
    pending: Mutex<HashMap<String, PendingIntent>>,
    timeout: std::time::Duration,
}

impl Default for SupervisionRegistry {
    fn default() -> Self {
        Self::new(std::time::Duration::from_millis(
            DEFAULT_DECISION_TIMEOUT_MS,
        ))
    }
}

impl SupervisionRegistry {
    pub fn new(timeout: std::time::Duration) -> Self {
        Self {
            supervisors: Mutex::new(HashMap::new()),
            pending: Mutex::new(HashMap::new()),
            timeout,
        }
    }

    pub fn timeout(&self) -> std::time::Duration {
        self.timeout
    }

    /// Register (or re-register, replacing the filter) a supervisor.
    pub async fn subscribe(
        &self,
        client_id: &str,
        filter: SupervisionFilter,
        channel: Arc<dyn EventSubscriber>,
    ) {
        self.supervisors
            .lock()
            .await
            .insert(client_id.to_string(), Supervisor { filter, channel });
    }

    /// Drop a supervisor. Returns whether one was registered.
    ///
    /// Any intent it left parked keeps waiting for its timeout rather than
    /// being released here: another supervisor may still answer, and releasing
    /// on unsubscribe would let a supervisor turn a pending decision into an
    /// allow by disconnecting.
    pub async fn unsubscribe(&self, client_id: &str) -> bool {
        self.supervisors.lock().await.remove(client_id).is_some()
    }

    pub async fn is_subscribed(&self, client_id: &str) -> bool {
        self.supervisors.lock().await.contains_key(client_id)
    }

    pub async fn subscriber_count(&self) -> usize {
        self.supervisors.lock().await.len()
    }

    /// Every currently-parked intent, newest last. The batching half of the
    /// design — one model call can cover all of them.
    pub async fn pending(&self) -> Vec<SupervisionIntent> {
        let mut intents: Vec<SupervisionIntent> = self
            .pending
            .lock()
            .await
            .values()
            .map(|p| p.intent.clone())
            .collect();
        intents.sort_by(|a, b| a.created_at.cmp(&b.created_at).then(a.id.cmp(&b.id)));
        intents
    }

    /// Record a decision and wake whoever is parked on it.
    ///
    /// `Err` when the intent is unknown — already decided, already timed out,
    /// or never existed. Deliberately an error rather than a silent no-op: a
    /// supervisor that believes it denied something needs to hear that the
    /// denial did not land.
    pub async fn decide(
        &self,
        intent_id: &str,
        decision: SupervisionDecision,
    ) -> Result<(), String> {
        let mut pending = self.pending.lock().await;
        let entry = pending
            .get_mut(intent_id)
            .ok_or_else(|| format!("no pending supervision intent '{intent_id}'"))?;
        if entry.decision.is_some() {
            return Err(format!("intent '{intent_id}' was already decided"));
        }
        entry.decision = Some(decision);
        entry.notify.notify_waiters();
        Ok(())
    }

    /// Publish an intent to every matching supervisor and park until one
    /// decides or the timeout fires.
    ///
    /// `None` means the timeout won — the caller fails closed.
    async fn publish_and_wait(&self, intent: SupervisionIntent) -> Option<SupervisionDecision> {
        let targets: Vec<Arc<dyn EventSubscriber>> = {
            let supervisors = self.supervisors.lock().await;
            supervisors
                .values()
                .filter(|s| s.filter.matches(&intent))
                .map(|s| s.channel.clone())
                .collect()
        };
        if targets.is_empty() {
            return Some(SupervisionDecision::Allow);
        }

        let intent_id = intent.id.clone();
        let notify = Arc::new(Notify::new());
        {
            let mut pending = self.pending.lock().await;
            if pending.len() >= MAX_PENDING_INTENTS {
                return None;
            }
            pending.insert(
                intent_id.clone(),
                PendingIntent {
                    intent: intent.clone(),
                    decision: None,
                    notify: notify.clone(),
                },
            );
        }

        // Subscribe to the wakeup BEFORE sending, so a supervisor that answers
        // synchronously inside its send cannot land the decision in the window
        // between publish and park.
        let waiter = notify.notified();
        tokio::pin!(waiter);

        if let Ok(frame) = serde_json::to_string(&serde_json::json!({
            "jsonrpc": "2.0",
            "method": "supervision.intent",
            "params": intent,
        })) {
            for target in targets {
                target.send_text(frame.clone()).await;
            }
        }

        let outcome = tokio::time::timeout(self.timeout, waiter).await;

        let mut pending = self.pending.lock().await;
        let entry = pending.remove(&intent_id);
        match (outcome, entry) {
            // Take the decision whenever one is recorded, even if the timeout
            // also fired: a verdict that arrived is a verdict, and discarding
            // it on a race would turn a supervisor's `allow` into a block.
            (
                _,
                Some(PendingIntent {
                    decision: Some(d), ..
                }),
            ) => Some(d),
            _ => None,
        }
    }
}

/// The `AdmissionGate` that consults out-of-process supervisors.
pub struct SupervisionGate {
    registry: Arc<SupervisionRegistry>,
}

impl SupervisionGate {
    pub fn new(registry: Arc<SupervisionRegistry>) -> Self {
        Self { registry }
    }

    fn intent_for(proposal: &ActionProposal, ctx: &GateContext<'_>) -> SupervisionIntent {
        SupervisionIntent {
            id: format!("intent-{}", uuid_like()),
            proposal_id: proposal.id.clone(),
            source: proposal.source.clone(),
            session_id: ctx.session_id.map(|s| s.to_string()),
            scope: ctx.scope.map(|s| format!("{s:?}")),
            actions: proposal
                .actions
                .iter()
                .map(IntentAction::from_action)
                .collect(),
            reversibility: match serde_json::to_value(proposal.rollback_contract()) {
                Ok(Value::String(s)) => s,
                _ => "irreversible".to_string(),
            },
            created_at: Utc::now(),
        }
    }
}

fn uuid_like() -> String {
    use std::sync::atomic::{AtomicU64, Ordering};
    static COUNTER: AtomicU64 = AtomicU64::new(0);
    let n = COUNTER.fetch_add(1, Ordering::Relaxed);
    format!("{:x}-{:x}", Utc::now().timestamp_micros(), n)
}

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

    async fn check(&self, proposal: &ActionProposal, ctx: &GateContext<'_>) -> GateOutcome {
        // Inert when nobody is watching — see the module docs. This is the
        // check that keeps the gate honest about `AdmissionGate`'s "must be
        // cheap" contract on the hot path.
        if self.registry.subscriber_count().await == 0 {
            return GateOutcome::Allow;
        }

        let intent = Self::intent_for(proposal, ctx);
        let all_actions: HashSet<String> = proposal.actions.iter().map(|a| a.id.clone()).collect();

        match self.registry.publish_and_wait(intent).await {
            Some(SupervisionDecision::Allow) => GateOutcome::Allow,
            Some(SupervisionDecision::Deny { reason }) => GateOutcome::Reject {
                blocked: all_actions,
                reason: format!("supervisor denied: {reason}"),
            },
            Some(SupervisionDecision::Escalate { reason }) => GateOutcome::NeedsApproval {
                fingerprint: format!("supervision:{}", proposal.id),
                actions: all_actions,
                reason: format!("supervisor escalated: {reason}"),
            },
            // Fail closed. A supervisor that died mid-decision must not become
            // an open door.
            None => GateOutcome::Reject {
                blocked: all_actions,
                reason: format!(
                    "no supervisor decision within {}ms (fail-closed)",
                    self.registry.timeout().as_millis()
                ),
            },
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use car_ir::{ActionType, Reversibility};
    use std::sync::Mutex as StdMutex;

    /// Records what was sent, and can auto-answer to exercise the race where a
    /// decision lands during the publish loop.
    struct Recorder {
        frames: Arc<StdMutex<Vec<String>>>,
        auto: Option<(Arc<SupervisionRegistry>, SupervisionDecision)>,
    }

    #[async_trait::async_trait]
    impl EventSubscriber for Recorder {
        async fn send_text(&self, json: String) {
            self.frames.lock().unwrap().push(json.clone());
            if let Some((registry, decision)) = &self.auto {
                let parsed: Value = serde_json::from_str(&json).unwrap();
                let id = parsed["params"]["id"].as_str().unwrap().to_string();
                let _ = registry.decide(&id, decision.clone()).await;
            }
        }
    }

    fn recorder() -> (Arc<Recorder>, Arc<StdMutex<Vec<String>>>) {
        let frames = Arc::new(StdMutex::new(Vec::new()));
        (
            Arc::new(Recorder {
                frames: frames.clone(),
                auto: None,
            }),
            frames,
        )
    }

    fn proposal(tool: &str, reversibility: Reversibility) -> ActionProposal {
        let mut action = Action::tool_call(tool);
        action.reversibility = reversibility;
        action = action.with_param("path", Value::from("/tmp/x"));
        ActionProposal {
            id: "prop-1".to_string(),
            source: "test".to_string(),
            actions: vec![action],
            timestamp: Utc::now(),
            context: HashMap::new(),
        }
    }

    async fn check(gate: &SupervisionGate, p: &ActionProposal) -> GateOutcome {
        let state = HashMap::new();
        let versions = HashMap::new();
        let ctx = GateContext {
            session_id: Some("sess-1"),
            scope: None,
            state: &state,
            versions: &versions,
        };
        gate.check(p, &ctx).await
    }

    #[tokio::test]
    async fn a_gate_with_no_subscribers_is_inert() {
        let registry = Arc::new(SupervisionRegistry::default());
        let gate = SupervisionGate::new(registry.clone());
        // Would block for the full timeout if it published; the test finishing
        // is itself the assertion that it did not.
        assert!(matches!(
            check(&gate, &proposal("write_file", Reversibility::Irreversible)).await,
            GateOutcome::Allow
        ));
        assert!(registry.pending().await.is_empty());
    }

    #[tokio::test]
    async fn an_allow_decision_admits_the_proposal() {
        let registry = Arc::new(SupervisionRegistry::default());
        let sub = Arc::new(Recorder {
            frames: Arc::new(StdMutex::new(Vec::new())),
            auto: Some((registry.clone(), SupervisionDecision::Allow)),
        });
        registry
            .subscribe("sup-1", SupervisionFilter::default(), sub)
            .await;
        let gate = SupervisionGate::new(registry);
        assert!(matches!(
            check(&gate, &proposal("write_file", Reversibility::Reversible)).await,
            GateOutcome::Allow
        ));
    }

    #[tokio::test]
    async fn a_deny_blocks_every_action_in_the_proposal() {
        let registry = Arc::new(SupervisionRegistry::default());
        let sub = Arc::new(Recorder {
            frames: Arc::new(StdMutex::new(Vec::new())),
            auto: Some((
                registry.clone(),
                SupervisionDecision::Deny {
                    reason: "not on a Friday".to_string(),
                },
            )),
        });
        registry
            .subscribe("sup-1", SupervisionFilter::default(), sub)
            .await;
        let gate = SupervisionGate::new(registry);
        match check(&gate, &proposal("deploy", Reversibility::Irreversible)).await {
            GateOutcome::Reject { blocked, reason } => {
                assert_eq!(blocked.len(), 1);
                assert!(reason.contains("not on a Friday"), "{reason}");
            }
            other => panic!("expected Reject, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn an_escalation_becomes_a_human_approval() {
        let registry = Arc::new(SupervisionRegistry::default());
        let sub = Arc::new(Recorder {
            frames: Arc::new(StdMutex::new(Vec::new())),
            auto: Some((
                registry.clone(),
                SupervisionDecision::Escalate {
                    reason: "unsure".to_string(),
                },
            )),
        });
        registry
            .subscribe("sup-1", SupervisionFilter::default(), sub)
            .await;
        let gate = SupervisionGate::new(registry);
        match check(&gate, &proposal("deploy", Reversibility::Irreversible)).await {
            GateOutcome::NeedsApproval { fingerprint, .. } => {
                assert_eq!(fingerprint, "supervision:prop-1");
            }
            other => panic!("expected NeedsApproval, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn a_silent_supervisor_fails_closed() {
        let registry = Arc::new(SupervisionRegistry::new(std::time::Duration::from_millis(
            60,
        )));
        let (sub, frames) = recorder();
        registry
            .subscribe("sup-1", SupervisionFilter::default(), sub)
            .await;
        let gate = SupervisionGate::new(registry.clone());
        match check(&gate, &proposal("rm", Reversibility::Irreversible)).await {
            GateOutcome::Reject { reason, .. } => {
                assert!(reason.contains("fail-closed"), "{reason}")
            }
            other => panic!("expected fail-closed Reject, got {other:?}"),
        }
        assert_eq!(
            frames.lock().unwrap().len(),
            1,
            "intent should be published once"
        );
        // The parked entry must be reaped, or a dead supervisor leaks memory.
        assert!(registry.pending().await.is_empty());
    }

    #[tokio::test]
    async fn unsubscribing_does_not_release_a_parked_intent_as_allow() {
        // Otherwise a supervisor could turn "deny" into "allow" by disconnecting.
        let registry = Arc::new(SupervisionRegistry::new(std::time::Duration::from_millis(
            60,
        )));
        let (sub, _) = recorder();
        registry
            .subscribe("sup-1", SupervisionFilter::default(), sub)
            .await;
        let gate = SupervisionGate::new(registry.clone());
        let reg = registry.clone();
        tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            reg.unsubscribe("sup-1").await;
        });
        assert!(matches!(
            check(&gate, &proposal("rm", Reversibility::Irreversible)).await,
            GateOutcome::Reject { .. }
        ));
    }

    #[tokio::test]
    async fn deciding_an_unknown_intent_is_an_error_not_a_silent_noop() {
        let registry = SupervisionRegistry::default();
        let err = registry
            .decide("intent-nope", SupervisionDecision::Allow)
            .await
            .unwrap_err();
        assert!(err.contains("no pending supervision intent"), "{err}");
    }

    #[tokio::test]
    async fn an_intent_cannot_be_decided_twice() {
        let registry = Arc::new(SupervisionRegistry::new(std::time::Duration::from_millis(
            200,
        )));
        let (sub, _) = recorder();
        registry
            .subscribe("sup-1", SupervisionFilter::default(), sub)
            .await;
        let gate = SupervisionGate::new(registry.clone());
        let reg = registry.clone();
        let handle =
            tokio::spawn(
                async move { check(&gate, &proposal("rm", Reversibility::Reversible)).await },
            );
        // Wait for the intent to park.
        let id = loop {
            if let Some(i) = reg.pending().await.first() {
                break i.id.clone();
            }
            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        };
        reg.decide(&id, SupervisionDecision::Allow).await.unwrap();
        let second = reg.decide(&id, SupervisionDecision::Allow).await;
        assert!(
            second.is_err(),
            "a decided intent must not accept a second verdict"
        );
        assert!(matches!(handle.await.unwrap(), GateOutcome::Allow));
    }

    #[tokio::test]
    async fn a_filter_that_does_not_match_leaves_the_proposal_unsupervised() {
        let registry = Arc::new(SupervisionRegistry::new(std::time::Duration::from_millis(
            60,
        )));
        let (sub, frames) = recorder();
        registry
            .subscribe(
                "sup-1",
                SupervisionFilter {
                    tools: Some(vec!["deploy".to_string()]),
                    ..Default::default()
                },
                sub,
            )
            .await;
        let gate = SupervisionGate::new(registry);
        // Subscriber count is non-zero, so the gate does build an intent — but
        // no supervisor matches it, so it must admit rather than fail closed.
        assert!(matches!(
            check(&gate, &proposal("read_file", Reversibility::Reversible)).await,
            GateOutcome::Allow
        ));
        assert!(frames.lock().unwrap().is_empty());
    }

    #[tokio::test]
    async fn min_reversibility_matches_this_severity_and_worse() {
        let registry = Arc::new(SupervisionRegistry::new(std::time::Duration::from_millis(
            60,
        )));
        let (sub, frames) = recorder();
        registry
            .subscribe(
                "sup-1",
                SupervisionFilter {
                    min_reversibility: Some("compensable".to_string()),
                    ..Default::default()
                },
                sub,
            )
            .await;
        let gate = SupervisionGate::new(registry);
        // reversible is BELOW the floor — not shown.
        let _ = check(&gate, &proposal("read", Reversibility::Reversible)).await;
        assert!(frames.lock().unwrap().is_empty());
        // irreversible is ABOVE it — shown.
        let _ = check(&gate, &proposal("rm", Reversibility::Irreversible)).await;
        assert_eq!(frames.lock().unwrap().len(), 1);
    }

    #[test]
    fn the_parameter_digest_is_order_independent_and_value_sensitive() {
        let mut a = HashMap::new();
        a.insert("x".to_string(), Value::from(1));
        a.insert("y".to_string(), Value::from("two"));
        let mut b = HashMap::new();
        b.insert("y".to_string(), Value::from("two"));
        b.insert("x".to_string(), Value::from(1));
        assert_eq!(digest_parameters(&a), digest_parameters(&b));

        let mut c = HashMap::new();
        c.insert("x".to_string(), Value::from(2));
        c.insert("y".to_string(), Value::from("two"));
        assert_ne!(digest_parameters(&a), digest_parameters(&c));
    }

    #[test]
    fn an_intent_carries_key_names_but_never_parameter_values() {
        let mut action = Action::tool_call("run");
        action = action.with_param("command", Value::from("rm -rf /secret/path"));
        let trimmed = IntentAction::from_action(&action);
        let json = serde_json::to_string(&trimmed).unwrap();
        assert!(json.contains("command"), "key names are useful and cheap");
        assert!(
            !json.contains("secret"),
            "parameter VALUES must not ride along: {json}"
        );
    }

    #[test]
    fn an_unknown_reversibility_label_sorts_as_most_severe() {
        // A future variant must be over-reported to supervisors, never
        // silently filtered out of view.
        assert_eq!(severity("something_new"), severity("irreversible"));
    }

    #[test]
    fn the_decision_wire_form_is_tagged_and_snake_case() {
        let json = serde_json::to_string(&SupervisionDecision::Deny {
            reason: "no".to_string(),
        })
        .unwrap();
        assert_eq!(json, r#"{"kind":"deny","reason":"no"}"#);
        let parsed: SupervisionDecision = serde_json::from_str(r#"{"kind":"allow"}"#).unwrap();
        assert_eq!(parsed, SupervisionDecision::Allow);
    }

    #[test]
    fn action_type_and_reversibility_labels_come_from_serde_not_a_second_table() {
        let mut action = Action::new(ActionType::StateWrite);
        action.reversibility = Reversibility::Compensable;
        let trimmed = IntentAction::from_action(&action);
        assert_eq!(trimmed.action_type, "state_write");
        assert_eq!(trimmed.reversibility, "compensable");
    }
}