car-policy 0.55.0

//! Permission tiers, risk classification, and human-in-the-loop approval
//! as **durable harness state**.
//!
//! Motivated by the "Code as Agent Harness" survey (arXiv 2605.18747)
//! §3.4.3 and §5.2.5: a harness must act as a *safety governor* between
//! model intent and real-world consequence, not merely a tool executor.
//! Two ideas from that section are made concrete here:
//!
//! 1. **A multi-tier permission model.** Every action is classified by
//!    risk into [`PermissionTier::ReadOnly`], [`PermissionTier::SandboxEdit`],
//!    or [`PermissionTier::FullAccess`]. The session holds a *granted*
//!    standing tier; an action whose required tier exceeds it cannot run
//!    autonomously.
//! 2. **Human-in-the-loop as durable, auditable state.** Top-tier
//!    (externally-consequential / irreversible) actions are gated behind a
//!    mandatory human decision. That decision is not an ephemeral prompt:
//!    it is recorded in an [`ApprovalLedger`] — who approved or rejected
//!    what, when, on what evidence — that persists and feeds back into
//!    every later evaluation. "Each approval, rejection, policy exception,
//!    or reviewer correction should become durable harness state."
//!
//! A **third** idea arrived later, from a different paper, and it is the one
//! that corrects a mistake in the first two:
//!
//! 3. **"Who may authorize this?" and "can this be undone?" are two axes,
//!    not one.** [`PermissionTier`] answers only the first. It used to be
//!    documented as though it answered both — `SandboxEdit` as "reversible
//!    local mutation", `FullAccess` as "externally-consequential **or**
//!    irreversible" — and that `or` fused a `git push` (undo by force-pushing
//!    the prior ref), a production `INSERT` (undo by deleting the row), and a
//!    charged card (no undo at all) onto one rung. The rollback contract now
//!    has its own type, [`car_ir::Reversibility`], classified here by
//!    [`classify_reversibility`] from its own independently curated keyword
//!    sets. The two are reported side by side rather than collapsed. See
//!    `docs/proposals/shepherd-substrate-adoption.md`, "The finding worth
//!    acting on first: two axes, one enum".
//!
//! The classifier and gate are pure and synchronous; the engine bridges
//! [`PermissionGate`] into its async authorization pipeline (see
//! `car-engine`'s `TierPermissionHandler`).

use car_ir::{Action, ActionType, Reversibility};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::io::Write as _;
use std::path::PathBuf;

/// Permission tiers — **who may authorize this action?** — ordered by the
/// authority an action demands (survey §3.4.3). The `Ord` derive makes
/// `ReadOnly < SandboxEdit < FullAccess`, so "does the granted tier cover the
/// required tier?" is a single `>=`.
///
/// # This ladder does not answer "can it be undone?"
///
/// It used to read as though it did. `SandboxEdit` was documented as
/// "reversible local mutation" and `FullAccess` as "externally-consequential
/// **or** irreversible", and that `or` quietly fused two independent
/// questions onto one rung: a `git push`, a production `INSERT`, and a charged
/// card are all `FullAccess` and have three different rollback contracts.
/// Collapsed, the runtime had two options and no third — gate every
/// `FullAccess` action identically (approval fatigue, and the predictable
/// response is that someone turns the gate off), or relax the tier and lose
/// the permanent cases along with the recoverable ones.
///
/// The rollback contract is now [`car_ir::Reversibility`], classified by
/// [`classify_reversibility`] from its own keyword sets, and the two axes are
/// reported side by side. **Nothing about this enum changed when that axis
/// landed** — not its ordering, not its variants, not what
/// [`RiskClassifier::classify`] returns for any action. Only this
/// documentation changed, because it had been describing the other axis.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PermissionTier {
    /// Observation only — state reads, retrieval, static inspection, log
    /// analysis. Nothing is mutated, so the only authority at stake is the
    /// authority to look.
    ReadOnly,
    /// Mutation whose blast radius stops at the session's own workspace —
    /// state writes, local patches, sandboxed tool calls, temporary dependency
    /// installs inside an isolated workspace. Authorizable by a standing grant
    /// precisely because nothing crosses the sandbox boundary.
    SandboxEdit,
    /// Externally consequential — effects that cross the sandbox boundary:
    /// network egress, credentials/secrets, deployment, destructive filesystem
    /// or VCS operations, financial/medical actions, physical control. Needs
    /// the top standing grant and, by default (`require_approval_at`), a human
    /// decision as well.
    ///
    /// Whether such an action can afterwards be *undone* is a separate
    /// question this rung does not answer: it holds `read_secret` (nothing to
    /// undo), `git push` (force-push the prior ref), and a charged card (no
    /// undo at all) alike. Ask [`classify_reversibility`] for that.
    FullAccess,
}

impl PermissionTier {
    /// Does a session granted `self` cover an action requiring `required`?
    pub fn covers(self, required: PermissionTier) -> bool {
        self >= required
    }

    pub fn as_str(self) -> &'static str {
        match self {
            PermissionTier::ReadOnly => "read_only",
            PermissionTier::SandboxEdit => "sandbox_edit",
            PermissionTier::FullAccess => "full_access",
        }
    }

    pub fn from_str_opt(s: &str) -> Option<PermissionTier> {
        match s {
            "read_only" | "readonly" | "read" => Some(PermissionTier::ReadOnly),
            "sandbox_edit" | "sandbox" | "edit" => Some(PermissionTier::SandboxEdit),
            "full_access" | "full" => Some(PermissionTier::FullAccess),
            _ => None,
        }
    }
}

/// Substrings that mark an action as [`PermissionTier::FullAccess`] when they
/// appear **on the command surface** — the reconstructed `command`/`args`
/// command line (see [`command_line`]). Conservative and additive there: the
/// cost of over-classifying a command line is an approval prompt; the cost of
/// under-classifying one is an ungated consequential action.
///
/// # Why this list is scoped to the command line and not the whole action
///
/// It used to be substring-scanned over [`action_haystack`] — the entire
/// flattened parameter blob — and it contains tokens that are ordinary English
/// and ordinary JSON: `http`, `send`, `apply`, `format`, `token`, `release`,
/// `request`, `network`, `fetch`, `transfer`. Against free text every one of
/// those fires constantly: a URL in a `url` parameter matched `http`, a search
/// for "release notes" matched `release`, a `format_date` call matched
/// `format`. That was survivable while the classification was advisory, and
/// stopped being survivable at Parslee-ai/car#915, which made a `full_access`
/// classification mandatory HITL at `proposal.submit` — on an automated
/// connection with no approver, a false positive became a hard rejection
/// (Parslee-ai/car#917).
///
/// The sibling list [`FULL_ACCESS_NAME_SEGMENTS`] had already reached this
/// conclusion for tool *names* and deliberately excludes the same tokens. The
/// reasoning simply had not been carried across. On a command line, though,
/// they mean what the list says they mean — `curl https://…`, `terraform
/// apply`, `git push` — so the list is kept verbatim and re-aimed rather than
/// pruned. **For a command-shaped action the classification is byte-for-byte
/// what it was before #917**; only non-command parameters changed.
///
/// The other three signals that reach the rest of the action are
/// [`tool_name_is_full_access`] (whole name segments),
/// [`CREDENTIAL_PARAM_KEYS`] (parameter keys, not values), and
/// [`FULL_ACCESS_TEXT_PHRASES`] (the narrow always-dangerous phrases).
const FULL_ACCESS_COMMAND_KEYWORDS: &[&str] = &[
    // Deploy / publish / release
    "deploy",
    "publish",
    "release",
    "kubectl",
    "terraform",
    "helm",
    "docker push",
    "npm publish",
    "cargo publish",
    "aws ",
    "gcloud",
    "az ",
    "apply",
    "rollout",
    // Credentials / secrets
    "credential",
    "secret",
    "token",
    "password",
    "api_key",
    "apikey",
    "ssh",
    "private key",
    "private_key",
    // Destructive filesystem / VCS
    "delete",
    "destroy",
    "drop",
    "drop table",
    "delete from",
    "truncate",
    "rm ",
    "rmdir",
    "unlink",
    "mkfs",
    "dd ",
    "format",
    "wipe",
    "git push",
    "push",
    "force-push",
    "force_push",
    "reset --hard",
    "git clean",
    "git reset",
    // Network egress
    "network",
    "http",
    "https",
    "curl",
    "wget",
    "fetch",
    "request",
    "egress",
    "upload",
    "download",
    // Money / messaging
    "payment",
    "charge",
    "refund",
    "transfer",
    "wire",
    "email",
    "send",
    "sms",
    // Privilege
    "sudo",
    "chmod",
    "chown",
    "setuid",
];

/// Phrases dangerous enough to escalate **wherever** they appear in an action —
/// including inside a message body, a document, a SQL string, or a diff.
///
/// This is the free-text half of the split described on
/// [`FULL_ACCESS_COMMAND_KEYWORDS`], and every entry earns its place by being a
/// token *sequence* that is essentially never ordinary prose. A single common
/// word (`send`, `release`, `format`) is exactly what this list must not
/// contain; `drop table` and `rm -rf` are unambiguous no matter what parameter
/// carries them.
///
/// The direction of error is still conservative — a `curl` example inside a
/// README body escalates, which costs an approval — but the *rate* is what
/// #917 is about, and prose does not accidentally spell `reset --hard`.
///
/// Two families are load-bearing rather than belt-and-braces, because the
/// command surface cannot see them:
/// - **SQL under a `sql`/`query` parameter.** `query` is not a
///   [`COMMAND_KEYS`] entry, so `{"tool": "run_sql", "query": "DROP TABLE
///   users"}` has an empty command line and a benign tool name. `drop table`
///   here is the only signal there is.
/// - **A script passed as content.** `{"tool": "write_file", "contents": "rm
///   -rf /"}` likewise reaches nothing else.
const FULL_ACCESS_TEXT_PHRASES: &[&str] = &[
    // Destructive SQL — a bare `drop`/`delete` is far too common, the
    // statement head is not.
    "drop table",
    "drop database",
    "drop schema",
    "delete from",
    "truncate table",
    // Destructive shell.
    "rm -rf",
    "rm -fr",
    "rm -r ",
    "mkfs",
    "shred -",
    "dd if=",
    // Privilege escalation — always a command, never prose.
    "sudo ",
    "chmod ",
    "chown ",
    "setuid",
    // VCS history rewriting / publishing.
    "git push",
    "git reset",
    "git clean",
    "reset --hard",
    "force-push",
    "force_push",
    // Publish / deploy toolchains.
    "docker push",
    "npm publish",
    "cargo publish",
    "kubectl ",
    "terraform ",
    "helm ",
    "aws ",
    "gcloud ",
    // Network egress invoked as a command.
    "curl ",
    "wget ",
    // Credential material appearing literally.
    "private key",
    "private_key",
    "-----begin",
];

/// Parameter **keys** whose presence means the action handles credential
/// material, whatever the value happens to be.
///
/// This is the "parameter-key awareness" half of the #917 fix, and it is what
/// lets the broad value-scan be narrowed without losing the credential signal:
/// `{"api_key": "…"}` is a full-access action because of the key, so `token`
/// and `secret` no longer have to be substring-hunted through every value in
/// the payload to catch it.
///
/// Matched two ways against each key (and recursively against nested object
/// keys), both boundary-aware, never as a raw substring:
/// 1. the key with `_`/`-` removed, whole (`api_key` → `apikey`);
/// 2. any single [`name_segments`] segment against [`CREDENTIAL_KEY_SEGMENTS`].
///
/// The pair is what keeps `max_tokens` — in virtually every LLM tool call in
/// existence — from reading as a credential: `maxtokens` is not in this set,
/// and `tokens` is deliberately not in [`CREDENTIAL_KEY_SEGMENTS`]. A bare
/// `token` key is, since that one really is the credential itself.
const CREDENTIAL_PARAM_KEYS: &[&str] = &[
    "token",
    "apikey",
    "apitoken",
    "accesstoken",
    "accesskey",
    "authtoken",
    "authorization",
    "bearer",
    "bearertoken",
    "clientsecret",
    "privatekey",
    "refreshtoken",
    "secretkey",
    "sessiontoken",
    "signingkey",
];

/// Path fragments naming credential material *on disk*, matched against
/// [`path_parameter_haystack`] — the paths an action declares as its targets.
///
/// The third place a credential signal can live, after the tool name and the
/// parameter keys. Before Parslee-ai/car#917 this was covered incidentally: the
/// broad keyword list contained bare `ssh`, `secret`, and `credential`, so
/// `{"path": "~/.ssh/id_rsa"}` escalated because the whole payload was
/// substring-scanned. Narrowing the value scan removed that by accident, and
/// `car-server-core`'s `agent_permissions` tests caught it — reading or writing
/// someone's private key is exactly the action that should need the top grant.
///
/// Scoped to declared target paths rather than the whole action on purpose: a
/// diff or a README that *mentions* `~/.ssh` is not an action against it, and
/// re-scanning free text is the mistake this issue exists to undo.
const CREDENTIAL_PATH_FRAGMENTS: &[&str] = &[
    // SSH / GPG key material
    "ssh",
    "id_rsa",
    "id_dsa",
    "id_ecdsa",
    "id_ed25519",
    ".gnupg",
    "secring",
    // Cloud + registry credential files
    ".aws/credentials",
    ".aws\\credentials",
    ".kube/config",
    ".kube\\config",
    ".docker/config.json",
    ".netrc",
    ".pgpass",
    ".npmrc",
    "service-account",
    "service_account",
    // Generic credential-bearing names and key/cert extensions
    "credential",
    "secret",
    "password",
    "keychain",
    ".pem",
    ".p12",
    ".pfx",
    ".env",
];

/// Key segments that name credential material on their own, at any position in
/// a parameter key (`secret_name`, `db_password`, `svc_credentials`).
///
/// Strictly the words that carry the meaning alone. `token`/`tokens` are
/// excluded on purpose — see [`CREDENTIAL_PARAM_KEYS`] for `max_tokens` — as is
/// `key`, which is the single most common object key there is.
const CREDENTIAL_KEY_SEGMENTS: &[&str] = &[
    "secret",
    "secrets",
    "credential",
    "credentials",
    "password",
    "passwd",
    "passphrase",
];

/// Tool-name segments that signal an irreversible / externally-consequential
/// capability.
///
/// Curated for IDENTIFIER matching (whole snake/camel segments), NOT command
/// text like [`FULL_ACCESS_COMMAND_KEYWORDS`]. Two deliberate differences from
/// that list:
/// 1. Short verbs that collide with benign tool names as substrings are
///    EXCLUDED — `http` (`http_get` is a read), `token` (`count_tokens`,
///    `tokenize`), `request` (`request_id`), `fetch` (`prefetch`), `apply`
///    (`apply_template`), `network`, `format`, `transfer`. A substring matcher
///    would mis-route every one of those to quality-first.
/// 2. The space-bearing command keywords (`git push`, `rm -rf`, `git reset`)
///    are represented by their bare COMMAND segments (`push`, `rm`, `reset`, …)
///    so they actually fire on an identifier — a substring scan of the
///    command list never could (no tool name contains the literal `"rm "`).
///
/// Since Parslee-ai/car#917 this list is not merely the analogue used by the
/// in-process loops: it *is* the tool-name signal inside
/// [`RiskClassifier::classify`], which is what makes the two agree. Before that
/// they disagreed on every name in this doc comment — `classify` substring-
/// scanned the command list over the name and called `http_get` full-access
/// while `tool_name_is_full_access` called it benign.
const FULL_ACCESS_NAME_SEGMENTS: &[&str] = &[
    // deploy / publish / release
    "deploy",
    "publish",
    "release",
    "kubectl",
    "terraform",
    "helm",
    "rollout",
    // credentials / secrets
    "credential",
    "credentials",
    "secret",
    "secrets",
    "password",
    "passwd",
    // destructive filesystem / VCS
    "delete",
    "destroy",
    "drop",
    "truncate",
    "rm",
    "rmdir",
    "unlink",
    "mkfs",
    "dd",
    "wipe",
    "push",
    "reset",
    "clean",
    // network egress / external I/O
    "curl",
    "wget",
    "egress",
    "upload",
    "download",
    "send",
    // money
    "payment",
    "charge",
    "refund",
    "wire",
    // privilege
    "sudo",
    "chmod",
    "chown",
    "setuid",
];

/// Split a tool name into lowercase segments on non-alphanumeric boundaries AND
/// camelCase transitions, so `gitPush`, `git_push`, and `git-push` all yield
/// `["git", "push"]`. Matching whole segments (not substrings) is what makes the
/// name check honest: `count_tokens` → `["count", "tokens"]` does NOT match the
/// `token`-class danger, and `git_reset` → `["git", "reset"]` DOES match `reset`.
fn name_segments(name: &str) -> Vec<String> {
    let mut segs = Vec::new();
    let mut cur = String::new();
    let mut prev_lower_or_digit = false;
    for ch in name.chars() {
        if ch.is_alphanumeric() {
            // camelCase boundary: a lower/digit followed by an uppercase letter
            // starts a new segment (`gitPush` → `git` | `push`).
            if prev_lower_or_digit && ch.is_uppercase() && !cur.is_empty() {
                segs.push(std::mem::take(&mut cur));
            }
            cur.extend(ch.to_lowercase());
            prev_lower_or_digit = ch.is_lowercase() || ch.is_numeric();
        } else {
            if !cur.is_empty() {
                segs.push(std::mem::take(&mut cur));
            }
            prev_lower_or_digit = false;
        }
    }
    if !cur.is_empty() {
        segs.push(cur);
    }
    segs
}

/// True if a tool *name* names an irreversible / externally-consequential
/// capability — the stakes signal at the granularity available *before* an
/// action is built: a planning/agent loop knows its tool palette, not yet the
/// concrete action. Matches whole identifier segments against the curated
/// [`FULL_ACCESS_NAME_SEGMENTS`] set.
///
/// This is a SEGMENT-LEVEL APPROXIMATION of the irreversibility signal, NOT the
/// whole of the per-[`Action`] [`RiskClassifier::classify`] — it sees only the
/// tool name, so param-derived escalation (e.g. `rm -rf /` in a `cmd` arg under
/// a generic `shell` tool) is invisible here by construction. It is
/// deliberately cleaner than substring-scanning
/// [`FULL_ACCESS_COMMAND_KEYWORDS`] over a name (no `count_tokens`/`http_get`
/// false positives, no space-keyword false negatives). Used by the in-process
/// autonomous loops (active-planner, agents) to route generation quality-first
/// — the analogue of the daemon's session-tier `high_stakes` gate for the paths
/// that bypass it.
///
/// **Since Parslee-ai/car#917 a mis-classification here is not free.**
/// `classify` calls this as its tool-name signal, so a name wrongly in
/// [`FULL_ACCESS_NAME_SEGMENTS`] costs an approval prompt and a name wrongly
/// absent removes one — an authz decision, not just which model generates.
/// Weigh additions to that list accordingly.
pub fn tool_name_is_full_access(name: &str) -> bool {
    name_segments(name)
        .iter()
        .any(|seg| FULL_ACCESS_NAME_SEGMENTS.contains(&seg.as_str()))
}

/// True if *any* of the supplied tool names is full-access — convenience over
/// [`tool_name_is_full_access`] for a loop assessing its whole tool palette.
/// The `AsRef<str>` item bound lets callers pass `&[String]`, `&HashSet<String>`,
/// or `&[&str]` without an explicit `.map(String::as_str)`.
pub fn any_tool_full_access<I>(names: I) -> bool
where
    I: IntoIterator,
    I::Item: AsRef<str>,
{
    names
        .into_iter()
        .any(|n| tool_name_is_full_access(n.as_ref()))
}

/// Append every string scalar reachable under `v` to `out`, space-
/// separated. Arrays are flattened in order, so an argv array like
/// `["git","push","--force"]` becomes `"git push --force"` and matches a
/// space-containing keyword that the raw JSON (`["git","push"]`) would
/// hide (neo review). Non-string scalars are stringified too.
fn collect_strings(v: &serde_json::Value, out: &mut String) {
    use serde_json::Value;
    match v {
        Value::String(s) => {
            out.push_str(s);
            out.push(' ');
        }
        Value::Array(items) => {
            for it in items {
                collect_strings(it, out);
            }
        }
        Value::Object(map) => {
            for val in map.values() {
                collect_strings(val, out);
            }
        }
        Value::Number(_) | Value::Bool(_) | Value::Null => {
            out.push_str(&v.to_string());
            out.push(' ');
        }
    }
}

/// One lowercase haystack per action: the tool name followed by every
/// (possibly nested, possibly argv-array) string reachable in its parameters.
/// Shared by both classifiers so the authority axis and the reversibility axis
/// read the *same* text and differ only in what they look for in it.
///
/// Flattening is what makes space-bearing phrases matchable at all: an argv
/// array `["git","push","--force"]` becomes `"git push --force "`, which the
/// raw JSON never contained.
///
/// # Two properties this function must hold, and why
///
/// **Deterministic.** `Action::parameters` is a `std::collections::HashMap`,
/// whose iteration order is randomized per process. Walking it directly makes
/// the haystack — and therefore both classifications — differ run to run for
/// the same action. Parameters are visited in sorted key order instead.
///
/// **Parameter-separated.** Values are joined with `\n` rather than a space, so
/// a phrase can only match *within* one top-level parameter. Concatenating two
/// unrelated parameters can spell a phrase neither of them contains: with
/// `{"command": "git", "args": ["push", ...]}` the flattened text reads
/// `"git push"` and matches `COMPENSABLE_PARAM_PHRASES` — half the time, on
/// whichever iteration order the process happened to draw.
///
/// **Command-line adjacency is reconstructed, not left to chance.** The one
/// cross-parameter concatenation that carries real meaning is
/// `{"command": "rm", "args": ["-rf", …]}`, where `"rm -rf"` exists only once
/// the two are joined. That is why the fix is not "sort the keys": sorting
/// alone puts `args` before `command` and loses `"rm -rf"` — deterministically,
/// which is worse than losing it at random. [`COMMAND_KEYS`] are emitted first,
/// then [`ARG_KEYS`], space-joined into one line; every other parameter follows
/// on its own line in sorted order.
///
/// Together these fix a defect that mattered in both directions and was
/// invisible in one. For the tier classifier a phantom match only ever
/// escalates. For [`classify_reversibility`] a phantom
/// `COMPENSABLE_PARAM_PHRASES` hit *lowers* the contract from the
/// `Irreversible` default — a force-push came back `compensable` or
/// `irreversible` depending on the draw — while a *missed* `rm -rf` left a
/// destructive shell call unrecognized on exactly the same coin flip.
///
/// `\n` is whitespace, so [`sandbox_confined_paths`]' tokenization is
/// unaffected, and no phrase in any set here spans a newline.
fn action_haystack(action: &Action) -> String {
    let mut hay = String::new();
    if let Some(tool) = &action.tool {
        hay.push_str(tool);
        hay.push('\n');
    }

    // The command line, rebuilt in argv order so `command` + `args` form the
    // one adjacency the phrase sets are written against. Shared with
    // `command_line` so the text scanned by `FULL_ACCESS_COMMAND_KEYWORDS` and
    // the text embedded here can never drift apart.
    if let Some(cmdline) = command_line(action) {
        hay.push_str(&cmdline);
        hay.push('\n');
    }

    // Everything else, one parameter per line, in a deterministic order.
    let mut keys: Vec<&String> = action
        .parameters
        .keys()
        .filter(|k| {
            let k = k.as_str();
            !COMMAND_KEYS.contains(&k) && !ARG_KEYS.contains(&k)
        })
        .collect();
    keys.sort();
    for k in keys {
        if let Some(v) = action.parameters.get(k) {
            collect_strings(v, &mut hay);
            hay.push('\n');
        }
    }
    hay.to_ascii_lowercase()
}

/// The action's command line — the [`COMMAND_KEYS`] program followed by its
/// [`ARG_KEYS`] argument vector, flattened in argv order and lowercased —
/// or `None` when the action is not command-shaped.
///
/// This is the surface [`FULL_ACCESS_COMMAND_KEYWORDS`] is scanned over, and
/// isolating it is what lets that list stay broad: `http`, `apply`, and
/// `format` are honest danger signals in `curl https://…`, `terraform apply`,
/// and `mkfs.ext4 --format`, and were only ever wrong when the same scan ran
/// across message bodies and URLs (Parslee-ai/car#917).
///
/// It reads the same keys in the same order as [`action_haystack`]'s command
/// line, so the two never disagree about what the command was. It is computed
/// from the action rather than sliced out of the shared haystack because the
/// haystack's line structure is positional — with no tool name, line 1 *is* the
/// command line — and a classifier must not depend on that. The extra
/// allocation is bounded by the argv, not by the payload: the parameters that
/// make flattening expensive (a file body, a diff) are exactly the ones
/// excluded here.
fn command_line(action: &Action) -> Option<String> {
    let mut cmd = String::new();
    for group in [COMMAND_KEYS, ARG_KEYS] {
        for key in group {
            if let Some(v) = action.parameters.get(*key) {
                collect_strings(v, &mut cmd);
            }
        }
    }
    if cmd.is_empty() {
        None
    } else {
        Some(cmd.to_ascii_lowercase())
    }
}

/// True if any parameter key — at any nesting depth — names credential
/// material, per [`CREDENTIAL_PARAM_KEYS`] and [`CREDENTIAL_KEY_SEGMENTS`].
///
/// Reads **keys only, never values**. That is the point: the value of an
/// `api_key` parameter is opaque high-entropy text that no keyword list can
/// recognize, while the key states outright what the action is handling. It is
/// also why narrowing the value scan in #917 did not lose the credential
/// signal — it moved to where the evidence actually is.
///
/// Array elements inherit nothing: unlike [`path_parameter_haystack`] there is
/// no "descend under a matched key" mode, because the question is whether a
/// credential-named key *exists*, not what is stored beneath it.
fn hits_credential_param_key(action: &Action) -> bool {
    fn key_is_credential(key: &str) -> bool {
        let depunct: String = key
            .chars()
            .filter(|c| c.is_alphanumeric())
            .flat_map(char::to_lowercase)
            .collect();
        if CREDENTIAL_PARAM_KEYS.contains(&depunct.as_str()) {
            return true;
        }
        name_segments(key)
            .iter()
            .any(|seg| CREDENTIAL_KEY_SEGMENTS.contains(&seg.as_str()))
    }

    fn walk(v: &serde_json::Value) -> bool {
        use serde_json::Value;
        match v {
            Value::Object(map) => map
                .iter()
                .any(|(k, inner)| key_is_credential(k) || walk(inner)),
            Value::Array(items) => items.iter().any(walk),
            _ => false,
        }
    }

    action
        .parameters
        .iter()
        .any(|(k, v)| key_is_credential(k) || walk(v))
}

/// Both authorization axes for one action, from [`PermissionGate::evaluate_axes`].
///
/// `#[non_exhaustive]` for the reason `car_ir::Action` is (Parslee-ai/car#855):
/// `car-policy` is published, and a third axis should not break every consumer
/// that destructures this.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct ActionAxes {
    /// May this run — the gate's verdict, carrying the required tier.
    pub decision: GateDecision,
    /// Could it be taken back. Independent of `decision`: the gate's verdict
    /// says nothing about whether the effect survives a rollback.
    pub reversibility: Reversibility,
}

/// The flattened, lowercased text both classifiers read for one action.
///
/// Public so a caller computing both axes can build it once and hand it to
/// [`RiskClassifier::classify_with_haystack`] and
/// [`classify_reversibility_with_haystack`]. The flattening walks every nested
/// string in `Action::parameters` and then allocates a lowercase copy, which is
/// not free when a parameter carries a large file body or diff.
pub fn action_text(action: &Action) -> String {
    action_haystack(action)
}

/// Parameter keys naming the program in a command-shaped invocation.
/// Emitted before [`ARG_KEYS`] by [`action_haystack`] so the flattened text
/// reads as a command line rather than in whatever order a `HashMap` yields.
const COMMAND_KEYS: &[&str] = &["command", "cmd", "executable", "program", "bin", "binary"];

/// Parameter keys naming the argument vector in a command-shaped invocation.
const ARG_KEYS: &[&str] = &["args", "argv", "arguments", "flags", "options"];

/// Keys whose value names a filesystem location the tool will act *on*.
///
/// [`sandbox_confined_paths`] reads only these, so that a scratch path
/// appearing somewhere incidental — inside a file's `contents`, a diff body, a
/// rendered template — cannot vouch for a write whose actual target is
/// somewhere else entirely.
///
/// A tool that names its target with a key not listed here contributes no
/// paths, so the sandbox rule finds nothing to qualify on and the ladder falls
/// through to the conservative default. That is the intended direction: this
/// set failing to recognize a target costs an over-ask, while reading the wrong
/// parameter would cost a silent under-classification.
///
/// Crucially, this set gates only the *qualifying* half of
/// [`sandbox_confined_paths`]. The disqualifying half reads the whole action,
/// so a dangerous path under a key missing from this list still vetoes — an
/// omission here can never make a write look discardable when it is not.
const PATH_PARAM_KEYS: &[&str] = &[
    "path",
    "paths",
    "file",
    "files",
    "file_path",
    "filepath",
    "filename",
    "dir",
    "directory",
    "folder",
    "dest",
    "destination",
    "target",
    "target_path",
    "output",
    "output_path",
    "out",
    "src",
    "source",
    "source_path",
    "cwd",
    "workdir",
    "working_dir",
];

/// The paths an action declares as its *targets*, lowercased and
/// whitespace-joined — the input to [`sandbox_confined_paths`].
///
/// Walks `parameters` in sorted key order (same determinism requirement as
/// [`action_haystack`]) and descends into nested objects, collecting only
/// values reached through a [`PATH_PARAM_KEYS`] key.
fn path_parameter_haystack(action: &Action) -> String {
    fn walk(v: &serde_json::Value, key_matched: bool, out: &mut String) {
        use serde_json::Value;
        match v {
            Value::Object(map) => {
                let mut keys: Vec<&String> = map.keys().collect();
                keys.sort();
                for k in keys {
                    let hit =
                        key_matched || PATH_PARAM_KEYS.contains(&k.to_ascii_lowercase().as_str());
                    if let Some(inner) = map.get(k) {
                        walk(inner, hit, out);
                    }
                }
            }
            Value::Array(items) => {
                for it in items {
                    walk(it, key_matched, out);
                }
            }
            Value::String(s) if key_matched => {
                out.push_str(s);
                out.push(' ');
            }
            _ => {}
        }
    }

    let mut out = String::new();
    let mut keys: Vec<&String> = action.parameters.keys().collect();
    keys.sort();
    for k in keys {
        let hit = PATH_PARAM_KEYS.contains(&k.to_ascii_lowercase().as_str());
        if let Some(v) = action.parameters.get(k) {
            walk(v, hit, &mut out);
        }
    }
    out.to_ascii_lowercase()
}

/// Classifies an [`Action`] into the minimum [`PermissionTier`] required
/// to perform it. Combines a built-in heuristic with optional custom
/// rules; the result is the **highest** tier any signal implies, since
/// risk is monotonic (one high-risk signal escalates the whole action).
pub struct RiskClassifier {
    rules: Vec<ClassifierRule>,
}

struct ClassifierRule {
    name: String,
    tier: PermissionTier,
    matcher: Box<dyn Fn(&Action) -> bool + Send + Sync>,
}

impl std::fmt::Debug for RiskClassifier {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RiskClassifier")
            .field(
                "rules",
                &self
                    .rules
                    .iter()
                    .map(|r| r.name.as_str())
                    .collect::<Vec<_>>(),
            )
            .finish()
    }
}

impl RiskClassifier {
    /// A classifier with only the built-in heuristic.
    pub fn new() -> Self {
        Self { rules: Vec::new() }
    }

    /// Add a custom rule. A matching rule can only *raise* the required
    /// tier (via `max`), never lower it — safety is the default.
    pub fn add_rule<F>(&mut self, name: &str, tier: PermissionTier, matcher: F)
    where
        F: Fn(&Action) -> bool + Send + Sync + 'static,
    {
        self.rules.push(ClassifierRule {
            name: name.to_string(),
            tier,
            matcher: Box::new(matcher),
        });
    }

    /// The built-in, keyword-free baseline from the action's *type*.
    fn baseline(action: &Action) -> PermissionTier {
        match action.action_type {
            // Reads and assertions never mutate or reach outside.
            ActionType::StateRead | ActionType::Assertion => PermissionTier::ReadOnly,
            // Local state mutation is reversible (snapshot/rollback).
            ActionType::StateWrite => PermissionTier::SandboxEdit,
            // A tool call's effects are opaque to static analysis; assume
            // it can mutate, but escalate to FullAccess only on a signal.
            ActionType::ToolCall => PermissionTier::SandboxEdit,
        }
    }

    /// Does anything about this action require the top grant?
    ///
    /// Four independent signals, each matched at the granularity its evidence
    /// actually supports — which is the whole of the Parslee-ai/car#917 fix.
    /// The predecessor was one substring scan of the broad keyword list over
    /// the entire flattened payload, so `http` in a URL, `send` in a message
    /// body, and `format` in a tool name all read as full-access.
    ///
    /// 1. **Tool name**, whole snake/camel segments against the curated
    ///    [`FULL_ACCESS_NAME_SEGMENTS`] — [`tool_name_is_full_access`].
    ///    `deploy_service` and `send_email` hit; `http_get`, `format_date`, and
    ///    `count_tokens` no longer do.
    /// 2. **Command line**, substring against [`FULL_ACCESS_COMMAND_KEYWORDS`]
    ///    — the unchanged broad list, over `command`/`args` only, where those
    ///    tokens genuinely name commands.
    /// 3. **Parameter keys**, boundary-matched against
    ///    [`CREDENTIAL_PARAM_KEYS`] — `{"api_key": …}` is full-access because
    ///    of the key, so credential values need no substring hunt.
    /// 4. **Declared target paths**, substring against
    ///    [`CREDENTIAL_PATH_FRAGMENTS`] — `{"path": "~/.ssh/id_rsa"}` acts
    ///    *on* key material, which a mention of `~/.ssh` in a file body does
    ///    not.
    /// 5. **Everything else**, substring against the narrow
    ///    [`FULL_ACCESS_TEXT_PHRASES`] — `drop table`, `rm -rf`, phrases that
    ///    are never accidental prose.
    ///
    /// Signal 5 reads the full shared haystack rather than only the leftover
    /// parameters. Scanning the command line twice is harmless (both lists
    /// escalate) and it keeps the one-allocation contract of
    /// [`classify_with_haystack`] intact.
    fn hits_full_access(action: &Action, hay: &str) -> bool {
        if action.tool.as_deref().is_some_and(tool_name_is_full_access) {
            return true;
        }
        if let Some(cmd) = command_line(action) {
            if FULL_ACCESS_COMMAND_KEYWORDS.iter().any(|k| cmd.contains(k)) {
                return true;
            }
        }
        if hits_credential_param_key(action) {
            return true;
        }
        let targets = path_parameter_haystack(action);
        if CREDENTIAL_PATH_FRAGMENTS
            .iter()
            .any(|f| targets.contains(f))
        {
            return true;
        }
        FULL_ACCESS_TEXT_PHRASES.iter().any(|p| hay.contains(p))
    }

    /// Classify an action into its minimum required tier.
    pub fn classify(&self, action: &Action) -> PermissionTier {
        self.classify_with_haystack(action, None)
    }

    /// [`RiskClassifier::classify`] reusing a haystack the caller already
    /// built. A caller computing *both* axes for one action would otherwise
    /// flatten and lowercase the whole parameter payload twice — see
    /// [`classify_reversibility_with_haystack`], and [`action_text`] for the
    /// haystack itself.
    pub fn classify_with_haystack(
        &self,
        action: &Action,
        haystack: Option<&str>,
    ) -> PermissionTier {
        let mut tier = Self::baseline(action);
        if action.action_type == ActionType::ToolCall {
            let owned;
            let hay: &str = match haystack {
                Some(h) => h,
                None => {
                    owned = action_haystack(action);
                    &owned
                }
            };
            if Self::hits_full_access(action, hay) {
                tier = tier.max(PermissionTier::FullAccess);
            }
        }
        for rule in &self.rules {
            if (rule.matcher)(action) {
                tier = tier.max(rule.tier);
            }
        }
        tier
    }
}

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

// ---------------------------------------------------------------------------
// The second axis: can this be undone?
//
// Everything below classifies `car_ir::Reversibility` and reads NOTHING from
// the tier machinery above — not `RiskClassifier`, not `PermissionTier`, not
// `FULL_ACCESS_COMMAND_KEYWORDS`. That separation is the entire point: deriving one
// axis from the other would reproduce the conflation this exists to undo.
// The two sets overlap in places (both care about `delete`) and diverge in
// more (`secret` is top-authority and perfectly reversible; `insert` is
// low-authority and needs a compensating delete), and each is curated against
// its own question.
// ---------------------------------------------------------------------------

/// Tool-name segments naming an effect that cannot be undone once it runs.
///
/// Curated for IDENTIFIER matching against whole snake/camel segments (see
/// [`name_segments`]), like [`FULL_ACCESS_NAME_SEGMENTS`] and unlike the
/// free-text [`FULL_ACCESS_TEXT_PHRASES`].
///
/// **Mostly verbs, deliberately.** The object a tool touches tells you what is
/// at stake; only the verb tells you what happens to it. `email` as a segment
/// would make `read_email` and `list_email` irreversible, which is nonsense —
/// so the egress family is matched by `send` / `notify` / `dispatch` instead.
/// The exception is money (see the fn docs for [`classify_reversibility`]):
/// every mutation on a payment rail is permanent, so those are matched as
/// nouns and the cost is that `get_payment` over-classifies.
///
/// Segments that would collide with common benign names are excluded on the
/// same grounds the tier segment list excludes them: `page` (`get_page`),
/// `format` (`format_date`), `capture` (`capture_screenshot`), `launch`
/// (`launch_browser`), `call` (`call_tool`), `message` (`get_messages`).
const IRREVERSIBLE_NAME_SEGMENTS: &[&str] = &[
    // Egress to a person or a third party. By the time anyone objects the
    // recipient has already read it. A retraction is a *mitigation*, not a
    // compensation — "send a correction" does not unsend the first message.
    "send",
    "sendmail",
    "notify",
    "dispatch",
    "broadcast",
    "announce",
    // Publication. An artifact on a public registry can be yanked but never
    // un-published, and a message a consumer has already drained off a queue
    // cannot be recalled.
    "publish",
    "enqueue",
    // Money — matched by its object, not its verb. See the fn docs.
    "pay",
    "payment",
    "payments",
    "charge",
    "refund",
    "payout",
    "invoice",
    "checkout",
    "wire",
    "remit",
    // Destruction with no retained copy. `car_engine::Checkpoint` restores the
    // state map and nothing else, so whatever a tool erased from disk stays
    // erased — the rollback hole the Shepherd proposal calls gap #2.
    "delete",
    "destroy",
    "purge",
    "wipe",
    "shred",
    "erase",
    "truncate",
    "mkfs",
    "rm",
    "rmdir",
    "unlink",
    "drop",
    // Privilege withdrawal and physical actuation, on the world side of the
    // gate. A reissued credential is a new credential, not the old one back.
    "revoke",
    "actuate",
    "unlock",
];

/// Tool-name segments naming an effect that left the scope but has a known,
/// mechanical inverse — the `Compensable` contract.
///
/// **Deliberately stingier than [`IRREVERSIBLE_NAME_SEGMENTS`]**, because the
/// two sets fail in opposite directions. Omitting something here drops it to
/// the conservative `Irreversible` default (over-asks, visible, locally
/// fixable); adding something wrongly *lowers* the assessed contract, which is
/// the failure mode that stays quiet. Generic verbs that would lower a pure
/// function are therefore left out: `add` (`add_numbers`), `set`, `merge`
/// (`merge_dicts`), `scale` (`scale_image`), `branch` (`branch_decision`),
/// `apply` (`apply_template` — the tier segment list excludes it for the same
/// reason; the genuinely consequential `kubectl apply` / `terraform apply`
/// cases are caught as parameter phrases instead).
const COMPENSABLE_NAME_SEGMENTS: &[&str] = &[
    // Row/record creation and mutation: the inverse is a delete or a rewrite.
    "insert",
    "upsert",
    "create",
    "put",
    "update",
    "patch",
    "clone",
    // Resource lifecycle: the inverse is the paired verb (deregister,
    // unsubscribe, detach, unmount, stop, revoke-the-grant).
    "register",
    "provision",
    "allocate",
    "attach",
    "mount",
    "subscribe",
    "enable",
    "disable",
    "start",
    "stop",
    "restart",
    "grant",
    // VCS: force-push the prior ref, delete the tag, `git reset` the commit.
    "push",
    "tag",
    "commit",
    // Deployment: the inverse is a rollback deploy. `deployment` is listed
    // separately because segment matching is exact — `scale_deployment` splits
    // to `["scale", "deployment"]` and never equals `deploy`.
    "deploy",
    "deployment",
    "deployments",
    "rollout",
    "install",
    "upgrade",
    // Object storage: the inverse is deleting the object written.
    "upload",
];

/// Tool-name segments naming pure retrieval. A read leaves nothing behind, so
/// there is nothing to undo — but this only decides when the name carries no
/// mutating segment of either kind, so `get_and_delete` still falls through to
/// `Irreversible`.
const RETRIEVAL_NAME_SEGMENTS: &[&str] = &[
    "read",
    "get",
    "list",
    "search",
    "query",
    "select",
    "find",
    "grep",
    "stat",
    "describe",
    "inspect",
    "show",
    "view",
    "count",
    "head",
    "tail",
    "cat",
    "diff",
    "status",
    "lookup",
    "fetch",
    "load",
    "scan",
    "peek",
    "exists",
    "resolve",
    "summarize",
    "analyze",
    "classify",
    "parse",
    "validate",
    "check",
];

/// Tool-name segments that mark a tool as filesystem-shaped. Gates the
/// sandbox-path rule so that a *non*-filesystem tool carrying an incidental
/// `/tmp` path in its parameters — `http_post` with a `body_file`, say — is
/// never talked down to `Reversible` by it.
const FILESYSTEM_NAME_SEGMENTS: &[&str] = &[
    "file",
    "files",
    "fs",
    "filesystem",
    "dir",
    "directory",
    "folder",
    "path",
    "write",
    "edit",
    "append",
    "mkdir",
    "touch",
    "save",
];

/// Command-shaped destruction, matched as free text against the whole
/// [`action_haystack`]. This is the only signal that fires on a *generic*
/// tool: `shell` with `["rm","-rf","/var/data"]` names nothing dangerous in
/// its tool name, and the argv array never contained the literal phrase until
/// [`collect_strings`] flattened it.
const IRREVERSIBLE_PARAM_PHRASES: &[&str] = &[
    "rm -rf",
    "rm -r ",
    "rm -f ",
    "mkfs",
    "shred ",
    "dd if=",
    "drop table",
    "drop database",
    "delete from",
    "truncate table",
    "terraform destroy",
    "kubectl delete",
];

/// Free-text phrases that mark a `Compensable` effect.
///
/// Every entry is **multi-word and command-shaped**, and that is a rule rather
/// than a coincidence: a hit here *lowers* the assessed contract from the
/// `Irreversible` default, so a phrase loose enough to appear in an English
/// parameter ("update the docs", "insert a paragraph") would quietly downgrade
/// unrelated actions. A phrase that only occurs as a shell or SQL fragment
/// cannot.
const COMPENSABLE_PARAM_PHRASES: &[&str] = &[
    "git push",
    "git commit",
    "git tag",
    "docker push",
    "kubectl apply",
    "kubectl rollout",
    "helm install",
    "helm upgrade",
    "terraform apply",
    "insert into",
];

/// Absolute path **prefixes** the OS designates as scratch space — somewhere a
/// write can be discarded wholesale.
///
/// Every entry is anchored and matched with `starts_with`, not `contains`, and
/// that is load-bearing twice over. This set once carried the bare fragments
/// `"sandbox"` and `"scratch"`, which matched anywhere in a path token and so
/// accepted `/srv/sandbox-prod/index.html` and `/System/Library/Sandbox/…` as
/// discardable; and `contains` alone would still accept `/etc/tmp/hosts` on the
/// strength of `/tmp/`.
///
/// A directory merely *named* "sandbox" is a naming convention, not a
/// guarantee that anything will restore it, so no such entry belongs here. Only
/// locations whose disposability the operating system itself defines qualify.
const SANDBOX_PATH_PREFIXES: &[&str] = &[
    "/tmp/",
    "/private/tmp/",
    "/var/tmp/",
    "/private/var/tmp/",
    "/var/folders/",
    "/private/var/folders/",
    "/dev/shm/",
    "c:\\temp\\",
    "c:\\windows\\temp\\",
    "\\\\?\\c:\\temp\\",
];

/// The absolute-path tokens in a flattened parameter blob, de-quoted.
fn absolute_path_tokens(hay: &str) -> impl Iterator<Item = &str> {
    hay.split_whitespace()
        .map(|raw| {
            raw.trim_matches(|c: char| matches!(c, '"' | '\'' | '`' | ',' | ';' | ')' | '('))
        })
        .filter(|token| {
            let bytes = token.as_bytes();
            // POSIX absolute, a UNC share, or a Windows drive-qualified path.
            token.starts_with('/')
                || token.starts_with("\\\\")
                || (bytes.len() > 2 && bytes[1] == b':' && (bytes[2] == b'\\' || bytes[2] == b'/'))
        })
}

fn is_scratch_path(token: &str) -> bool {
    SANDBOX_PATH_PREFIXES.iter().any(|p| token.starts_with(p))
}

/// True when the action's writes are confined to scratch space. Two halves,
/// reading two different blobs, because qualifying and disqualifying have
/// opposite safe directions.
///
/// **Disqualify — read everything.** Any absolute path *anywhere* in the action
/// that is not scratch space vetoes the rule, whether or not it arrived under a
/// key [`PATH_PARAM_KEYS`] recognizes. This is what keeps a write touching both
/// `/tmp/in` and `/etc/hosts` out of the `Reversible` bucket, and it is
/// deliberately not restricted to recognized keys: an unrecognized key holding
/// the dangerous path (`dst`, `where`, `loc`) would otherwise be *invisible*,
/// turning a missed key from an over-ask into a silent under-classification.
///
/// **Qualify — read only the declared targets.** At least one absolute path
/// must appear among [`path_parameter_haystack`]'s output. A `/tmp` path in a
/// file's `contents` says nothing about where that file lands, so it may veto
/// but never vouch.
///
/// Relative paths are invisible to both halves by construction — the classifier
/// cannot see the working directory they would resolve against — so they
/// neither qualify an action nor disqualify it, and a tool called with only
/// relative paths falls through to the conservative default.
fn sandbox_confined_paths(target_paths: &str, whole_action: &str) -> bool {
    if absolute_path_tokens(whole_action).any(|t| !is_scratch_path(t)) {
        return false;
    }
    absolute_path_tokens(target_paths).next().is_some()
}

/// Statement shapes that prove a *mutation* whatever the tool is called.
///
/// Used only as a **veto** on the retrieval short-circuit (step 2), never to
/// assign a contract. That asymmetry is what lets this set be generous where
/// [`COMPENSABLE_PARAM_PHRASES`] must be stingy: an over-match here costs a
/// fall-through to the conservative ladder (an over-ask, visible and locally
/// fixable), while an over-match there silently downgrades a permanent effect.
///
/// The bare SQL verbs are deliberately absent — "update the docs" and "create a
/// summary" are ordinary English. `UPDATE … SET` is matched as a pair by
/// [`mutating_parameter_evidence`] for the same reason.
const MUTATING_PARAM_PHRASES: &[&str] = &[
    // SQL DML / DDL.
    "insert into",
    "delete from",
    "drop table",
    "drop column",
    "drop database",
    "drop index",
    "drop view",
    "drop constraint",
    "alter table",
    "alter column",
    "create table",
    "create index",
    "create database",
    "truncate table",
    "replace into",
    "merge into",
    "grant ",
    "revoke ",
    // Shell-shaped mutation a retrieval-sounding wrapper might carry.
    "-delete",
    "-exec rm",
    "--force",
    "--overwrite",
    "--prune",
    "rm -",
    "mv ",
    "chmod ",
    "chown ",
];

/// Does the flattened parameter text prove the action mutates something?
///
/// The veto behind step 2 of [`classify_reversibility`]. A retrieval verb in a
/// *tool name* is weak evidence — `execute_query`, `db_query`, and `find` all
/// read as retrieval and all routinely carry a mutation in their arguments —
/// and it must not be allowed to override evidence in the parameters
/// themselves.
fn mutating_parameter_evidence(hay: &str) -> bool {
    if MUTATING_PARAM_PHRASES.iter().any(|p| hay.contains(p)) {
        return true;
    }
    // `UPDATE <table> SET <col> = …`: neither half is command-shaped alone.
    hay.contains("update ") && hay.contains(" set ")
}

/// Classify an [`Action`] into its rollback contract — **can this be undone?**
///
/// This is the second of CAR's two authorization-adjacent axes, and it is
/// computed without consulting [`RiskClassifier`], [`PermissionTier`], or
/// [`FULL_ACCESS_COMMAND_KEYWORDS`] at any point. Deriving it from the tier would
/// reproduce exactly the conflation it exists to undo: a database `INSERT` and
/// a sent email are indistinguishable on the authority ladder and have nothing
/// in common on this one. The two are meant to be read side by side —
/// `car-engine`'s `TierPermissionHandler` puts `required_tier` and
/// `reversibility` on the same `PermissionDecision` event — and they disagree
/// in **both** directions:
///
/// | Action | Authority | Rollback contract |
/// |---|---|---|
/// | `state_read` | `ReadOnly` | `Reversible` |
/// | write a scratch file under `/tmp` | `SandboxEdit` | `Reversible` |
/// | `read_secret` | **`FullAccess`** | **`Reversible`** — a read leaves nothing to undo |
/// | `db_insert` | **`SandboxEdit`** | **`Compensable`** — delete the row |
/// | `git_push` | `FullAccess` | `Compensable` — force-push the prior ref |
/// | `send_email`, `charge_card` | `FullAccess` | `Irreversible` |
///
/// # It says nothing about disclosure
///
/// `read_secret` is `Reversible`, and that is correct rather than a bug: this
/// axis is about *effects*, and reading has none to reverse. An exfiltrating
/// read is maximally dangerous and maximally reversible at the same time,
/// which is precisely why this must never be used as a stand-in for
/// [`PermissionTier`]. Two axes, two questions, two answers.
///
/// # How it decides
///
/// Non-tool actions are settled by their type: a [`ActionType::StateRead`] or
/// [`ActionType::Assertion`] leaves nothing behind, and a
/// [`ActionType::StateWrite`]'s whole footprint is the KV store, which
/// `car_state`'s snapshot/rollback restores — that *is* the definition of
/// `Reversible`. A [`ActionType::ToolCall`] runs a ladder, most severe first:
///
/// 1. A command-shaped destructive phrase anywhere in the flattened
///    parameters → `Irreversible`. First, because it is the only signal that
///    fires on a generic tool (`shell` with `rm -rf` in an argv array).
/// 2. A retrieval verb with **no** mutating segment of either kind, no
///    compensable phrase, and no mutation spelled out in the parameters
///    ([`mutating_parameter_evidence`]) → `Reversible`.
/// 3. An [`IRREVERSIBLE_NAME_SEGMENTS`] hit → `Irreversible`.
/// 4. A [`COMPENSABLE_NAME_SEGMENTS`] or [`COMPENSABLE_PARAM_PHRASES`] hit →
///    `Compensable`.
/// 5. A filesystem-shaped tool whose every absolute **target** path
///    ([`path_parameter_haystack`]) is scratch space → `Reversible`.
/// 6. Otherwise `Irreversible` — the same conservative default, for the same
///    reason, as `Action::reversibility`'s `#[serde(default)]`.
///
/// The direction that is safe differs per set, and the sets are curated
/// accordingly: over-matching `Irreversible` over-asks (visible, locally
/// fixable), while over-matching `Compensable` or `Reversible` quietly
/// understates a permanent effect. So the irreversible set is generous and the
/// other two are stingy.
///
/// # What this is not
///
/// A heuristic over identifiers and strings — **not** a decision procedure,
/// and no more authoritative than the tier classifier above. It reads a tool
/// name and a flattened parameter blob. It does not know what a tool actually
/// does, cannot consult a tool's schema or documentation, and cannot resolve a
/// relative path. Known misses, recorded so nobody has to rediscover them:
///
/// - **An unrecognized tool comes back `Irreversible`.** Against a corpus of
///   tools nobody has taught it about, that is most of them. The fix is to
///   annotate `Action::reversibility` explicitly or extend a set here — not to
///   loosen the default.
/// - **A retrieval verb over a mutating noun over-classifies.**
///   `list_deployments` is a read, but `deployments` is in the compensable
///   set, so step 2 declines and step 4 answers `Compensable`.
/// - **Money is matched by its object.** `get_payment` therefore comes back
///   `Irreversible`. Every *mutation* on a payment rail is permanent and the
///   classifier does not try to be clever about which calls are reads.
/// - **`git push` is `Compensable` unconditionally**, where the honest hedge
///   is "if nobody has pulled yet". A force-push that destroys commits no one
///   else holds a copy of is irreversible in fact, and nothing here can tell.
/// - **A relative path defeats the sandbox rule** (step 5), so
///   `write_file("notes.txt")` falls through to `Irreversible` even when the
///   cwd is a sandbox.
/// - **A target named by an unrecognized key defeats it too.** Step 5 reads
///   only [`PATH_PARAM_KEYS`]; a tool that calls its destination `where` or
///   `loc` contributes no paths and falls through. Over-asking, deliberately.
/// - **Only OS scratch roots count as discardable.** A directory named
///   `sandbox` or `scratch` does not qualify — see [`SANDBOX_PATH_PREFIXES`].
/// - **Prose can trip a phrase.** A parameter containing the literal text
///   `insert into` reads as a SQL insert.
/// - **The retrieval veto is generous.** [`MUTATING_PARAM_PHRASES`] fires on
///   `--force` and `mv `, so `search_files` over a corpus that quotes either
///   loses the step-2 short-circuit and falls to the conservative default.
///
/// And it is not enforced. Nothing in the runtime consults this value to
/// decide whether an action runs; it is classified and audited. Deferring the
/// materialization of an irreversible effect needs a checkpoint coupled to the
/// filesystem, which CAR does not have. See the
/// [`car_ir::reversibility`] module docs.
pub fn classify_reversibility(action: &Action) -> Reversibility {
    classify_reversibility_with_haystack(action, None)
}

/// [`classify_reversibility`] reusing a haystack the caller already built.
///
/// [`action_haystack`] flattens every nested string in `Action::parameters`
/// into a `String` and then allocates a lowercase copy of it. A caller that
/// classifies *both* axes for the same action — `TierPermissionHandler` on the
/// execution path, `permission.evaluate` over a whole batch — otherwise pays
/// that twice for payloads that can be a multi-megabyte `contents` or diff.
///
/// The haystack must be [`action_haystack`]'s output for this same action;
/// passing anything else silently changes the classification.
pub fn classify_reversibility_with_haystack(
    action: &Action,
    haystack: Option<&str>,
) -> Reversibility {
    match action.action_type {
        // A read and an assertion observe; there is no effect to reverse.
        ActionType::StateRead | ActionType::Assertion => Reversibility::Reversible,
        // The whole footprint is the KV store, and `car_state` can restore it.
        ActionType::StateWrite => Reversibility::Reversible,
        ActionType::ToolCall => classify_tool_call_reversibility(action, haystack),
    }
}

/// The keyword ladder for [`ActionType::ToolCall`] — the only action type
/// whose effects are opaque to the IR. Documented step by step on
/// [`classify_reversibility`].
fn classify_tool_call_reversibility(action: &Action, haystack: Option<&str>) -> Reversibility {
    let owned;
    let hay: &str = match haystack {
        Some(h) => h,
        None => {
            owned = action_haystack(action);
            &owned
        }
    };
    let segments = action
        .tool
        .as_deref()
        .map(name_segments)
        .unwrap_or_default();
    let has = |set: &[&str]| segments.iter().any(|s| set.contains(&s.as_str()));
    let compensable_phrase = COMPENSABLE_PARAM_PHRASES.iter().any(|p| hay.contains(p));

    // 1. Destruction spelled out in the parameters, whatever the tool is called.
    if IRREVERSIBLE_PARAM_PHRASES.iter().any(|p| hay.contains(p)) {
        return Reversibility::Irreversible;
    }

    // 2. Unambiguous retrieval: a read verb and no mutation signal anywhere.
    //    Guarded on BOTH mutating name sets so `get_and_delete` and
    //    `list_and_push` fall through to the ladder below rather than being
    //    waved past it — and on the PARAMETERS too, because a retrieval verb in
    //    the tool name is weak evidence that the arguments can contradict.
    //    `execute_query` carrying `UPDATE accounts SET balance = 0` is a read
    //    by name and a permanent overwrite in fact; this is the only step that
    //    reaches `Reversible` without positive evidence of confinement, so it
    //    is the one that must not be reachable on the strength of a name alone.
    if has(RETRIEVAL_NAME_SEGMENTS)
        && !has(IRREVERSIBLE_NAME_SEGMENTS)
        && !has(COMPENSABLE_NAME_SEGMENTS)
        && !compensable_phrase
        && !mutating_parameter_evidence(hay)
    {
        return Reversibility::Reversible;
    }

    // 3. Permanent effects, before compensable ones: severity wins ties, so a
    //    `create_payment` is answered by `payment`, not by `create`.
    if has(IRREVERSIBLE_NAME_SEGMENTS) {
        return Reversibility::Irreversible;
    }

    // 4. Effects that left the scope but have a mechanical inverse.
    if has(COMPENSABLE_NAME_SEGMENTS) || compensable_phrase {
        return Reversibility::Compensable;
    }

    // 5. A filesystem write confined to scratch space: discard the tree and
    //    the effect is gone. Two independent gates, because either alone lets
    //    an unrelated scratch path vouch for a real write. The tool must be
    //    filesystem-shaped (so `http_post` with a `body_file` is not talked
    //    down), AND the paths considered are only those the action declares as
    //    its *target* (so `/tmp` inside a file's `contents` cannot vouch for a
    //    write whose `path` is `config.yaml`).
    if has(FILESYSTEM_NAME_SEGMENTS)
        && sandbox_confined_paths(&path_parameter_haystack(action), hay)
    {
        return Reversibility::Reversible;
    }

    // 6. Nothing recognized. Assume the worst — see the fn docs.
    Reversibility::Irreversible
}

/// Recursively canonicalize a JSON value so logically-identical values
/// produce byte-identical serializations: every object's keys are sorted
/// at *every* depth. Without this, two semantically identical params that
/// differ only in nested-object key order would fingerprint differently —
/// which would let a model evade a standing **rejection** simply by
/// permuting nested keys (neo review).
fn canonical_json(v: &serde_json::Value) -> serde_json::Value {
    use serde_json::Value;
    match v {
        Value::Object(map) => {
            let sorted: BTreeMap<&String, Value> = map
                .iter()
                .map(|(k, val)| (k, canonical_json(val)))
                .collect();
            Value::Object(
                sorted
                    .into_iter()
                    .map(|(k, val)| (k.clone(), val))
                    .collect(),
            )
        }
        Value::Array(items) => Value::Array(items.iter().map(canonical_json).collect()),
        other => other.clone(),
    }
}

/// The serde (snake_case) name of an action type — a **stable** wire
/// representation, unlike `Debug`, which carries no stability contract and
/// must never anchor a persisted key.
fn action_type_tag(t: &ActionType) -> &'static str {
    match t {
        ActionType::ToolCall => "tool_call",
        ActionType::StateWrite => "state_write",
        ActionType::StateRead => "state_read",
        ActionType::Assertion => "assertion",
    }
}

/// A stable fingerprint identifying "this kind of operation" so a human
/// approval/rejection can be matched against future occurrences — across
/// processes and builds. Built from the action type (stable serde tag),
/// tool, and **recursively** canonicalized parameters — *not* the action
/// id, which is not stable across proposals.
pub fn action_fingerprint(action: &Action) -> String {
    let canonical: BTreeMap<&String, serde_json::Value> = action
        .parameters
        .iter()
        .map(|(k, v)| (k, canonical_json(v)))
        .collect();
    let params = serde_json::to_string(&canonical).unwrap_or_default();
    let tool = action.tool.as_deref().unwrap_or("-");
    format!(
        "{}|{}|{}",
        action_type_tag(&action.action_type),
        tool,
        params
    )
}

/// Whether a recorded human decision approved or rejected an operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ApprovalDecision {
    Approved,
    Rejected,
}

/// A durable record of a human-in-the-loop decision — the auditable state
/// transition §5.2.5 calls for: what was proposed, who decided, why, and
/// against what evidence.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ApprovalRecord {
    /// Stable [`action_fingerprint`] this decision applies to.
    pub fingerprint: String,
    /// Tier the action required when the decision was made.
    pub required_tier: PermissionTier,
    pub decision: ApprovalDecision,
    /// Identity of the human (or delegated authority) who decided.
    pub reviewer: String,
    /// Why the decision was made — the recorded rationale.
    pub reason: String,
    /// Evidence shown at decision time (diff summary, risk surface, etc.).
    #[serde(default)]
    pub evidence: Option<String>,
    /// RFC3339 timestamp of the decision.
    pub decided_at: String,
}

/// An append-only ledger of human-in-the-loop decisions, keyed by
/// fingerprint (last decision wins). Optionally persisted as JSONL so the
/// approval state survives restarts — HITL decisions are *durable* harness
/// state, not transient prompts.
///
/// Concurrency: a single writer per journal file is assumed. The
/// stateless FFI opens a fresh ledger per call, so a product that drives
/// approvals from multiple processes against one journal must serialize
/// those writes itself (e.g. route them through the daemon). Reads
/// tolerate the writer appending concurrently; a torn final line is
/// skipped on load and counted in [`ApprovalLedger::skipped_on_load`].
#[derive(Debug, Default)]
pub struct ApprovalLedger {
    records: HashMap<String, ApprovalRecord>,
    journal: Option<PathBuf>,
    /// Count of unparseable lines skipped during the last load — nonzero
    /// signals journal corruption or a concurrent torn write, so callers
    /// can surface it rather than silently trusting a partial ledger.
    skipped_on_load: usize,
}

impl ApprovalLedger {
    pub fn new() -> Self {
        Self::default()
    }

    /// Create a ledger backed by a JSONL journal at `path`, loading any
    /// existing decisions. Each line is one [`ApprovalRecord`]; the last
    /// line for a fingerprint wins, so a later rejection overrides an
    /// earlier approval.
    pub fn with_journal(path: impl Into<PathBuf>) -> std::io::Result<Self> {
        let path = path.into();
        let mut ledger = Self {
            records: HashMap::new(),
            journal: Some(path.clone()),
            skipped_on_load: 0,
        };
        if path.exists() {
            let contents = std::fs::read_to_string(&path)?;
            // Lines are read in file (append) order, so a later decision
            // for a fingerprint overwrites an earlier one — last wins.
            for line in contents.lines() {
                let line = line.trim();
                if line.is_empty() {
                    continue;
                }
                match serde_json::from_str::<ApprovalRecord>(line) {
                    Ok(rec) => {
                        ledger.records.insert(rec.fingerprint.clone(), rec);
                    }
                    Err(_) => ledger.skipped_on_load += 1,
                }
            }
        }
        Ok(ledger)
    }

    /// Number of unparseable lines skipped during the load. Nonzero means
    /// the journal is corrupt or was torn by a concurrent writer.
    pub fn skipped_on_load(&self) -> usize {
        self.skipped_on_load
    }

    /// Record a decision, persisting it to the journal when configured.
    /// Returns the stored record.
    ///
    /// A journal write failure is an **error, not best-effort** (review A7):
    /// callers emit `ApprovalRecorded` audit events on the strength of this
    /// call, so a decision that only landed in memory must not be reported
    /// as durable. On `Err` the decision is NOT stored (memory and journal
    /// stay consistent — both lack it) and the caller must surface the
    /// failure. An in-memory ledger (no journal) cannot fail.
    pub fn record(&mut self, record: ApprovalRecord) -> std::io::Result<&ApprovalRecord> {
        if let Some(path) = &self.journal {
            // Durable append: write the whole line, flush Rust's writer, ask
            // the OS to commit file data + metadata, then close the handle
            // before publishing the decision in memory. `flush` alone is not
            // a durability boundary and left a fresh Windows process able to
            // observe the pre-approval journal (Parslee-ai/car#1227).
            let mut f = std::fs::OpenOptions::new()
                .create(true)
                .append(true)
                .open(path)?;
            let mut line = serde_json::to_string(&record)
                .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
            line.push('\n');
            f.write_all(line.as_bytes())?;
            f.flush()?;
            f.sync_all()?;
            drop(f);
        }
        // Last-wins: overwrite any prior decision for this fingerprint.
        use std::collections::hash_map::Entry;
        Ok(match self.records.entry(record.fingerprint.clone()) {
            Entry::Occupied(mut o) => {
                o.insert(record);
                o.into_mut()
            }
            Entry::Vacant(v) => v.insert(record),
        })
    }

    /// The current decision for a fingerprint, if any.
    pub fn lookup(&self, fingerprint: &str) -> Option<&ApprovalRecord> {
        self.records.get(fingerprint)
    }

    pub fn all(&self) -> impl Iterator<Item = &ApprovalRecord> {
        self.records.values()
    }

    /// Build and [`record`](Self::record) a decision against an explicit
    /// fingerprint — the ledger-level twin of
    /// [`PermissionGate::record_for_fingerprint`], for callers that share
    /// ONE ledger across many gates/sessions (e.g. the daemon's shared
    /// approval substrate) and so must not route the write through any
    /// single session's gate. Errs when the journal write fails (the
    /// decision is then NOT recorded — see [`Self::record`]).
    #[allow(clippy::too_many_arguments)]
    pub fn record_decision(
        &mut self,
        fingerprint: &str,
        required_tier: PermissionTier,
        decision: ApprovalDecision,
        reviewer: &str,
        reason: &str,
        evidence: Option<String>,
    ) -> std::io::Result<ApprovalRecord> {
        let record = ApprovalRecord {
            fingerprint: fingerprint.to_string(),
            required_tier,
            decision,
            reviewer: reviewer.to_string(),
            reason: reason.to_string(),
            evidence,
            decided_at: chrono::Utc::now().to_rfc3339(),
        };
        self.record(record.clone())?;
        Ok(record)
    }
}

/// A set of hazards partitioned against the durable [`ApprovalLedger`]: those
/// that must not run, and those still awaiting a human decision (paired with the
/// fingerprint the decision is keyed by).
///
/// The shared shape behind the per-domain HITL bridges
/// ([`crate::flow_gate::enforce_flow`], [`crate::intent_gate::enforce_intent`]):
/// each maps this into its own domain-specific enforcement struct + reason.
#[derive(Debug, Clone)]
pub struct LedgerPartition<V> {
    /// Hazards that must not run: the hard blocks passed in, plus anything a
    /// human previously **rejected**.
    pub blocked: Vec<V>,
    /// Novel hazards awaiting a human decision, each paired with its ledger
    /// fingerprint.
    pub pending: Vec<(String, V)>,
}

/// Resolve a `require_approval` hazard set against the durable ledger, the one
/// place that fixes the HITL semantics: a hazard a human previously **approved**
/// is dropped (let through), one they **rejected** joins `hard_blocked`, and an
/// **unseen** one becomes pending. `fingerprint` maps a hazard to its stable
/// ledger key (the per-domain part callers keep specialized).
pub fn partition_by_ledger<V, F>(
    hard_blocked: Vec<V>,
    needs_approval: &[V],
    fingerprint: F,
    ledger: &ApprovalLedger,
) -> LedgerPartition<V>
where
    V: Clone,
    F: Fn(&V) -> String,
{
    let mut blocked = hard_blocked;
    let mut pending = Vec::new();
    for v in needs_approval {
        let fp = fingerprint(v);
        match ledger.lookup(&fp).map(|r| r.decision) {
            Some(ApprovalDecision::Approved) => { /* a human OK'd this hazard before */ }
            Some(ApprovalDecision::Rejected) => blocked.push(v.clone()),
            None => pending.push((fp, v.clone())),
        }
    }
    LedgerPartition { blocked, pending }
}

/// The outcome of evaluating an action against the gate.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "decision", rename_all = "snake_case")]
pub enum GateDecision {
    /// May proceed autonomously. `required` and `granted` explain why.
    Allow {
        required: PermissionTier,
        granted: PermissionTier,
    },
    /// Autonomy is suspended pending a human decision (survey §5.2.5).
    /// The caller surfaces the request; a later [`PermissionGate::approve`]
    /// or [`PermissionGate::reject`] resolves it.
    NeedsApproval {
        required: PermissionTier,
        granted: PermissionTier,
        fingerprint: String,
        reason: String,
    },
    /// Refused — a human already rejected this operation.
    Deny {
        required: PermissionTier,
        fingerprint: String,
        reason: String,
    },
}

impl GateDecision {
    pub fn is_allow(&self) -> bool {
        matches!(self, GateDecision::Allow { .. })
    }
}

/// The permission gate: a session's standing authority plus the classifier
/// and the durable approval ledger. Pure and synchronous so it can be
/// embedded anywhere; the engine wraps it for its async pipeline.
#[derive(Debug)]
pub struct PermissionGate {
    /// Standing authority granted to this session.
    granted: PermissionTier,
    /// Actions at or above this tier *always* require a human decision,
    /// even when the granted tier would cover them — the "mandatory HITL
    /// gate" for consequential actions (§5.2.5). Default: `FullAccess`.
    require_approval_at: PermissionTier,
    classifier: RiskClassifier,
    ledger: ApprovalLedger,
}

impl PermissionGate {
    /// A gate with the given standing tier, default classifier, mandatory
    /// approval at `FullAccess`, and an in-memory ledger.
    pub fn new(granted: PermissionTier) -> Self {
        Self {
            granted,
            require_approval_at: PermissionTier::FullAccess,
            classifier: RiskClassifier::new(),
            ledger: ApprovalLedger::new(),
        }
    }

    pub fn with_classifier(mut self, classifier: RiskClassifier) -> Self {
        self.classifier = classifier;
        self
    }

    pub fn with_ledger(mut self, ledger: ApprovalLedger) -> Self {
        self.ledger = ledger;
        self
    }

    /// Override the tier at and above which approval is mandatory.
    pub fn with_mandatory_approval_at(mut self, tier: PermissionTier) -> Self {
        self.require_approval_at = tier;
        self
    }

    pub fn granted_tier(&self) -> PermissionTier {
        self.granted
    }

    pub fn set_granted_tier(&mut self, tier: PermissionTier) {
        self.granted = tier;
    }

    pub fn classifier(&self) -> &RiskClassifier {
        &self.classifier
    }

    pub fn ledger(&self) -> &ApprovalLedger {
        &self.ledger
    }

    /// Evaluate an action. Precedence:
    /// 1. A prior **rejection** denies (a human said no).
    /// 2. A prior **approval** allows (a human elevated this operation).
    /// 3. Actions at/above the mandatory-approval tier need approval.
    /// 4. Otherwise the granted tier must cover the required tier.
    /// 5. Exceeding standing authority escalates to a human, not a hard
    ///    deny — autonomy is suspended, not the task abandoned.
    pub fn evaluate(&self, action: &Action) -> GateDecision {
        self.evaluate_with_granted(action, self.granted, &self.ledger)
    }

    /// [`Self::evaluate`], but consulting an **external** ledger instead of the
    /// gate's own. For deployments that share ONE approval ledger across many
    /// per-session gates (per-session *tier*, shared *approvals*): the daemon
    /// keeps a single journal-backed ledger on its server state so an approval
    /// recorded on one connection is visible to every other and survives
    /// restart — a per-connection in-memory ledger would strand the approver's
    /// decision where the runner never reads it.
    pub fn evaluate_against(&self, action: &Action, ledger: &ApprovalLedger) -> GateDecision {
        self.evaluate_with_granted(action, self.granted, ledger)
    }

    /// Evaluate an action as if the session's standing authority were capped at
    /// `ceiling` — the join between skill-trust governance (arXiv 2602.12430) and
    /// the action-level gate. The effective authority is `min(granted, ceiling)`,
    /// so an action driven by a skill whose deployment ceiling is `read_only`
    /// cannot perform a `sandbox_edit` operation even in a `full_access` session;
    /// it escalates to a human instead. A `None` ceiling is identical to
    /// [`Self::evaluate`].
    ///
    /// The caller supplies the ceiling — it is the persisted
    /// `SkillMeta::deployment_tier` of the skill that drove the action, which the
    /// caller already knows (it retrieved the skill). The gate does not invent
    /// action→skill provenance; it honours the ceiling it is handed.
    pub fn evaluate_with_ceiling(
        &self,
        action: &Action,
        ceiling: Option<PermissionTier>,
    ) -> GateDecision {
        self.evaluate_with_ceiling_against(action, ceiling, &self.ledger)
    }

    /// [`Self::evaluate_with_ceiling`] against an external ledger — see
    /// [`Self::evaluate_against`] for when a shared ledger is the substrate.
    pub fn evaluate_with_ceiling_against(
        &self,
        action: &Action,
        ceiling: Option<PermissionTier>,
        ledger: &ApprovalLedger,
    ) -> GateDecision {
        let effective = match ceiling {
            Some(c) => self.granted.min(c),
            None => self.granted,
        };
        self.evaluate_with_granted(action, effective, ledger)
    }

    /// Both authorization axes for one action, flattening its parameters
    /// **once**.
    ///
    /// Every caller that records a `PermissionDecision` needs both: the gate's
    /// verdict (*may this run?*) and the rollback contract (*could it be taken
    /// back?*). Computed separately they each call [`action_text`], which walks
    /// every nested string in `Action::parameters` into a `String` and then
    /// allocates a lowercase copy — twice, per action, on the execution path,
    /// for payloads that can be a whole document or diff. Parslee-ai/car#856.
    ///
    /// `ceiling` caps standing authority the way
    /// [`Self::evaluate_with_ceiling`] does; `ledger` selects an external
    /// approval ledger the way [`Self::evaluate_against`] does, or `None` to
    /// use the gate's own.
    ///
    /// The two fields answer independent questions and neither is derived from
    /// the other — see [`classify_reversibility`]. They travel together here
    /// only because they read the same text.
    pub fn evaluate_axes(
        &self,
        action: &Action,
        ceiling: Option<PermissionTier>,
        ledger: Option<&ApprovalLedger>,
    ) -> ActionAxes {
        let hay = action_text(action);
        let granted = match ceiling {
            Some(c) => self.granted.min(c),
            None => self.granted,
        };
        ActionAxes {
            decision: self.evaluate_with_granted_haystack(
                action,
                granted,
                ledger.unwrap_or(&self.ledger),
                Some(&hay),
            ),
            reversibility: classify_reversibility_with_haystack(action, Some(&hay)),
        }
    }

    /// Core evaluation against an explicit effective `granted` tier (so a skill
    /// ceiling can cap standing authority without mutating the gate). Precedence:
    /// 1. A prior **rejection** denies (a human said no).
    /// 2. A prior **approval** allows (a human elevated this operation).
    /// 3. Actions at/above the mandatory-approval tier need approval.
    /// 4. Otherwise the effective granted tier must cover the required tier.
    /// 5. Exceeding standing authority escalates to a human, not a hard
    ///    deny — autonomy is suspended, not the task abandoned.
    fn evaluate_with_granted(
        &self,
        action: &Action,
        granted: PermissionTier,
        ledger: &ApprovalLedger,
    ) -> GateDecision {
        self.evaluate_with_granted_haystack(action, granted, ledger, None)
    }

    /// [`Self::evaluate_with_granted`] reusing a haystack the caller already
    /// built. Private: the public door is [`Self::evaluate_axes`], which owns
    /// the "build it once" decision rather than leaving it to every caller.
    fn evaluate_with_granted_haystack(
        &self,
        action: &Action,
        granted: PermissionTier,
        ledger: &ApprovalLedger,
        haystack: Option<&str>,
    ) -> GateDecision {
        let required = self.classifier.classify_with_haystack(action, haystack);
        let fingerprint = action_fingerprint(action);

        if let Some(rec) = ledger.lookup(&fingerprint) {
            match rec.decision {
                ApprovalDecision::Rejected => {
                    return GateDecision::Deny {
                        required,
                        fingerprint,
                        reason: format!("previously rejected by {} ({})", rec.reviewer, rec.reason),
                    };
                }
                // An approval is scoped to the risk that was actually
                // reviewed. If the operation has since been reclassified
                // *upward* (a new keyword, a new custom rule), the stale
                // approval must not bypass the mandatory gate — re-prompt
                // instead (neo review: tier-blind stale approvals).
                ApprovalDecision::Approved if required <= rec.required_tier => {
                    return GateDecision::Allow { required, granted };
                }
                ApprovalDecision::Approved => {
                    return GateDecision::NeedsApproval {
                        required,
                        granted,
                        fingerprint,
                        reason: format!(
                            "operation reclassified {} → {} since it was approved; re-approval required",
                            rec.required_tier.as_str(),
                            required.as_str()
                        ),
                    };
                }
            }
        }

        if required >= self.require_approval_at {
            return GateDecision::NeedsApproval {
                required,
                granted,
                fingerprint,
                reason: format!(
                    "{} actions require human approval before execution",
                    required.as_str()
                ),
            };
        }

        if granted.covers(required) {
            GateDecision::Allow { required, granted }
        } else {
            GateDecision::NeedsApproval {
                required,
                granted,
                fingerprint,
                reason: format!(
                    "action requires {} but session is granted only {}",
                    required.as_str(),
                    granted.as_str()
                ),
            }
        }
    }

    /// Record a human approval for the operation `action` represents.
    /// Errs when the ledger journal write fails (the decision is then NOT
    /// recorded — see [`ApprovalLedger::record`]).
    pub fn approve(
        &mut self,
        action: &Action,
        reviewer: &str,
        reason: &str,
        evidence: Option<String>,
    ) -> std::io::Result<ApprovalRecord> {
        self.record_decision(
            action,
            ApprovalDecision::Approved,
            reviewer,
            reason,
            evidence,
        )
    }

    /// Record a human rejection for the operation `action` represents.
    /// Errs when the ledger journal write fails (the decision is then NOT
    /// recorded — see [`ApprovalLedger::record`]).
    pub fn reject(
        &mut self,
        action: &Action,
        reviewer: &str,
        reason: &str,
        evidence: Option<String>,
    ) -> std::io::Result<ApprovalRecord> {
        self.record_decision(
            action,
            ApprovalDecision::Rejected,
            reviewer,
            reason,
            evidence,
        )
    }

    /// Record a decision against an explicit fingerprint (when the caller
    /// holds the fingerprint from a prior `NeedsApproval`, not the action).
    /// Errs when the ledger journal write fails (the decision is then NOT
    /// recorded — see [`ApprovalLedger::record`]).
    pub fn record_for_fingerprint(
        &mut self,
        fingerprint: &str,
        required_tier: PermissionTier,
        decision: ApprovalDecision,
        reviewer: &str,
        reason: &str,
        evidence: Option<String>,
    ) -> std::io::Result<ApprovalRecord> {
        self.ledger.record_decision(
            fingerprint,
            required_tier,
            decision,
            reviewer,
            reason,
            evidence,
        )
    }

    /// Build (but do NOT store) the [`ApprovalRecord`] for a decision on
    /// `action` — the gate classifies the action and derives its fingerprint;
    /// the caller records the result on whichever ledger is the substrate
    /// (its own via [`ApprovalLedger::record`], or a shared daemon ledger).
    /// This is what lets a per-session gate keep its tier/classifier while the
    /// approval store is shared.
    pub fn decision_record(
        &self,
        action: &Action,
        decision: ApprovalDecision,
        reviewer: &str,
        reason: &str,
        evidence: Option<String>,
    ) -> ApprovalRecord {
        ApprovalRecord {
            fingerprint: action_fingerprint(action),
            required_tier: self.classifier.classify(action),
            decision,
            reviewer: reviewer.to_string(),
            reason: reason.to_string(),
            evidence,
            decided_at: chrono::Utc::now().to_rfc3339(),
        }
    }

    fn record_decision(
        &mut self,
        action: &Action,
        decision: ApprovalDecision,
        reviewer: &str,
        reason: &str,
        evidence: Option<String>,
    ) -> std::io::Result<ApprovalRecord> {
        let record = self.decision_record(action, decision, reviewer, reason, evidence);
        self.ledger.record(record.clone())?;
        Ok(record)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use car_ir::ActionType;
    use serde_json::json;
    use std::collections::HashMap as Map;

    fn action(
        action_type: ActionType,
        tool: Option<&str>,
        params: Map<String, serde_json::Value>,
    ) -> Action {
        {
            let mut a = Action::new(action_type);
            a.id = "a1".to_string();
            a.tool = tool.map(str::to_string);
            a.parameters = params;
            a
        }
    }

    fn tool_call(tool: &str) -> Action {
        action(ActionType::ToolCall, Some(tool), Map::new())
    }

    fn tool_call_with(tool: &str, params: &[(&str, serde_json::Value)]) -> Action {
        let mut p = Map::new();
        for (k, v) in params {
            p.insert((*k).to_string(), v.clone());
        }
        action(ActionType::ToolCall, Some(tool), p)
    }

    #[test]
    fn tool_name_full_access_matches_irreversible_capabilities() {
        // Irreversible / externally-consequential tool names are full-access —
        // including the space-bearing-keyword cases a substring scan would MISS:
        // `git_reset`/`git_clean` have no bare `reset`/`clean` in the free-text
        // list, only `"git reset"`/`"git clean"` (with spaces), which can't occur
        // in an identifier. Segment matching catches them.
        for name in [
            "deploy",
            "git_push",
            "gitPush",
            "delete_file",
            "kubectl_apply",
            "git_reset",
            "git_clean",
            "send_email",
            "sudo_run",
            "rm_rf_dir",
            "drop_table",
            "upload_artifact",
        ] {
            assert!(
                tool_name_is_full_access(name),
                "{name} should classify as full-access"
            );
        }
    }

    #[test]
    fn tool_name_full_access_does_not_false_positive_on_benign_names() {
        // The collision cases that a raw substring scan of the free-text keyword
        // list gets WRONG. `count_tokens`/`tokenize` (vs `token`), `http_get`
        // (vs `http`), `apply_template` (vs `apply`), `request_id` (vs
        // `request`), `prefetch_cache` (vs `fetch`) must all be benign — segment
        // matching + a curated name set is what buys this.
        for name in [
            "read_file",
            "grep",
            "search",
            "summarize",
            "classify",
            "count_tokens",
            "tokenize",
            "token_usage",
            "http_get",
            "https_health",
            "apply_template",
            "request_id",
            "parse_request",
            "prefetch_cache",
            "transfer_learning",
            "format_date",
            "network_topology",
            "dropdown_open",
        ] {
            assert!(
                !tool_name_is_full_access(name),
                "{name} should NOT classify as full-access (false positive)"
            );
        }
    }

    #[test]
    fn name_segments_splits_snake_and_camel() {
        assert_eq!(name_segments("git_push"), vec!["git", "push"]);
        assert_eq!(name_segments("gitPush"), vec!["git", "push"]);
        assert_eq!(name_segments("git-push"), vec!["git", "push"]);
        assert_eq!(name_segments("count_tokens"), vec!["count", "tokens"]);
        // `tokens` is its own segment and never equals the `token` danger word.
        assert!(!name_segments("count_tokens").iter().any(|s| s == "token"));
    }

    #[test]
    fn any_tool_full_access_scans_the_palette() {
        // Works over &[&str], owned String collections, and is false on empty.
        assert!(any_tool_full_access(["read_file", "grep", "deploy"]));
        assert!(!any_tool_full_access(["read_file", "grep", "summarize"]));
        assert!(!any_tool_full_access(std::iter::empty::<&str>()));
        let owned: Vec<String> = vec!["read_file".into(), "git_push".into()];
        assert!(any_tool_full_access(&owned));
    }

    #[test]
    fn tier_ordering() {
        assert!(PermissionTier::FullAccess.covers(PermissionTier::ReadOnly));
        assert!(PermissionTier::SandboxEdit.covers(PermissionTier::SandboxEdit));
        assert!(!PermissionTier::ReadOnly.covers(PermissionTier::SandboxEdit));
    }

    #[test]
    fn classifier_baseline_by_type() {
        let c = RiskClassifier::new();
        assert_eq!(
            c.classify(&action(ActionType::StateRead, None, Map::new())),
            PermissionTier::ReadOnly
        );
        assert_eq!(
            c.classify(&action(ActionType::Assertion, None, Map::new())),
            PermissionTier::ReadOnly
        );
        assert_eq!(
            c.classify(&action(ActionType::StateWrite, None, Map::new())),
            PermissionTier::SandboxEdit
        );
        assert_eq!(c.classify(&tool_call("echo")), PermissionTier::SandboxEdit);
    }

    #[test]
    fn classifier_escalates_on_keyword_in_tool_name() {
        let c = RiskClassifier::new();
        assert_eq!(
            c.classify(&tool_call("deploy_service")),
            PermissionTier::FullAccess
        );
    }

    #[test]
    fn classifier_tool_name_signal_agrees_with_tool_name_is_full_access() {
        // Parslee-ai/car#917. These two answered the same question — "does this
        // tool name name a consequential capability?" — and disagreed on every
        // name below, because `classify` substring-scanned the command keyword
        // list over the name while `tool_name_is_full_access` segment-matched a
        // curated set. `http_get` was full-access to one and benign to the
        // other. Now `classify` calls it, so drift is not expressible.
        let c = RiskClassifier::new();
        for name in [
            "http_get",
            "https_health",
            "format_date",
            "count_tokens",
            "apply_template",
            "request_id",
            "prefetch_cache",
            "transfer_learning",
            "network_topology",
            "deploy_service",
            "send_email",
            "git_push",
            "read_secret",
        ] {
            let expected = if tool_name_is_full_access(name) {
                PermissionTier::FullAccess
            } else {
                PermissionTier::SandboxEdit
            };
            assert_eq!(c.classify(&tool_call(name)), expected, "tool name {name}");
        }
    }

    #[test]
    fn classifier_escalates_on_keyword_in_params() {
        let mut params = Map::new();
        params.insert("cmd".to_string(), serde_json::json!("rm -rf /tmp/x"));
        let a = action(ActionType::ToolCall, Some("shell"), params);
        assert_eq!(
            RiskClassifier::new().classify(&a),
            PermissionTier::FullAccess
        );
    }

    #[test]
    fn classifier_does_not_escalate_on_ordinary_text_in_ordinary_parameters() {
        // Parslee-ai/car#917: the four false positives the issue names, each of
        // which became a HARD REJECTION at Parslee-ai/car#915 — a `full_access`
        // classification is mandatory HITL regardless of granted tier, so on an
        // automated connection with no approver there is nobody to say yes.
        let c = RiskClassifier::new();
        let cases: Vec<(&str, Action)> = vec![
            (
                "a URL in a documented url parameter is not `curl` in a shell string",
                tool_call_with("http_get", &[("url", json!("https://example.com/a"))]),
            ),
            (
                "a search whose query text happens to say `release notes`",
                tool_call_with("search", &[("query", json!("release notes for 0.47"))]),
            ),
            (
                "a date formatter is not `mkfs --format`",
                tool_call_with(
                    "format_date",
                    &[("value", json!("2026-08-15")), ("fmt", json!("%Y-%m-%d"))],
                ),
            ),
            (
                "prose containing `send` in a body the tool merely stores",
                tool_call_with(
                    "create_note",
                    &[(
                        "body",
                        json!("Remember to send the report and apply the patch"),
                    )],
                ),
            ),
            (
                "max_tokens is in nearly every LLM tool call and is not a credential",
                tool_call_with(
                    "complete",
                    &[("prompt", json!("hello")), ("max_tokens", json!(256))],
                ),
            ),
            (
                "a body that MENTIONS ~/.ssh is not an action against it — only \
                 declared target paths carry the credential-path signal",
                tool_call_with(
                    "write_file",
                    &[
                        ("path", json!("/tmp/notes.md")),
                        ("content", json!("Back up ~/.ssh/id_rsa before you start")),
                    ],
                ),
            ),
        ];
        for (why, act) in cases {
            assert_eq!(
                c.classify(&act),
                PermissionTier::SandboxEdit,
                "must not escalate — {why}"
            );
        }
    }

    #[test]
    fn classifier_still_escalates_on_the_signals_that_matter() {
        // The other half of #917: narrowing must not let any of these through.
        // Each one reaches a DIFFERENT signal, so this pins all four.
        let c = RiskClassifier::new();
        let cases: Vec<(&str, Action)> =
            vec![
            (
                "tool name — segment match",
                tool_call_with("send_email", &[("to", json!("a@example.com"))]),
            ),
            (
                "command line — broad keyword over command/args only",
                tool_call_with(
                    "shell",
                    &[("command", json!("curl")), ("args", json!(["-X", "POST"]))],
                ),
            ),
            (
                "command line — `aws ` needs the trailing space and the argv join",
                tool_call_with(
                    "shell",
                    &[("command", json!("aws")), ("args", json!(["s3", "rb"]))],
                ),
            ),
            (
                "parameter key — the value is opaque, the key is not",
                tool_call_with(
                    "call_api",
                    &[("url", json!("https://x/y")), ("api_key", json!("sk-abc"))],
                ),
            ),
            (
                "parameter key — nested under an options object",
                tool_call_with("call_api", &[("opts", json!({"auth": {"token": "t"}}))]),
            ),
            (
                "parameter key — a segment match on a compound key",
                tool_call_with("connect", &[("db_password", json!("hunter2"))]),
            ),
            (
                "free text — destructive SQL under a `query` key the command surface cannot see",
                tool_call_with("run_sql", &[("query", json!("DROP TABLE users"))]),
            ),
            (
                "free text — `delete from` likewise",
                tool_call_with("run_sql", &[("query", json!("DELETE FROM orders WHERE 1=1"))]),
            ),
            (
                "free text — a destructive script passed as file contents",
                tool_call_with(
                    "write_file",
                    &[("path", json!("/tmp/x.sh")), ("contents", json!("rm -rf /"))],
                ),
            ),
            (
                "free text — history rewriting in a body parameter",
                tool_call_with("run_script", &[("body", json!("git reset --hard HEAD~5"))]),
            ),
            (
                "target path — reading someone's private key is a top-grant action",
                tool_call_with("read_file", &[("path", json!("~/.ssh/id_rsa"))]),
            ),
            (
                "target path — writing over a cloud credential file",
                tool_call_with(
                    "write_file",
                    &[("path", json!("/home/u/.aws/credentials")), ("content", json!("x"))],
                ),
            ),
        ];
        for (why, act) in cases {
            assert_eq!(
                c.classify(&act),
                PermissionTier::FullAccess,
                "must escalate — {why}"
            );
        }
    }

    #[test]
    fn command_shaped_actions_classify_exactly_as_they_did_before_917() {
        // The safety argument for #917 is that the broad keyword list was
        // re-aimed, not pruned: on a command line every entry still fires. If
        // this ever fails, the narrowing has reached the command surface and
        // the change is no longer behaviour-preserving where it claimed to be.
        let c = RiskClassifier::new();
        for kw in FULL_ACCESS_COMMAND_KEYWORDS {
            let act = tool_call_with("shell", &[("command", json!(format!("x{kw}y")))]);
            assert_eq!(
                c.classify(&act),
                PermissionTier::FullAccess,
                "command keyword {kw:?} must still escalate on a command line"
            );
        }
    }

    #[test]
    fn credential_param_keys_are_boundary_matched_not_substring_matched() {
        // `max_tokens` is the case that makes this worth testing: a raw
        // substring scan for `token` flags every LLM tool call in existence.
        for key in [
            "token",
            "api_key",
            "apiKey",
            "access-token",
            "client_secret",
            "db_password",
            "svc_credentials",
            "passphrase",
        ] {
            let act = tool_call_with("t", &[(key, json!("v"))]);
            assert!(
                hits_credential_param_key(&act),
                "{key} should read as credential material"
            );
        }
        for key in [
            "max_tokens",
            "tokens",
            "token_count",
            "key",
            "keys",
            "sort_key",
            "keyword",
            "secretariat_id",
        ] {
            let act = tool_call_with("t", &[(key, json!("v"))]);
            assert!(
                !hits_credential_param_key(&act),
                "{key} should NOT read as credential material (false positive)"
            );
        }
    }

    #[test]
    fn command_line_is_none_when_the_action_is_not_command_shaped() {
        assert_eq!(command_line(&tool_call("search")), None);
        assert_eq!(
            command_line(&tool_call_with("search", &[("query", json!("git push"))])),
            None,
            "a non-command parameter must not become the command line"
        );
        assert_eq!(
            command_line(&tool_call_with(
                "shell",
                &[("command", json!("git")), ("args", json!(["push"]))]
            )),
            Some("git push ".to_string()),
        );
    }

    #[test]
    fn custom_rule_only_raises() {
        let mut c = RiskClassifier::new();
        c.add_rule("flag_search", PermissionTier::FullAccess, |a| {
            a.tool.as_deref() == Some("search")
        });
        assert_eq!(c.classify(&tool_call("search")), PermissionTier::FullAccess);
        // A read action a rule matches at a lower tier stays at the
        // higher baseline — rules never lower.
        let mut c2 = RiskClassifier::new();
        c2.add_rule("noop", PermissionTier::ReadOnly, |_| true);
        assert_eq!(
            c2.classify(&action(ActionType::StateWrite, None, Map::new())),
            PermissionTier::SandboxEdit
        );
    }

    #[test]
    fn gate_allows_within_granted_tier() {
        let gate = PermissionGate::new(PermissionTier::SandboxEdit);
        let d = gate.evaluate(&action(ActionType::StateWrite, None, Map::new()));
        assert!(d.is_allow(), "{d:?}");
    }

    #[test]
    fn gate_escalates_above_granted_tier() {
        let gate = PermissionGate::new(PermissionTier::ReadOnly);
        let d = gate.evaluate(&action(ActionType::StateWrite, None, Map::new()));
        assert!(matches!(d, GateDecision::NeedsApproval { .. }), "{d:?}");
    }

    #[test]
    fn skill_ceiling_caps_below_granted_tier() {
        // The session can perform sandbox edits...
        let gate = PermissionGate::new(PermissionTier::SandboxEdit);
        let act = action(ActionType::StateWrite, None, Map::new()); // classifies SandboxEdit
        assert!(gate.evaluate(&act).is_allow());
        // ...but an action driven by a skill capped at read_only cannot: the
        // same action escalates to a human instead of running.
        let capped = gate.evaluate_with_ceiling(&act, Some(PermissionTier::ReadOnly));
        assert!(
            matches!(capped, GateDecision::NeedsApproval { .. }),
            "{capped:?}"
        );
        // No ceiling is identical to evaluate.
        assert!(gate.evaluate_with_ceiling(&act, None).is_allow());
    }

    #[test]
    fn skill_ceiling_at_or_above_granted_is_noop() {
        let gate = PermissionGate::new(PermissionTier::SandboxEdit);
        let act = action(ActionType::StateWrite, None, Map::new());
        // A ceiling at or above the granted tier leaves the outcome unchanged
        // (effective authority is min(granted, ceiling)).
        assert!(gate
            .evaluate_with_ceiling(&act, Some(PermissionTier::SandboxEdit))
            .is_allow());
        assert!(gate
            .evaluate_with_ceiling(&act, Some(PermissionTier::FullAccess))
            .is_allow());
    }

    #[test]
    fn gate_full_access_always_needs_approval_even_when_granted() {
        // Mandatory HITL: a FullAccess action is gated even for a
        // FullAccess session.
        let gate = PermissionGate::new(PermissionTier::FullAccess);
        let d = gate.evaluate(&tool_call("deploy"));
        assert!(matches!(d, GateDecision::NeedsApproval { .. }), "{d:?}");
    }

    #[test]
    fn approval_makes_future_evaluation_allow() {
        let mut gate = PermissionGate::new(PermissionTier::ReadOnly);
        let a = tool_call("deploy");
        assert!(matches!(
            gate.evaluate(&a),
            GateDecision::NeedsApproval { .. }
        ));
        gate.approve(&a, "matt", "reviewed the deploy plan", None)
            .unwrap();
        assert!(gate.evaluate(&a).is_allow());
    }

    #[test]
    fn rejection_denies_future_evaluation() {
        let mut gate = PermissionGate::new(PermissionTier::FullAccess);
        let a = tool_call("transfer_funds");
        gate.reject(&a, "matt", "not authorized", None).unwrap();
        assert!(matches!(gate.evaluate(&a), GateDecision::Deny { .. }));
    }

    #[test]
    fn shared_ledger_approval_is_visible_across_gates() {
        // The daemon substrate: per-session gates (tier state), ONE shared
        // ledger. An approval recorded through gate A's classification must
        // flip gate B's evaluation — with per-gate in-memory ledgers the
        // approver's decision would be stranded where the runner never reads
        // it (kernel review C1).
        let gate_a = PermissionGate::new(PermissionTier::ReadOnly);
        let gate_b = PermissionGate::new(PermissionTier::ReadOnly);
        let mut shared = ApprovalLedger::new();
        let a = tool_call("deploy");

        assert!(matches!(
            gate_b.evaluate_against(&a, &shared),
            GateDecision::NeedsApproval { .. }
        ));
        // Approver (gate A's session) builds the record; the SHARED ledger
        // stores it.
        let rec = gate_a.decision_record(&a, ApprovalDecision::Approved, "matt", "ok", None);
        shared.record(rec).unwrap();
        // Runner (gate B's session) sees it.
        assert!(gate_b.evaluate_against(&a, &shared).is_allow());
        // The gates' own (empty) ledgers still gate — nothing leaked into them.
        assert!(matches!(
            gate_b.evaluate(&a),
            GateDecision::NeedsApproval { .. }
        ));
    }

    #[test]
    fn ledger_record_decision_round_trips_fingerprint() {
        let mut ledger = ApprovalLedger::new();
        let rec = ledger
            .record_decision(
                "harness:retry:abcd1234",
                PermissionTier::SandboxEdit,
                ApprovalDecision::Approved,
                "conn:1",
                "reviewed",
                None,
            )
            .unwrap();
        assert_eq!(rec.fingerprint, "harness:retry:abcd1234");
        assert_eq!(
            ledger.lookup("harness:retry:abcd1234").map(|r| r.decision),
            Some(ApprovalDecision::Approved)
        );
    }

    #[test]
    fn classifier_escalates_on_argv_array_command() {
        // The dangerous command is split across an argv array, so the raw
        // JSON never contains the literal "git push" substring — the
        // recursive haystack must still catch it.
        let mut params = Map::new();
        params.insert(
            "args".to_string(),
            serde_json::json!(["git", "push", "--force", "origin", "main"]),
        );
        let a = action(ActionType::ToolCall, Some("shell"), params);
        assert_eq!(
            RiskClassifier::new().classify(&a),
            PermissionTier::FullAccess
        );
    }

    #[test]
    fn fingerprint_canonicalizes_nested_object_key_order() {
        // Two semantically identical actions whose params differ only in
        // NESTED object key order must share a fingerprint — otherwise a
        // standing rejection is evadable by permuting nested keys.
        let mk = |json: serde_json::Value| {
            let mut p = Map::new();
            p.insert("opts".to_string(), json);
            action(ActionType::ToolCall, Some("t"), p)
        };
        let a = mk(serde_json::json!({"a": 1, "b": {"x": 1, "y": 2}}));
        let b = mk(serde_json::json!({"b": {"y": 2, "x": 1}, "a": 1}));
        assert_eq!(action_fingerprint(&a), action_fingerprint(&b));
    }

    #[test]
    fn fingerprint_uses_stable_type_tag_not_debug() {
        // The fingerprint must carry the stable serde tag, never the
        // Debug spelling.
        let fp = action_fingerprint(&tool_call("x"));
        assert!(fp.starts_with("tool_call|"), "got {fp}");
        assert!(!fp.contains("ToolCall"));
    }

    #[test]
    fn approval_does_not_survive_upward_reclassification() {
        // An operation approved at SandboxEdit must NOT remain allowed
        // after a classifier change pushes it to FullAccess.
        let a = tool_call("safe_tool");
        // Record an approval at SandboxEdit (the tier when reviewed).
        let mut classifier = RiskClassifier::new();
        let mut gate = PermissionGate::new(PermissionTier::SandboxEdit).with_classifier(classifier);
        gate.approve(&a, "matt", "looked fine", None).unwrap();
        assert!(gate.evaluate(&a).is_allow());

        // Now a stricter classifier reclassifies the same op as FullAccess.
        classifier = RiskClassifier::new();
        classifier.add_rule("now_dangerous", PermissionTier::FullAccess, |act| {
            act.tool.as_deref() == Some("safe_tool")
        });
        let gate = gate.with_classifier(classifier);
        assert!(
            matches!(gate.evaluate(&a), GateDecision::NeedsApproval { .. }),
            "stale low-tier approval must not bypass the FullAccess gate"
        );
    }

    #[test]
    fn fingerprint_is_param_sensitive_and_stable() {
        let a1 = {
            let mut p = Map::new();
            p.insert("x".to_string(), serde_json::json!(1));
            p.insert("y".to_string(), serde_json::json!(2));
            action(ActionType::ToolCall, Some("t"), p)
        };
        let a2 = {
            // same params, inserted in a different order → same fingerprint
            let mut p = Map::new();
            p.insert("y".to_string(), serde_json::json!(2));
            p.insert("x".to_string(), serde_json::json!(1));
            action(ActionType::ToolCall, Some("t"), p)
        };
        assert_eq!(action_fingerprint(&a1), action_fingerprint(&a2));
        // different params → different fingerprint
        let a3 = {
            let mut p = Map::new();
            p.insert("x".to_string(), serde_json::json!(99));
            action(ActionType::ToolCall, Some("t"), p)
        };
        assert_ne!(action_fingerprint(&a1), action_fingerprint(&a3));
    }

    #[test]
    fn ledger_journal_round_trips() {
        let dir = std::env::temp_dir();
        let path = dir.join(format!("car-approvals-test-{}.jsonl", std::process::id()));
        let _ = std::fs::remove_file(&path);

        let a = tool_call("deploy");
        {
            let ledger = ApprovalLedger::with_journal(&path).unwrap();
            let mut gate = PermissionGate::new(PermissionTier::ReadOnly).with_ledger(ledger);
            gate.approve(&a, "matt", "ok", Some("diff: +1 -0".to_string()))
                .unwrap();
        }
        // A fresh ledger loading the same journal sees the decision.
        let ledger2 = ApprovalLedger::with_journal(&path).unwrap();
        let gate2 = PermissionGate::new(PermissionTier::ReadOnly).with_ledger(ledger2);
        assert!(gate2.evaluate(&a).is_allow());

        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn unwritable_journal_errors_and_stores_nothing() {
        // Review A7: a journal write failure must surface as an error, and
        // the decision must NOT land in memory either — otherwise callers
        // emit ApprovalRecorded audit events for a decision that isn't
        // durable. A directory as the journal path makes the append fail.
        let dir = std::env::temp_dir().join(format!("car-approvals-dir-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();

        let mut ledger = ApprovalLedger {
            records: HashMap::new(),
            journal: Some(dir.clone()),
            skipped_on_load: 0,
        };
        let a = tool_call("deploy");
        let fp = action_fingerprint(&a);
        let err = ledger.record(ApprovalRecord {
            fingerprint: fp.clone(),
            required_tier: PermissionTier::FullAccess,
            decision: ApprovalDecision::Approved,
            reviewer: "matt".into(),
            reason: "ok".into(),
            evidence: None,
            decided_at: chrono::Utc::now().to_rfc3339(),
        });
        assert!(err.is_err(), "journal write failure must surface");
        assert!(
            ledger.lookup(&fp).is_none(),
            "failed record must not be stored in memory"
        );

        let _ = std::fs::remove_dir_all(&dir);
    }

    // --- The second axis: reversibility -----------------------------------

    /// The reason the axis exists. Every row asserts **both** classifiers,
    /// because a test that only checked reversibility would pass just as well
    /// if `classify_reversibility` were secretly derived from the tier — and
    /// that derivation is the bug this work removes.
    #[test]
    fn the_two_axes_disagree_in_both_directions() {
        let cases: Vec<(&str, Action, PermissionTier, Reversibility)> = vec![
            (
                "a state read observes and mutates nothing",
                action(ActionType::StateRead, None, Map::new()),
                PermissionTier::ReadOnly,
                Reversibility::Reversible,
            ),
            (
                "a scratch-file write is undone by discarding the sandbox",
                tool_call_with(
                    "write_file",
                    &[
                        ("path", serde_json::json!("/tmp/car-sandbox/notes.txt")),
                        ("contents", serde_json::json!("hello")),
                    ],
                ),
                PermissionTier::SandboxEdit,
                Reversibility::Reversible,
            ),
            (
                "reading a credential takes the top grant and leaves nothing to undo",
                tool_call("read_secret"),
                PermissionTier::FullAccess,
                Reversibility::Reversible,
            ),
            (
                "a row insert reads as low-authority but needs a compensating delete",
                tool_call_with(
                    "db_insert",
                    &[
                        ("table", serde_json::json!("orders")),
                        ("row", serde_json::json!({"id": 7})),
                    ],
                ),
                PermissionTier::SandboxEdit,
                Reversibility::Compensable,
            ),
            (
                "a push takes the top grant AND is recoverable — force-push the prior ref",
                tool_call("git_push"),
                PermissionTier::FullAccess,
                Reversibility::Compensable,
            ),
            (
                "a sent email takes the top grant and is permanent",
                tool_call_with(
                    "send_email",
                    &[
                        ("to", serde_json::json!("ops@example.com")),
                        ("subject", serde_json::json!("nightly status")),
                    ],
                ),
                PermissionTier::FullAccess,
                Reversibility::Irreversible,
            ),
            (
                "a charged card takes the top grant and is permanent",
                tool_call_with("charge_card", &[("amount_cents", serde_json::json!(4200))]),
                PermissionTier::FullAccess,
                Reversibility::Irreversible,
            ),
            (
                "a write outside the sandbox reads as low-authority and is unrecoverable",
                tool_call_with(
                    "write_file",
                    &[
                        ("path", serde_json::json!("/etc/hosts")),
                        ("contents", serde_json::json!("127.0.0.1 x")),
                    ],
                ),
                PermissionTier::SandboxEdit,
                Reversibility::Irreversible,
            ),
        ];

        let classifier = RiskClassifier::new();
        for (why, act, tier, rev) in cases {
            assert_eq!(classifier.classify(&act), tier, "required tier — {why}");
            assert_eq!(classify_reversibility(&act), rev, "reversibility — {why}");
        }
    }

    #[test]
    fn evaluate_axes_agrees_with_computing_the_axes_separately() {
        // #856 shares one flattened haystack between the two classifiers. That
        // is only safe if it changes no verdict, so pin both axes against the
        // independent paths across shapes that exercise the tier keywords, the
        // reversibility phrases, and the sandbox rule.
        let cases = vec![
            tool_call("send_email"),
            tool_call("read_secret"),
            tool_call_with(
                "shell",
                &[
                    ("command", json!("git")),
                    ("args", json!(["push", "--force", "origin", "main"])),
                ],
            ),
            tool_call_with("write_file", &[("path", json!("/tmp/scratch/a.txt"))]),
            tool_call_with("write_file", &[("path", json!("/etc/hosts"))]),
            tool_call_with("execute_query", &[("sql", json!("UPDATE t SET x = 1"))]),
            action(ActionType::StateRead, None, Map::new()),
            action(ActionType::StateWrite, None, Map::new()),
        ];

        for granted in [
            PermissionTier::ReadOnly,
            PermissionTier::SandboxEdit,
            PermissionTier::FullAccess,
        ] {
            let gate = PermissionGate::new(granted);
            for a in &cases {
                let axes = gate.evaluate_axes(a, None, None);
                assert_eq!(
                    axes.decision,
                    gate.evaluate(a),
                    "decision drifted for {:?} at {granted:?}",
                    a.tool
                );
                assert_eq!(
                    axes.reversibility,
                    classify_reversibility(a),
                    "reversibility drifted for {:?}",
                    a.tool
                );
            }
        }

        // The ceiling argument still caps standing authority.
        let gate = PermissionGate::new(PermissionTier::FullAccess);
        let a = tool_call("deploy_service");
        assert_eq!(
            gate.evaluate_axes(&a, Some(PermissionTier::ReadOnly), None)
                .decision,
            gate.evaluate_with_ceiling(&a, Some(PermissionTier::ReadOnly)),
        );
    }

    #[test]
    fn neither_axis_is_a_function_of_the_other() {
        // The structural claim, stated twice: one tier spans all three rollback
        // contracts, and one rollback contract spans all three tiers. No
        // mapping in either direction could reproduce both.
        let c = RiskClassifier::new();

        for (act, rev) in [
            (tool_call("read_secret"), Reversibility::Reversible),
            (tool_call("git_push"), Reversibility::Compensable),
            (tool_call("send_email"), Reversibility::Irreversible),
        ] {
            let name = act.tool.clone().unwrap_or_default();
            assert_eq!(c.classify(&act), PermissionTier::FullAccess, "tier {name}");
            assert_eq!(classify_reversibility(&act), rev, "reversibility {name}");
        }

        for (act, tier) in [
            (
                action(ActionType::StateRead, None, Map::new()),
                PermissionTier::ReadOnly,
            ),
            (
                action(ActionType::StateWrite, None, Map::new()),
                PermissionTier::SandboxEdit,
            ),
            (tool_call("read_secret"), PermissionTier::FullAccess),
        ] {
            assert_eq!(
                classify_reversibility(&act),
                Reversibility::Reversible,
                "reversibility {:?}",
                act.tool
            );
            assert_eq!(c.classify(&act), tier, "tier {:?}", act.tool);
        }
    }

    #[test]
    fn state_actions_are_settled_by_their_type() {
        for kind in [ActionType::StateRead, ActionType::Assertion] {
            assert_eq!(
                classify_reversibility(&action(kind, None, Map::new())),
                Reversibility::Reversible
            );
        }
        // A state write's whole footprint is the KV store, which snapshot /
        // rollback restores — so it stays `Reversible` even when its *value*
        // happens to read like something destructive. The keyword ladder only
        // runs for tool calls, whose effects are the opaque ones.
        let mut p = Map::new();
        p.insert("value".to_string(), serde_json::json!("rm -rf /"));
        assert_eq!(
            classify_reversibility(&action(ActionType::StateWrite, None, p)),
            Reversibility::Reversible
        );
    }

    #[test]
    fn a_generic_shell_is_classified_from_its_argv() {
        // The signal that fires when the tool name says nothing at all. Neither
        // phrase exists in the raw JSON — the argv array only becomes matchable
        // text once it is flattened.
        let destructive = tool_call_with(
            "shell",
            &[("args", serde_json::json!(["rm", "-rf", "/var/data"]))],
        );
        let recoverable = tool_call_with(
            "shell",
            &[("args", serde_json::json!(["git", "push", "origin", "main"]))],
        );
        assert_eq!(
            classify_reversibility(&destructive),
            Reversibility::Irreversible
        );
        assert_eq!(
            classify_reversibility(&recoverable),
            Reversibility::Compensable
        );

        // Both are FullAccess: the authority ladder cannot tell them apart,
        // which is exactly the gap the second axis fills.
        let c = RiskClassifier::new();
        assert_eq!(c.classify(&destructive), PermissionTier::FullAccess);
        assert_eq!(c.classify(&recoverable), PermissionTier::FullAccess);
    }

    #[test]
    fn severity_wins_when_both_families_match() {
        // `create` is compensable and `payment` is irreversible; the ladder is
        // most-severe-first, so the permanent reading wins.
        assert_eq!(
            classify_reversibility(&tool_call("create_payment")),
            Reversibility::Irreversible
        );
        // A retrieval verb does not rescue a destructive one — step 2 requires
        // the name to carry NO mutating segment.
        assert_eq!(
            classify_reversibility(&tool_call("get_and_delete")),
            Reversibility::Irreversible
        );
        assert_eq!(
            classify_reversibility(&tool_call("list_and_push")),
            Reversibility::Compensable
        );
    }

    #[test]
    fn sandbox_confinement_is_all_or_nothing() {
        // One absolute path outside scratch space disqualifies the action, even
        // though another is inside it.
        let mixed = tool_call_with(
            "write_file",
            &[
                ("src", serde_json::json!("/tmp/car-sandbox/in.txt")),
                ("dst", serde_json::json!("/etc/hosts")),
            ],
        );
        assert_eq!(classify_reversibility(&mixed), Reversibility::Irreversible);

        // A relative path is invisible to the rule (no cwd to resolve it
        // against), so it neither qualifies nor disqualifies — and the action
        // falls through to the conservative default.
        let relative = tool_call_with("write_file", &[("path", serde_json::json!("notes.txt"))]);
        assert_eq!(
            classify_reversibility(&relative),
            Reversibility::Irreversible
        );
    }

    #[test]
    fn the_sandbox_rule_does_not_leak_to_non_filesystem_tools() {
        // An incidental scratch path in a network tool's parameters must not
        // talk that tool down to `Reversible`.
        let a = tool_call_with(
            "http_post",
            &[("body_file", serde_json::json!("/tmp/payload.json"))],
        );
        assert_eq!(classify_reversibility(&a), Reversibility::Irreversible);
    }

    // ---- Regressions for the four classifier defects found in review. ----
    // Each of these fails on the pre-fix classifier; the failure direction is
    // the one the module docs commit to never taking (a destructive action
    // reported as recoverable), except the first, which failed in BOTH.

    #[test]
    fn classification_is_deterministic_across_parameter_orderings() {
        // `Action::parameters` is a std HashMap with per-process randomized
        // iteration. Flattening it directly made a phrase form between two
        // unrelated parameters on some draws and not others: a force-push came
        // back `compensable` ~285/500 and `irreversible` ~215/500 in one
        // process. Same action, same call, different audit row.
        //
        // Rebuilding the map many times exercises fresh orderings; the answer
        // must not move.
        let mk = || {
            tool_call_with(
                "shell",
                &[
                    ("command", json!("git")),
                    ("args", json!(["push", "--force", "origin", "main"])),
                    ("cwd", json!("/srv/app")),
                    ("timeout", json!(30)),
                ],
            )
        };
        let first = classify_reversibility(&mk());
        for i in 0..256 {
            assert_eq!(
                classify_reversibility(&mk()),
                first,
                "reversibility moved on iteration {i}"
            );
        }
        // And the command line is genuinely reconstructed: `git push` is only
        // spellable across the `command`/`args` boundary.
        assert_eq!(first, Reversibility::Compensable);
    }

    #[test]
    fn command_and_args_stay_adjacent_so_rm_rf_is_still_seen() {
        // The counterpart risk to the fix above: separating parameters to stop
        // phantom phrases must not lose the ONE cross-parameter adjacency that
        // is real. Sorting keys alphabetically puts `args` before `command` and
        // silently drops `rm -rf` — a deterministic miss, worse than a random
        // one, and in the unsafe direction.
        let a = tool_call_with(
            "shell",
            &[
                ("command", json!("rm")),
                ("args", json!(["-rf", "/var/data"])),
            ],
        );
        assert_eq!(classify_reversibility(&a), Reversibility::Irreversible);
    }

    #[test]
    fn a_read_verb_in_the_name_cannot_override_a_mutation_in_the_parameters() {
        // Step 2 is the only path that reaches `Reversible` without positive
        // evidence of confinement, and it used to fire on the tool NAME alone
        // without ever inspecting the parameters. `query` is the literal tool
        // name the MCP Postgres connector exposes.
        for (tool, params) in [
            ("execute_query", json!("UPDATE accounts SET balance = 0")),
            ("db_query", json!("ALTER TABLE users DROP COLUMN email")),
            ("query", json!("INSERT INTO audit VALUES (1)")),
            ("search_index", json!("DROP INDEX idx_users")),
        ] {
            let a = tool_call_with(tool, &[("sql", params)]);
            assert_ne!(
                classify_reversibility(&a),
                Reversibility::Reversible,
                "{tool} carries a mutation and must not classify as reversible"
            );
        }
        // A find that deletes what it finds.
        let a = tool_call_with(
            "find",
            &[
                ("command", json!("find")),
                ("args", json!(["/data", "-name", "*.log", "-delete"])),
            ],
        );
        assert_ne!(classify_reversibility(&a), Reversibility::Reversible);
        // The control: a genuine read still short-circuits.
        let a = tool_call_with("execute_query", &[("sql", json!("SELECT id FROM users"))]);
        assert_eq!(classify_reversibility(&a), Reversibility::Reversible);
    }

    #[test]
    fn only_os_scratch_roots_count_as_discardable() {
        // `SANDBOX_PATH_MARKERS` matched the bare substrings "sandbox" and
        // "scratch" anywhere in a path token, so ordinary production
        // directories that happen to contain those words read as throwaway.
        for path in [
            "/srv/sandbox-prod/index.html",
            "/System/Library/Sandbox/Profiles/x.sb",
            "/opt/scratch-data/customers.db",
            "/var/lib/sandbox/state.json",
            // `contains` rather than `starts_with` also accepted this one.
            "/etc/tmp/hosts",
        ] {
            let a = tool_call_with("write_file", &[("path", json!(path))]);
            assert_ne!(
                classify_reversibility(&a),
                Reversibility::Reversible,
                "{path} is not OS-designated scratch space"
            );
        }
        // Real scratch roots still qualify.
        for path in ["/tmp/build/out.txt", "/private/var/folders/xy/z/T/a.txt"] {
            let a = tool_call_with("write_file", &[("path", json!(path))]);
            assert_eq!(
                classify_reversibility(&a),
                Reversibility::Reversible,
                "{path} is scratch space"
            );
        }
    }

    #[test]
    fn an_incidental_scratch_path_cannot_vouch_for_a_write_elsewhere() {
        // Step 5 read every string in the parameters, so a `/tmp` path in a
        // file's CONTENTS licensed a write whose actual target was a relative
        // path the classifier has already said it cannot resolve.
        let a = tool_call_with(
            "write_file",
            &[
                ("path", json!("config.yaml")),
                ("contents", json!("cache_dir: /tmp/app\n")),
            ],
        );
        assert_eq!(
            classify_reversibility(&a),
            Reversibility::Irreversible,
            "the target is a relative path; the /tmp string is incidental"
        );

        // Same shape, but the target really is scratch space.
        let a = tool_call_with(
            "write_file",
            &[
                ("path", json!("/tmp/app/config.yaml")),
                ("contents", json!("cache_dir: ./app\n")),
            ],
        );
        assert_eq!(classify_reversibility(&a), Reversibility::Reversible);

        // A scratch target plus a non-scratch target is still not discardable.
        let a = tool_call_with(
            "write_file",
            &[("paths", json!(["/tmp/a.txt", "/etc/hosts"]))],
        );
        assert_eq!(classify_reversibility(&a), Reversibility::Irreversible);

        // ...and the veto does not depend on the dangerous path arriving under
        // a key `PATH_PARAM_KEYS` happens to know. `dst` is not in that set; if
        // only recognized keys were scanned, `/etc/hosts` would be invisible
        // and the scratch `src` would wrongly vouch for the whole action.
        let a = tool_call_with(
            "write_file",
            &[
                ("src", json!("/tmp/in.txt")),
                ("dst", json!("/etc/hosts")),
                ("mode", json!("0644")),
            ],
        );
        assert_eq!(
            classify_reversibility(&a),
            Reversibility::Irreversible,
            "an unrecognized key must still be able to veto"
        );
    }

    #[test]
    fn unrecognized_tools_default_to_irreversible() {
        for name in ["frobnicate", "acme_widget", "run"] {
            assert_eq!(
                classify_reversibility(&tool_call(name)),
                Reversibility::Irreversible,
                "{name} is unrecognized and must assume the worst"
            );
        }
    }

    #[test]
    fn documented_over_classifications_stay_documented() {
        // The misses named in `classify_reversibility`'s "What this is not"
        // section, pinned so the docs cannot drift away from the behavior.
        // Both err toward the conservative answer, which is the point.
        assert_eq!(
            classify_reversibility(&tool_call("list_deployments")),
            Reversibility::Compensable,
            "a read over a mutating noun over-classifies"
        );
        assert_eq!(
            classify_reversibility(&tool_call("get_payment")),
            Reversibility::Irreversible,
            "money is matched by its object, not its verb"
        );
    }
}