openlatch-client 0.1.18

//! Local policy evaluation — the leaf module behind the daemon's verdict path.
//!
//! Holds the resident policy bundle (the rule set fetched from the platform and
//! cached on disk), the two-metacharacter glob matcher, and the deterministic
//! evaluator that turns a `pre_tool_use` envelope into an allow / deny decision.
//! Evaluation is local-authoritative: a resident bundle keeps enforcing while the
//! host is offline, and a failed refresh is fail-static — the last-known-good
//! bundle stays active.
//!
//! # Module boundary (hard)
//!
//! This is a **leaf**. It must not import `daemon/`, `cli/`, or a sibling
//! `core/` module, with exactly two documented exceptions:
//!
//! 1. `crate::core::error` — for the `ERR_BUNDLE_*` code constants. Hardcoding
//!    the `"OL-1211"` literals here would put the registry out of one place.
//! 2. `store.rs` takes its base directory as a `PathBuf` **parameter** rather
//!    than calling `crate::config::openlatch_dir()`. The caller (the poller /
//!    `AppState` construction) resolves the path, which also makes the store
//!    testable against a `tempfile::TempDir`.
//!
//! The wire types (`PolicyBundle`, `PolicyRule`, …) come from
//! `crate::generated::types`, the typify output, **not** from
//! `crate::core::envelope` — that module only re-exports them and importing it
//! would breach the leaf rule for no benefit.
//!
//! `core/cloud/` and `core/policy/` must not import each other in either
//! direction. `daemon/handlers.rs` is the only place both leaves are visible,
//! which is why the CloudEvent extension stamping lives there.
//!
//! The module is declared `#[cfg(feature = "full-cli")]` in `src/core/mod.rs`:
//! it depends on `sha2` and `arc-swap`, both `full-cli`-gated, and must never be
//! compiled into `openlatch-hook`.

pub mod evaluate;
pub mod matcher;
pub mod store;
pub mod validate;

use std::sync::Arc;
use std::time::{Duration, SystemTime};

use arc_swap::ArcSwap;
// `serde_json` deserializes straight out of a `&Value`, which is what spares
// `parse_bundle_tolerant` a deep clone of every rule it only means to inspect.
use serde::Deserialize as _;

use crate::core::error::ERR_RULE_SKIPPED;
use crate::generated::types::{
    AgentFunction, PolicyBundle, PolicyRule, PolicyRuleConditionsItem,
    PolicyRuleConditionsItemField, PolicyRuleConditionsItemOp, PolicyRuleMode, PolicyRuleSeverity,
};

pub use evaluate::{evaluate, evaluate_command, normalize, PolicyMatch, SHELL_TOOL_NAMES};
pub use matcher::matches;
pub use store::{BundleMeta, CachedBundle, StoreError};
pub use validate::validate_rule;

/// The `rules[].kind` evaluated on the `pre_tool_use` hot path.
///
/// A rule carrying a kind this client does not implement is **skipped** and the
/// rest of the bundle stays active, so a v1 client tolerates a v1.1 bundle (PRD
/// § "Edge Cases", row *Unknown `rule.kind`*). The document is never rejected
/// for it.
pub const RULE_KIND_COMMAND: &str = "command";

/// The `rules[].kind` targeting an outbound model request at the boundary.
///
/// Held and exposed, never evaluated (D-U17): nothing modifies a live request in
/// this phase (D-26). [`ResidentBundle::request_rules`] is where the boundary
/// listener reads them from.
pub const RULE_KIND_REQUEST: &str = "request";

/// The only `rules[].action` this client can carry out on the command plane.
///
/// `action` is an open string on the wire "for the same forward-compatibility
/// reason as `kind`" (see `schemas/policy-bundle.schema.json`), so the same
/// skip applies: a v1 client that silently turned a future
/// `action: "require_approval"` into a hard **deny** would be producing a
/// verdict the rule's author never asked for.
///
/// Which actions are legal on the request plane is **not** mirrored here — the
/// schema gate owns that list, and a second copy in Rust is exactly the drift
/// D-U7 exists to prevent.
pub const RULE_ACTION_DENY: &str = "deny";

/// One evaluation-ready rule.
///
/// This is deliberately *not* the generated [`PolicyRule`]: by the time a rule
/// reaches this struct its `kind` and `action` have been recognised and dropped,
/// so the evaluator has no forward-compat branches left to get wrong.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Rule {
    /// Stable public identifier, e.g. `OL-CMD-001`. Also the tiebreak key: the
    /// lexicographically first rule *within the deciding set* is reported.
    pub rule_id: String,
    /// Fully-anchored glob, matched by [`matcher::matches`].
    pub match_pattern: String,
    /// `observe` records a shadow verdict and allows; `enforce` denies.
    pub mode: PolicyRuleMode,
    /// Author-assigned severity, surfaced on the verdict returned to the hook.
    pub severity: PolicyRuleSeverity,
    /// Plain-language explanation shown to the developer verbatim on a deny.
    pub reason: String,
    /// Which agent functions this rule applies to, or `None` when it applies to
    /// every agent.
    ///
    /// Projected once, here, from the wire rule's `conditions[]` — the AND of
    /// every `agent.function in <set>` condition, which for `in`-sets is their
    /// intersection (see [`scoped_functions`]). Holding the reduced set rather
    /// than the raw conditions keeps this struct `Eq` (the generated
    /// [`AgentFunction`] is `Copy + Eq`) and leaves the evaluator one membership
    /// test with no vocabulary branches left to get wrong.
    ///
    /// Three states, and the distinction is the whole feature:
    ///
    /// - `None` — unconditional; the rule matches on any agent, exactly as
    ///   every rule did before `conditions` existed.
    /// - `Some(set)` — the rule participates only when the resident bundle's
    ///   [`ResidentBundle::agent_function`] is `Some(f)` with `f` in `set`.
    ///   [`AgentFunction::Unknown`] is an ordinary member here, never a
    ///   wildcard: a rule scoped to `["unknown"]` matches only an agent whose
    ///   context says `unknown`.
    /// - `Some(empty)` — the conditions were disjoint; the rule matches nothing.
    ///   Kept rather than dropped so the resident bundle still reports the rule
    ///   the author wrote (a scoped rule that reaches no agent is an authoring
    ///   fact, not a load failure).
    pub scoped_functions: Option<Vec<AgentFunction>>,
}

/// The bundle the daemon is currently enforcing — the type the [`ArcSwap`]
/// holds.
///
/// It carries everything the verdict path needs to stamp the six `ol*`
/// extensions **without touching disk or the poller**. `handlers.rs` reads this
/// and nothing else on the hot path.
///
/// `last_fetch_ok` (→ `olpolicyoffline`) is deliberately **not** a field here:
/// it is poll state, not bundle state, and it changes without the bundle
/// changing. It lives in a separate `Arc<AtomicBool>` the poller writes and
/// `handlers.rs` reads.
///
/// `Eq` is deliberately absent from the derive: [`PolicyRule`] is typify output
/// and `build.rs` adds only `PartialEq` to it, so `Vec<PolicyRule>: Eq` does not
/// hold. Nothing in the tree needs `Eq` on a resident bundle. The internal
/// [`Rule`] keeps it — all of its fields are `Eq`.
#[derive(Debug, Clone, PartialEq)]
pub struct ResidentBundle {
    /// The command plane, already filtered — rules that failed the schema gate
    /// or carried an unrecognised `kind`/`action` were dropped by
    /// [`ResidentBundle::from_bundle`]. This is the only set [`evaluate`] sees,
    /// so the `pre_tool_use` path gained no branch and the lexicographic
    /// tie-break is unchanged by construction rather than by care (D-U19).
    pub command_rules: Vec<Rule>,
    /// The request plane, held in the wire form and **not evaluated** (D-U17).
    /// Kept on this struct rather than in a second store so there is one
    /// lifecycle and one hot-swap path — the daemon already republishes the
    /// whole `ResidentBundle` on every refresh (D-U21).
    pub request_rules: Vec<PolicyRule>,
    /// The organization-wide kill switch. `false` forces every match to shadow.
    pub enforcement_enabled: bool,
    /// → `olpolicybundlerev`.
    pub revision: i64,
    /// → `olpolicybundleage` (`now - built_at`, clamped at 0).
    pub built_at: SystemTime,
    /// The org this bundle belongs to; the poller checks later bundles against
    /// it (trust-on-first-use).
    pub organization_id: String,
    /// The org's identity-capture switch, carried on `client_config` (I-1 D-02).
    ///
    /// **Fail closed, and only on an explicit `true`.** A bundle with no
    /// `client_config`, a `client_config` that omits the key, or a client that
    /// has never received a bundle at all are all the same posture: capture
    /// nothing. That is why the projection below reads `unwrap_or(false)` twice
    /// rather than defaulting the generated struct — "absent" must never be
    /// distinguishable from "off" on the capture path.
    ///
    /// This is the ONLY switch. There is deliberately no config.toml key and no
    /// CLI flag (D-12): a local opt-out would race the organization's decision,
    /// and the unrecognized-key reporter would rightly flag it.
    pub capture_identity_signals: bool,
    /// This install's business function, from `client_config.agent_context
    /// .function` — the value the platform composed for THIS agent when it
    /// served the bundle (I-4 C-02).
    ///
    /// **`None` is not [`AgentFunction::Unknown`].** `Unknown` is a real
    /// platform-assigned value ("nobody has claimed this agent") and a rule
    /// scoped to it matches exactly those agents. `None` means the context
    /// never arrived — no `client_config`, no `agent_context`, no `function`, or
    /// a **string** value this build cannot parse — and every scoped rule then
    /// matches nothing while unconditional rules keep enforcing. That is the
    /// designed degrade for a pre-provisioning poll (no agent id ⇒ the platform
    /// serves the org row without context) and for a vocabulary this client
    /// predates.
    ///
    /// The degrade is scoped to the *vocabulary*, not to the *shape*: the open
    /// string buys tolerance for `"astrology"`, not for `"function": 42` or a
    /// non-object `agent_context`. Those fail `PolicyBundle` deserialization
    /// and the whole bundle is rejected (`OL-1212`, last-known-good keeps
    /// enforcing) — the same exposure `capture_identity_signals` has carried
    /// since I-1, because `client_config` is one typed object. The platform is
    /// the only writer and emits an enum at source, so the shape is its
    /// contract to keep.
    ///
    /// The wire field is an open string, not the enum, on purpose:
    /// `client_config` is deserialized as one typed object outside the per-rule
    /// tolerance, so an enum there would fail the WHOLE bundle on a single
    /// out-of-vocabulary value. Parsing happens once, in
    /// [`ResidentBundle::from_bundle`]; the evaluator compares enums only.
    pub agent_function: Option<AgentFunction>,
}

impl ResidentBundle {
    /// Project a wire [`PolicyBundle`] into the resident form.
    ///
    /// Five things happen here and nowhere else:
    ///
    /// 1. **Gate validation.** Every rule is checked individually against the
    ///    rule subschema (D-U13) and a failure skips that rule alone. Validating
    ///    the document instead would fail the whole bundle on one bad rule and
    ///    cost the organization its denies — the inversion the prime invariant
    ///    exists to prevent.
    /// 2. **Recognition and partition.** Rules whose `kind` or `action` this
    ///    client does not implement are skipped; the survivors are split into
    ///    the command and request planes (D-U19).
    /// 3. **Phase-1 mode coercion.** A `kind: request` rule authored `enforce`
    ///    is held as `observe`, because nothing acts on the request plane yet
    ///    (D-26). Listed here because it is the one step expected to be
    ///    **deleted** when the boundary listener starts consuming these rules —
    ///    it lives in `project_rule`'s request arm.
    /// 4. **`built_at` parsing.** The schema declares `built_at` a plain string
    ///    (not `format: date-time`) precisely so the generated type stays a
    ///    `String` and the conversion happens once, here, instead of on every
    ///    verdict. An unparsable value is not fatal — it degrades
    ///    `olpolicybundleage` to `0` rather than disarming the host, which is
    ///    what "never stop enforcing on a metadata problem" means in practice.
    /// 5. **Agent-context parsing.** `client_config.agent_context.function` is
    ///    an open string on the wire; it is parsed into [`AgentFunction`] here,
    ///    once. Absent, or a string this build cannot parse, reads `None` —
    ///    scoped rules match nothing, the bundle stays active — for the same
    ///    reason as (4): a context problem must never disarm the host. A
    ///    `function` that is not a string at all never reaches this step: it
    ///    fails deserialization upstream, like any other malformed
    ///    `client_config` (see [`ResidentBundle::agent_function`]).
    ///
    /// This is the single projection choke point: the network path
    /// (`policy_poller.rs`) and the disk-cache path (`daemon/mod.rs`) both come
    /// through here, and `store.rs` only deserializes.
    pub fn from_bundle(bundle: &PolicyBundle) -> Self {
        let mut command_rules = Vec::new();
        let mut request_rules = Vec::new();

        for wire in &bundle.rules {
            if let Err(detail) = validate::validate_rule(wire) {
                skip_rule(&wire.rule_id, "gate_violation", &detail);
                continue;
            }
            match project_rule(wire) {
                Some(Plane::Command(rule)) => command_rules.push(rule),
                Some(Plane::Request(rule)) => request_rules.push(rule),
                None => {}
            }
        }

        let built_at = parse_built_at(&bundle.built_at).unwrap_or_else(|| {
            tracing::warn!(
                target: "policy",
                built_at = %bundle.built_at,
                "bundle built_at is not RFC 3339; reporting bundle age as 0"
            );
            SystemTime::now()
        });
        // The client-bound channel, navigated once: both fields below read it,
        // and every way the agent context can be *absent* is a `None` on this
        // chain — so the parser under it only ever sees a value that arrived.
        let client_config = bundle.client_config.as_ref();
        Self {
            command_rules,
            request_rules,
            enforcement_enabled: bundle.enforcement_enabled,
            revision: bundle.revision,
            built_at,
            organization_id: bundle.organization_id.clone(),
            capture_identity_signals: client_config
                .and_then(|c| c.capture_identity_signals)
                .unwrap_or(false),
            agent_function: client_config
                .and_then(|c| c.agent_context.as_ref())
                .and_then(|c| c.function.as_deref())
                .and_then(parse_agent_function),
        }
    }

    /// `now - built_at` in whole seconds, **clamped at 0** so a host whose clock
    /// sits behind the platform's never reports a negative `olpolicybundleage`.
    pub fn age_seconds(&self, now: SystemTime) -> u64 {
        now.duration_since(self.built_at)
            .unwrap_or(Duration::ZERO)
            .as_secs()
    }
}

/// Above this, [`parse_agent_function`] logs a placeholder in place of the
/// value. The length always goes out.
const MAX_LOGGED_FUNCTION_LEN: usize = 64;

/// Parse one *present* `client_config.agent_context.function` into the closed
/// vocabulary.
///
/// Every absent shape — no `client_config`, no `agent_context`, no `function` —
/// is filtered by the call site in [`ResidentBundle::from_bundle`], which is
/// the pre-I-4 bundle shape and the pre-provisioning poll, both entirely
/// normal and both silent. So reaching here means the platform sent a value,
/// and returning `None` means this build could not read it. That is not
/// normal: this build predates the vocabulary the platform is using, so every
/// scoped rule on this host silently stops matching. Without the line below
/// that is invisible from the host — the bundle activates, the rules are all
/// there, and nothing ever fires.
///
/// The value itself is shown only when it is short enough to be a vocabulary
/// word. `function` is an open string on the wire, so its length is bounded
/// only by the bundle size, and a runaway value would otherwise land whole in
/// the daemon log on every swap.
fn parse_agent_function(raw: &str) -> Option<AgentFunction> {
    match raw.parse::<AgentFunction>() {
        Ok(function) => Some(function),
        Err(_) => {
            let shown = if raw.len() > MAX_LOGGED_FUNCTION_LEN {
                "<oversized>"
            } else {
                raw
            };
            tracing::warn!(
                target: "policy",
                function = %shown,
                function_len = raw.len(),
                "agent_context.function is not a known AgentFunction; scoped rules will match nothing"
            );
            None
        }
    }
}

/// Which plane a recognised rule lands on.
///
/// The variants differ substantially in size — the wire `PolicyRule` carries
/// the whole request-plane shape — but boxing the larger one would be a
/// pessimisation, not a fix. `Plane` is a transient dispatch result: it is
/// constructed and immediately destructured inside [`ResidentBundle::from_bundle`],
/// never stored in a collection and never held across an await. Boxing would
/// add a heap allocation per rule to save stack bytes on a path that runs once
/// per bundle swap (300s by default), and the value would be unboxed again on
/// the very next line to push into `request_rules`.
#[allow(clippy::large_enum_variant)]
enum Plane {
    Command(Rule),
    Request(PolicyRule),
}

/// One rule dropped; the bundle stays active. See [`ERR_RULE_SKIPPED`] for the
/// closed set of `reason` values.
fn skip_rule(rule_id: &str, reason: &'static str, detail: &str) {
    tracing::warn!(
        target: "policy",
        code = ERR_RULE_SKIPPED,
        rule_id = %rule_id,
        reason,
        detail = %detail,
        "skipping one rule; the rest of the bundle stays active"
    );
}

/// Recognise one wire rule and route it to its plane, or drop it.
fn project_rule(rule: &PolicyRule) -> Option<Plane> {
    match rule.kind.as_str() {
        RULE_KIND_COMMAND => {
            if rule.action != RULE_ACTION_DENY {
                skip_rule(&rule.rule_id, "unknown_action", &rule.action);
                return None;
            }
            // The gate requires `match_pattern` on this plane, so `None` here
            // means the gate did not get to run — a broken embedded schema, not
            // a bad rule. Skip rather than substitute a pattern: a wrong glob
            // denies commands nobody asked to deny.
            let Some(match_pattern) = rule.match_pattern.clone() else {
                skip_rule(
                    &rule.rule_id,
                    "gate_violation",
                    "kind=command carries no match_pattern",
                );
                return None;
            };
            let scoped_functions = scoped_functions(&rule.conditions);
            // Kept, not dropped: the author wrote conditions that cannot all
            // hold at once (`in [sales]` AND `in [legal]`), which the gate
            // cannot catch — each condition is individually valid. The rule is
            // inert rather than wrong, so it stays, and the line is what tells
            // whoever wrote it that it will never fire.
            //
            // Deliberately NOT an `ERR_RULE_SKIPPED` line: the rule is neither
            // dropped nor weakened, and that code's `reason` set exists to count
            // lost enforcement coverage.
            if scoped_functions.as_ref().is_some_and(|s| s.is_empty()) {
                tracing::warn!(
                    target: "policy",
                    rule_id = %rule.rule_id,
                    "rule conditions intersect to no agent function; it can never match on any agent"
                );
            }
            Some(Plane::Command(Rule {
                rule_id: rule.rule_id.clone(),
                match_pattern,
                mode: rule.mode,
                severity: rule.severity,
                reason: rule.reason.clone(),
                scoped_functions,
            }))
        }
        RULE_KIND_REQUEST => {
            // D-U9's client half. Nothing acts on the request plane in this
            // phase (D-26), so an authored `enforce` is held as `observe`.
            //
            // The coercion is written into the stored rule, not merely logged.
            // Leaving `mode` as authored would make "held as observe" true only
            // for as long as nothing reads `request_rules` — the first consumer
            // (I-3's boundary listener) would read `Enforce` and act on it, and
            // the guarantee would have been a comment rather than a fact.
            let mut held = rule.clone();
            if held.mode == PolicyRuleMode::Enforce {
                tracing::warn!(
                    target: "policy",
                    code = ERR_RULE_SKIPPED,
                    rule_id = %rule.rule_id,
                    reason = "enforce_coerced",
                    "request rule authored enforce is held as observe; nothing acts on the request plane in this phase"
                );
                held.mode = PolicyRuleMode::Observe;
            }
            Some(Plane::Request(held))
        }
        unknown => {
            skip_rule(&rule.rule_id, "unknown_kind", unknown);
            None
        }
    }
}

/// Reduce a wire rule's `conditions[]` to the set of agent functions it applies
/// to — `None` when it applies to every agent.
///
/// Every condition must hold (AND semantics, per the schema). v1 has one field
/// and one op, so the AND of `agent.function in S1`, `agent.function in S2`, …
/// is membership in `S1 ∩ S2 ∩ …`; that intersection is what gets stored. The
/// result is sorted and deduplicated so two rules authored with the same set in
/// a different order project to equal [`Rule`]s.
///
/// The `match` on `(field, op)` is exhaustive over the generated enums on
/// purpose: when the schema widens the vocabulary, the regenerated types make
/// this a compile error, and whoever adds the arm decides how the new
/// predicate composes rather than having it silently reduce to a set the
/// author did not write. Until a client carries that arm, the schema gate and
/// `parse_bundle_tolerant` drop such a rule per-rule (`unrecognized_field`) —
/// nothing unknown reaches here.
fn scoped_functions(conditions: &[PolicyRuleConditionsItem]) -> Option<Vec<AgentFunction>> {
    let mut scope: Option<Vec<AgentFunction>> = None;
    for condition in conditions {
        match (condition.field, condition.op) {
            (PolicyRuleConditionsItemField::AgentFunction, PolicyRuleConditionsItemOp::In) => {
                let allowed = &condition.value;
                scope = Some(match scope {
                    None => allowed.clone(),
                    Some(mut so_far) => {
                        so_far.retain(|f| allowed.contains(f));
                        so_far
                    }
                });
            }
        }
    }
    scope.map(|mut set| {
        set.sort_unstable();
        set.dedup();
        set
    })
}

/// Parse a bundle document, dropping only the rules this build cannot fully read.
///
/// [`PolicyRule`] is typify-generated from a schema with
/// `additionalProperties: false`, so it carries `#[serde(deny_unknown_fields)]`.
/// One rule carrying a key from a newer schema therefore fails deserialization
/// of the **whole document**: the bundle is rejected as `ERR_BUNDLE_INVALID`,
/// and — because a failed refresh is fail-static — the host keeps enforcing
/// whatever it had. A fleet on an older client silently stops receiving policy
/// the moment anyone authors a rule using a new key.
///
/// That contradicts the contract [`RULE_KIND_COMMAND`] states above: an
/// unrecognised `kind` skips one rule and "the document is never rejected for
/// it". An unrecognised **field** now behaves the same way. It is also what
/// makes an additive schema change genuinely non-breaking rather than
/// non-breaking on paper.
///
/// The rule is **dropped**, not partially honoured. Ignoring the unknown key and
/// keeping the rest of the rule would be worse: a key this build does not
/// implement is typically a *narrowing*, and running a narrowed rule without its
/// narrowing applies it to more traffic than its author wrote it for.
///
/// Everything else still fails the way it did — a malformed document, a bad
/// `schema_version`, a rule the per-kind gate rejects.
pub fn parse_bundle_tolerant(
    mut doc: serde_json::Value,
) -> Result<PolicyBundle, serde_json::Error> {
    if let Some(rules) = doc
        .get_mut("rules")
        .and_then(serde_json::Value::as_array_mut)
    {
        // Borrowed, not `from_value(raw.clone())`: the deserialized rule is
        // discarded (the real one is built by the whole-document pass below),
        // so deep-cloning each rule's `Value` just to inspect it is pure cost.
        rules.retain(|raw| match PolicyRule::deserialize(raw) {
            Ok(_) => true,
            Err(e) => {
                skip_rule(
                    raw.get("rule_id")
                        .and_then(serde_json::Value::as_str)
                        .unwrap_or("<unnamed>"),
                    "unrecognized_field",
                    &e.to_string(),
                );
                false
            }
        });
    }
    serde_json::from_value(doc)
}

/// Parse the bundle's RFC 3339 UTC `built_at` (microsecond precision allowed).
fn parse_built_at(raw: &str) -> Option<SystemTime> {
    chrono::DateTime::parse_from_rfc3339(raw)
        .ok()
        .map(|dt| SystemTime::from(dt.with_timezone(&chrono::Utc)))
}

/// The shared handle the poller writes and the verdict path reads.
///
/// `None` means **no bundle has ever been activated** — the fail-open case, in
/// which the daemon allows everything and marks the event as having no policy.
/// It is *not* the same as a resident bundle with an empty `rules[]`, which is
/// "allow everything, revision N" and is distinguishable in telemetry.
///
/// An in-flight evaluation must see one consistent bundle, which is exactly what
/// the atomic pointer swap buys: the poller never mutates a resident bundle in
/// place, it publishes a whole new one.
pub type PolicyHandle = Arc<ArcSwap<Option<ResidentBundle>>>;

/// Build a [`PolicyHandle`], optionally pre-populated from the disk load that
/// runs before `axum::serve` starts accepting requests.
pub fn new_handle(initial: Option<ResidentBundle>) -> PolicyHandle {
    Arc::new(ArcSwap::from_pointee(initial))
}

#[cfg(test)]
pub(crate) mod test_support {
    //! Fixtures shared by the tests in this module tree.

    use super::*;
    use crate::generated::types::{ChurnLayer, PolicyRuleParams, PolicyRuleSelect};

    pub fn wire_rule(
        rule_id: &str,
        pattern: &str,
        mode: PolicyRuleMode,
        severity: PolicyRuleSeverity,
    ) -> PolicyRule {
        PolicyRule {
            action: RULE_ACTION_DENY.to_string(),
            conditions: Vec::new(),
            kind: RULE_KIND_COMMAND.to_string(),
            match_pattern: Some(pattern.to_string()),
            mode,
            params: None,
            reason: format!("{rule_id} says no"),
            rule_id: rule_id.to_string(),
            rule_version: None,
            select: None,
            severity,
        }
    }

    /// A `kind: command` wire rule carrying `conditions[]` — enforce/high, the
    /// shape every conditions test wants. Named because "a wire rule with
    /// conditions" is the fixture; building one by mutating an unconditional
    /// rule spelled the same three lines at every call site.
    pub fn scoped_wire_rule(
        rule_id: &str,
        pattern: &str,
        conditions: Vec<PolicyRuleConditionsItem>,
    ) -> PolicyRule {
        PolicyRule {
            conditions,
            ..wire_rule(
                rule_id,
                pattern,
                PolicyRuleMode::Enforce,
                PolicyRuleSeverity::High,
            )
        }
    }

    /// A `kind: request` rule, valid under the gate for the given action.
    pub fn wire_request_rule(rule_id: &str, action: &str, mode: PolicyRuleMode) -> PolicyRule {
        let (select, params) = match action {
            // Only the fields the gate makes meaningful for each action are set;
            // the rest default, so a reader does not have to work out which of
            // the explicit `None`s were a deliberate choice.
            "prefix_reorder" => (
                Some(PolicyRuleSelect {
                    exclude_layers: vec![ChurnLayer::Tools],
                    model_in: vec!["claude-opus-5".to_string()],
                    ..Default::default()
                }),
                None,
            ),
            "history_trim" => (
                Some(PolicyRuleSelect {
                    min_messages: Some(40),
                    ..Default::default()
                }),
                Some(PolicyRuleParams {
                    keep_messages: Some(20),
                    ..Default::default()
                }),
            ),
            "prompt_edit" => (
                None,
                Some(PolicyRuleParams {
                    marker: Some("<!--openlatch-->".to_string()),
                    ..Default::default()
                }),
            ),
            other => panic!("no fixture for request action {other}"),
        };

        PolicyRule {
            action: action.to_string(),
            conditions: Vec::new(),
            kind: RULE_KIND_REQUEST.to_string(),
            match_pattern: None,
            mode,
            params,
            reason: format!("{rule_id} reshapes the request"),
            rule_id: rule_id.to_string(),
            rule_version: Some(3),
            select,
            severity: PolicyRuleSeverity::Low,
        }
    }

    pub fn wire_bundle(rules: Vec<PolicyRule>, enforcement_enabled: bool) -> PolicyBundle {
        PolicyBundle {
            built_at: "2026-07-21T09:00:00Z".to_string(),
            client_config: None,
            enforcement_enabled,
            organization_id: "0192f8a1-4c3b-7e2a-9f10-5d8c3b1a7e42".to_string(),
            revision: 42,
            rules,
            schema_version: 1,
            signature: None,
        }
    }

    pub fn rule(rule_id: &str, pattern: &str, mode: PolicyRuleMode) -> Rule {
        Rule {
            rule_id: rule_id.to_string(),
            match_pattern: pattern.to_string(),
            mode,
            severity: PolicyRuleSeverity::High,
            reason: format!("{rule_id} says no"),
            scoped_functions: None,
        }
    }

    /// One `agent.function in <functions>` wire condition.
    pub fn function_in(functions: &[AgentFunction]) -> PolicyRuleConditionsItem {
        PolicyRuleConditionsItem {
            field: PolicyRuleConditionsItemField::AgentFunction,
            op: PolicyRuleConditionsItemOp::In,
            value: functions.to_vec(),
        }
    }

    /// Keep every callsite live for the rest of the test process.
    ///
    /// `with_default` is thread-local, but whether a callsite is live *at all*
    /// is not. `tracing` caches one `Interest` per callsite for the whole
    /// process, and `tracing-core`'s `rebuild_callsite_interest` computes it
    /// from whichever dispatcher is default on the thread that first reaches
    /// that callsite — falling back to `Interest::never()` when it finds none.
    ///
    /// The `warn!` inside [`skip_rule`] is shared with tests that call
    /// `from_bundle` *without* a subscriber (`from_bundle_skips_unknown_kind_and_action`
    /// and its neighbours), so on a parallel run one of those can win the race
    /// to touch it first and cache it as `never`. From then on the macro
    /// short-circuits before consulting any thread-local subscriber:
    /// [`capture_logs`] captures nothing and the assertion reads an empty
    /// string. That is exactly how `gate_violation_skips_one_rule` failed under
    /// `cargo tarpaulin` while passing under plain `cargo test` — ptrace
    /// instrumentation widens the window enough to make the race reliable.
    ///
    /// One permissive global default closes both halves. Constructing its
    /// `Dispatch` rebuilds the interest cache, healing any callsite already
    /// cached as `never`; and afterwards every thread has a dispatcher that
    /// answers "yes", so no callsite can be cached as dead again. The
    /// thread-local subscriber in [`capture_logs`] still wins for the capture
    /// itself.
    fn keep_callsites_live() {
        static ONCE: std::sync::Once = std::sync::Once::new();
        ONCE.call_once(|| {
            // An `Err` here means some other global default is already
            // installed, which serves the same purpose — the callsites are
            // live either way, so there is nothing to recover from.
            let _ = tracing::subscriber::set_global_default(
                tracing_subscriber::fmt()
                    .with_writer(std::io::sink)
                    .with_max_level(tracing::Level::TRACE)
                    .finish(),
            );
        });
    }

    /// Capture everything `tracing` emits on this thread while `f` runs.
    ///
    /// `OL-1214` — and D-U9's `enforce_coerced` in particular — is observable
    /// only as a log line: the RFC asks for "an error code on bundle activation
    /// naming each coerced rule", so asserting on the line is the only honest
    /// test of it. See [`keep_callsites_live`] for why a global subscriber is
    /// installed before the thread-local one.
    pub fn capture_logs<T>(f: impl FnOnce() -> T) -> (T, String) {
        use std::io::Write;
        use std::sync::Mutex;

        #[derive(Clone, Default)]
        struct Buffer(Arc<Mutex<Vec<u8>>>);

        impl Write for Buffer {
            fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
                self.0.lock().expect("log buffer").extend_from_slice(buf);
                Ok(buf.len())
            }
            fn flush(&mut self) -> std::io::Result<()> {
                Ok(())
            }
        }

        impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for Buffer {
            type Writer = Self;
            fn make_writer(&'a self) -> Self::Writer {
                self.clone()
            }
        }

        keep_callsites_live();

        let buffer = Buffer::default();
        let subscriber = tracing_subscriber::fmt()
            .with_writer(buffer.clone())
            .with_max_level(tracing::Level::DEBUG)
            .with_ansi(false)
            .finish();
        let out = tracing::subscriber::with_default(subscriber, f);
        let logs = String::from_utf8(buffer.0.lock().expect("log buffer").clone())
            .expect("log output is utf-8");
        (out, logs)
    }

    pub fn resident(rules: Vec<Rule>, enforcement_enabled: bool) -> ResidentBundle {
        ResidentBundle {
            command_rules: rules,
            request_rules: Vec::new(),
            enforcement_enabled,
            revision: 42,
            built_at: SystemTime::UNIX_EPOCH + Duration::from_secs(1_784_000_000),
            organization_id: "0192f8a1-4c3b-7e2a-9f10-5d8c3b1a7e42".to_string(),
            capture_identity_signals: false,
            agent_function: None,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::test_support::*;
    use super::*;

    #[test]
    fn from_bundle_keeps_known_rules() {
        let bundle = wire_bundle(
            vec![wire_rule(
                "OL-CMD-001",
                "*rm -rf*",
                PolicyRuleMode::Enforce,
                PolicyRuleSeverity::Critical,
            )],
            true,
        );
        let resident = ResidentBundle::from_bundle(&bundle);

        assert_eq!(resident.command_rules.len(), 1);
        assert_eq!(resident.command_rules[0].rule_id, "OL-CMD-001");
        assert_eq!(
            resident.command_rules[0].severity,
            PolicyRuleSeverity::Critical
        );
        assert_eq!(resident.revision, 42);
        assert!(resident.enforcement_enabled);
        assert_eq!(
            resident.organization_id,
            "0192f8a1-4c3b-7e2a-9f10-5d8c3b1a7e42"
        );
    }

    /// Forward compat: a v1.1 bundle carrying a kind we do not know must not
    /// take the whole document down with it.
    #[test]
    fn from_bundle_skips_unknown_kind_and_action() {
        let mut unknown_kind = wire_rule(
            "OL-NET-001",
            "*curl*",
            PolicyRuleMode::Enforce,
            PolicyRuleSeverity::High,
        );
        unknown_kind.kind = "network".to_string();

        let mut unknown_action = wire_rule(
            "OL-CMD-002",
            "*sudo*",
            PolicyRuleMode::Enforce,
            PolicyRuleSeverity::High,
        );
        unknown_action.action = "require_approval".to_string();

        let known = wire_rule(
            "OL-CMD-001",
            "*rm -rf*",
            PolicyRuleMode::Enforce,
            PolicyRuleSeverity::High,
        );

        let bundle = wire_bundle(vec![unknown_kind, unknown_action, known], true);
        let resident = ResidentBundle::from_bundle(&bundle);

        let ids: Vec<&str> = resident
            .command_rules
            .iter()
            .map(|r| r.rule_id.as_str())
            .collect();
        assert_eq!(ids, vec!["OL-CMD-001"]);
    }

    /// The field-level half of the same contract. An unrecognised `kind` already
    /// skipped one rule; an unrecognised KEY used to reject the whole document,
    /// which made every additive schema change a real break for older clients.
    #[test]
    fn a_rule_with_an_unrecognised_field_is_dropped_and_the_bundle_survives() {
        let doc = serde_json::json!({
            "schema_version": 1,
            "revision": 42,
            "built_at": "2026-07-21T09:00:00Z",
            "organization_id": "0192f8a1-4c3b-7e2a-9f10-5d8c3b1a7e42",
            "enforcement_enabled": true,
            "signature": null,
            "rules": [
                {
                    "rule_id": "OL-FUTURE-001",
                    "kind": "command",
                    "action": "deny",
                    "match_pattern": "*curl*",
                    "mode": "enforce",
                    "severity": "high",
                    "reason": "authored against a newer schema",
                    "unknown_key_from_the_future": { "nested": true }
                },
                {
                    "rule_id": "OL-CMD-001",
                    "kind": "command",
                    "action": "deny",
                    "match_pattern": "*rm -rf*",
                    "mode": "enforce",
                    "severity": "high",
                    "reason": "OL-CMD-001 says no"
                }
            ]
        });

        let bundle = parse_bundle_tolerant(doc).expect("the bundle still parses");
        let ids: Vec<&str> = bundle.rules.iter().map(|r| r.rule_id.as_str()).collect();
        assert_eq!(
            ids,
            vec!["OL-CMD-001"],
            "the future rule is dropped; every rule this build CAN read survives"
        );
        assert!(
            bundle.enforcement_enabled,
            "the rest of the document is untouched"
        );
    }

    /// Tolerance is scoped to the rules array. A document that is malformed in
    /// any other way must still be rejected — otherwise the guard would swallow
    /// the corruption it exists to distinguish itself from.
    #[test]
    fn a_malformed_envelope_still_fails() {
        let doc = serde_json::json!({
            "schema_version": 1,
            "revision": "not-a-number",
            "built_at": "2026-07-21T09:00:00Z",
            "organization_id": "0192f8a1-4c3b-7e2a-9f10-5d8c3b1a7e42",
            "enforcement_enabled": true,
            "rules": []
        });
        assert!(parse_bundle_tolerant(doc).is_err());
    }

    #[test]
    fn built_at_parses_rfc3339_with_microseconds() {
        let mut bundle = wire_bundle(vec![], true);
        bundle.built_at = "2026-07-21T09:00:00.123456Z".to_string();
        let resident = ResidentBundle::from_bundle(&bundle);
        let secs = resident
            .built_at
            .duration_since(SystemTime::UNIX_EPOCH)
            .expect("after epoch")
            .as_secs();
        // 2026-07-21T09:00:00Z; the sub-second remainder is preserved but not
        // asserted on — only whole seconds reach `olpolicybundleage`.
        assert_eq!(secs, 1_784_624_400);
    }

    #[test]
    fn unparsable_built_at_does_not_disarm_the_bundle() {
        let mut bundle = wire_bundle(
            vec![wire_rule(
                "OL-CMD-001",
                "*rm*",
                PolicyRuleMode::Enforce,
                PolicyRuleSeverity::High,
            )],
            true,
        );
        bundle.built_at = "not a timestamp".to_string();
        let resident = ResidentBundle::from_bundle(&bundle);
        assert_eq!(resident.command_rules.len(), 1);
        assert_eq!(resident.age_seconds(SystemTime::now()), 0);
    }

    /// Clock skew must never produce a negative `olpolicybundleage`.
    #[test]
    fn age_is_clamped_at_zero_when_the_clock_is_behind() {
        let bundle = resident(vec![], true);
        let before_built = bundle.built_at - Duration::from_secs(3_600);
        assert_eq!(bundle.age_seconds(before_built), 0);
        assert_eq!(
            bundle.age_seconds(bundle.built_at + Duration::from_secs(90)),
            90
        );
    }

    #[test]
    fn handle_starts_empty_and_hot_swaps() {
        let handle = new_handle(None);
        assert!(handle.load().is_none());

        handle.store(Arc::new(Some(resident(
            vec![rule("OL-CMD-001", "*rm*", PolicyRuleMode::Enforce)],
            true,
        ))));

        let loaded = handle.load();
        let bundle = loaded.as_ref().as_ref().expect("bundle resident");
        assert_eq!(bundle.command_rules.len(), 1);
    }

    // -- client_config / identity capture (I-1 D-02) -------------------------

    /// The shared wire fixture, serialized, with `client_config` set to whatever
    /// the caller passes — or left off entirely when passed `None`.
    ///
    /// Goes back through JSON rather than setting the generated field directly
    /// so the *wire* shape is what gets exercised: these cases are about what the
    /// platform sends, and `parse_bundle_tolerant` is the code that receives it.
    fn bundle_doc(client_config: Option<serde_json::Value>) -> serde_json::Value {
        let mut doc = serde_json::to_value(wire_bundle(vec![], true)).expect("fixture serializes");
        if let Some(cc) = client_config {
            doc.as_object_mut()
                .expect("object")
                .insert("client_config".to_string(), cc);
        }
        doc
    }

    fn capture_flag_of(client_config: Option<serde_json::Value>) -> bool {
        let bundle =
            parse_bundle_tolerant(bundle_doc(client_config)).expect("the bundle document parses");
        ResidentBundle::from_bundle(&bundle).capture_identity_signals
    }

    /// D-02, the whole posture in one test: capture is on for exactly one input
    /// — an explicit `true`. Every other shape of the document, including the
    /// one every bundle in the fleet carries today, means capture nothing.
    #[test]
    fn identity_capture_is_off_unless_the_bundle_says_true() {
        assert!(
            !capture_flag_of(None),
            "no client_config at all — the shape every pre-I-1 bundle has"
        );
        assert!(
            !capture_flag_of(Some(serde_json::json!({}))),
            "client_config present but silent on the key"
        );
        assert!(!capture_flag_of(Some(
            serde_json::json!({"capture_identity_signals": false})
        )));
        assert!(capture_flag_of(Some(
            serde_json::json!({"capture_identity_signals": true})
        )));
    }

    /// `client_config` is `additionalProperties: true` on purpose: a future
    /// client-bound key must not repeat the D-03 breakage one level down, where
    /// an older client would reject the whole bundle over a field it simply does
    /// not read yet.
    #[test]
    fn an_unknown_key_inside_client_config_is_ignored_not_fatal() {
        assert!(capture_flag_of(Some(serde_json::json!({
            "capture_identity_signals": true,
            "some_future_client_knob": {"nested": ["shape"]}
        }))));
    }

    // -- agent context + conditions (I-4 C-02 / C-04) -----------------------

    fn agent_function_of(client_config: Option<serde_json::Value>) -> Option<AgentFunction> {
        let bundle =
            parse_bundle_tolerant(bundle_doc(client_config)).expect("the bundle document parses");
        ResidentBundle::from_bundle(&bundle).agent_function
    }

    /// C-02. The open wire string is parsed into the enum once, here; every
    /// shape in which the context can fail to arrive reads `None`.
    #[test]
    fn agent_function_projects_from_client_config() {
        assert_eq!(
            agent_function_of(Some(serde_json::json!({
                "agent_context": {"function": "marketing"}
            }))),
            Some(AgentFunction::Marketing)
        );
        assert_eq!(
            agent_function_of(Some(serde_json::json!({
                "capture_identity_signals": true,
                "agent_context": {"function": "it_ops"}
            }))),
            Some(AgentFunction::ItOps),
            "the snake_case wire spelling parses"
        );
        assert_eq!(
            agent_function_of(Some(serde_json::json!({
                "agent_context": {"function": "unknown"}
            }))),
            Some(AgentFunction::Unknown),
            "`unknown` is a real value, not the absence of one"
        );

        // The absent-channel / absent-key / absent-field triple.
        assert_eq!(agent_function_of(None), None, "no client_config at all");
        assert_eq!(
            agent_function_of(Some(serde_json::json!({}))),
            None,
            "client_config without agent_context"
        );
        assert_eq!(
            agent_function_of(Some(serde_json::json!({"agent_context": {}}))),
            None,
            "agent_context without function"
        );
    }

    /// The whole reason `function` is an open string on the wire: a value this
    /// build cannot parse costs the context, not the bundle. Scoped rules step
    /// aside; the unconditional deny beside them keeps denying.
    #[test]
    fn an_unparsable_agent_function_is_none_not_fatal() {
        let scoped = scoped_wire_rule(
            "OL-CMD-002",
            "*psql*",
            vec![function_in(&[AgentFunction::Marketing])],
        );
        let unconditional = wire_rule(
            "OL-CMD-001",
            "*rm -rf*",
            PolicyRuleMode::Enforce,
            PolicyRuleSeverity::Critical,
        );
        // The shared envelope with only `rules` swapped: the two rules are the
        // subject, and building the envelope by hand would let this test drift
        // onto a different document from its neighbours.
        let mut doc = bundle_doc(Some(
            serde_json::json!({"agent_context": {"function": "astrology"}}),
        ));
        doc["rules"] = serde_json::to_value([scoped, unconditional]).expect("rules serialize");

        let bundle = parse_bundle_tolerant(doc).expect("the bundle survives");
        let resident = ResidentBundle::from_bundle(&bundle);

        assert_eq!(resident.agent_function, None);
        assert_eq!(resident.command_rules.len(), 2, "no rule is dropped");
        assert!(
            evaluate_command(&resident, "sudo rm -rf /tmp").is_some_and(|m| !m.shadow),
            "the unconditional deny still blocks"
        );
        assert!(
            evaluate_command(&resident, "psql -h prod").is_none(),
            "the scoped rule matches nothing on an install with no readable context"
        );

        assert_eq!(
            agent_function_of(Some(serde_json::json!({"agent_context": {"function": ""}}))),
            None,
            "an empty string is no context either"
        );

        // The tolerance is over the VOCABULARY, not the shape. `client_config`
        // is one typed object outside the per-rule channel, so a `function`
        // that is not a string — or an `agent_context` that is not an object —
        // fails the whole document, exactly as a malformed
        // `capture_identity_signals` has since I-1. Documented on
        // `ResidentBundle::agent_function`; asserted here so the claim cannot
        // rot into "any bad value is survivable".
        for shape in [
            serde_json::json!({"agent_context": {"function": 42}}),
            serde_json::json!({"agent_context": "marketing"}),
        ] {
            assert!(
                parse_bundle_tolerant(bundle_doc(Some(shape.clone()))).is_err(),
                "expected a whole-document failure for {shape}"
            );
        }
    }

    /// …and it is not swallowed. Losing the context silently is the failure
    /// mode this warn exists for: the bundle activates, every rule is present,
    /// and the scoped ones simply stop firing on this host with nothing
    /// anywhere to say why. The value rides in the line because "which word did
    /// the platform send that this build predates" is the whole diagnosis.
    #[test]
    fn an_unparsable_agent_function_says_so_in_the_log() {
        let (function, logs) = capture_logs(|| {
            agent_function_of(Some(
                serde_json::json!({"agent_context": {"function": "astrology"}}),
            ))
        });
        assert_eq!(function, None);
        assert!(logs.contains("astrology"), "{logs}");
        assert!(logs.contains("not a known AgentFunction"), "{logs}");

        // A value long enough to be a payload rather than a vocabulary word is
        // reported by length — an open string is bounded only by the bundle
        // size, and the daemon log must not carry it whole on every swap.
        let long = "x".repeat(MAX_LOGGED_FUNCTION_LEN + 1);
        let (function, logs) = capture_logs(|| {
            agent_function_of(Some(
                serde_json::json!({"agent_context": {"function": long}}),
            ))
        });
        assert_eq!(function, None);
        assert!(
            !logs.contains("xxxxx"),
            "the value itself must not appear: {logs}"
        );
        assert!(
            logs.contains(&format!("function_len={}", MAX_LOGGED_FUNCTION_LEN + 1)),
            "{logs}"
        );

        // The absent shapes stay silent — a pre-I-4 bundle is not a problem.
        let (_, logs) = capture_logs(|| agent_function_of(None));
        assert!(!logs.contains("not a known AgentFunction"), "{logs}");
    }

    /// L7. Two `in` sets that do not overlap reduce to an empty scope: every
    /// condition is individually valid, so the gate passes the rule and it is
    /// kept — inert. Nothing else in the system would ever mention it again.
    #[test]
    fn a_rule_whose_conditions_cannot_all_hold_is_kept_and_reported() {
        let dead = scoped_wire_rule(
            "OL-CMD-002",
            "*psql*",
            vec![
                function_in(&[AgentFunction::Sales]),
                function_in(&[AgentFunction::Legal]),
            ],
        );

        let (resident, logs) =
            capture_logs(|| ResidentBundle::from_bundle(&wire_bundle(vec![dead], true)));

        assert_eq!(resident.command_rules.len(), 1, "kept, not dropped");
        assert_eq!(
            resident.command_rules[0].scoped_functions,
            Some(Vec::new()),
            "the intersection is empty"
        );
        assert!(logs.contains("OL-CMD-002"), "{logs}");
        assert!(logs.contains("can never match"), "{logs}");
        assert!(
            !logs.contains(ERR_RULE_SKIPPED),
            "an inert rule is not lost enforcement coverage: {logs}"
        );
    }

    /// C-04. A condition vocabulary this build predates drops that one rule at
    /// deserialization — `unrecognized_field`, the same channel as an unknown
    /// key — and the rest of the bundle stays active. All three closed enums
    /// inside a condition are exercised (`field`, `op`, and `value`'s
    /// `AgentFunction`), because the `OL-1214` reason table promises the row
    /// covers each of them.
    #[test]
    fn a_rule_with_an_unknown_condition_field_is_dropped_not_fatal() {
        // The shared envelope with only `rules` swapped: the four rules are the
        // subject here, and hand-writing the envelope a third time would let
        // this test drift onto a different document from its neighbours.
        let mut doc = bundle_doc(Some(
            serde_json::json!({"agent_context": {"function": "marketing"}}),
        ));
        doc["rules"] = serde_json::json!([
            {
                "rule_id": "OL-CMD-002",
                "kind": "command",
                "action": "deny",
                "match_pattern": "*psql*",
                "conditions": [
                    {"field": "agent.owner", "op": "in", "value": ["marketing"]}
                ],
                "mode": "enforce",
                "severity": "high",
                "reason": "authored against a wider condition vocabulary"
            },
            {
                "rule_id": "OL-CMD-003",
                "kind": "command",
                "action": "deny",
                "match_pattern": "*curl*",
                "conditions": [
                    {"field": "agent.function", "op": "not_in", "value": ["marketing"]}
                ],
                "mode": "enforce",
                "severity": "high",
                "reason": "authored against a wider op vocabulary"
            },
            {
                "rule_id": "OL-CMD-004",
                "kind": "command",
                "action": "deny",
                "match_pattern": "*ssh*",
                "conditions": [
                    {"field": "agent.function", "op": "in", "value": ["astrology"]}
                ],
                "mode": "enforce",
                "severity": "high",
                "reason": "authored against a wider function vocabulary"
            },
            {
                "rule_id": "OL-CMD-001",
                "kind": "command",
                "action": "deny",
                "match_pattern": "*rm -rf*",
                "conditions": [
                    {"field": "agent.function", "op": "in", "value": ["marketing"]}
                ],
                "mode": "enforce",
                "severity": "high",
                "reason": "OL-CMD-001 says no"
            }
        ]);

        let (resident, logs) = capture_logs(|| {
            let bundle = parse_bundle_tolerant(doc).expect("the bundle still parses");
            ResidentBundle::from_bundle(&bundle)
        });

        let ids: Vec<&str> = resident
            .command_rules
            .iter()
            .map(|r| r.rule_id.as_str())
            .collect();
        assert_eq!(ids, vec!["OL-CMD-001"], "only the readable rule survives");
        assert_eq!(resident.agent_function, Some(AgentFunction::Marketing));
        assert!(
            evaluate_command(&resident, "sudo rm -rf /tmp").is_some_and(|m| !m.shadow),
            "the surviving scoped deny still blocks on a member install"
        );
        assert!(logs.contains(ERR_RULE_SKIPPED), "{logs}");
        assert!(logs.contains("unrecognized_field"), "{logs}");
        assert!(logs.contains("OL-CMD-002"), "{logs}");
        assert!(logs.contains("OL-CMD-003"), "{logs}");
        assert!(logs.contains("OL-CMD-004"), "{logs}");
    }

    /// The projection contract in isolation: no conditions ⇒ `None`; one
    /// condition ⇒ its set, sorted and deduplicated; several ⇒ their
    /// intersection; disjoint ⇒ `Some(empty)`, kept rather than dropped.
    #[test]
    fn conditions_project_to_the_intersection_of_their_sets() {
        let projected = |conditions: Vec<PolicyRuleConditionsItem>| {
            let wire = scoped_wire_rule("OL-CMD-002", "*psql*", conditions);
            let resident = ResidentBundle::from_bundle(&wire_bundle(vec![wire], true));
            assert_eq!(resident.command_rules.len(), 1, "the rule loads");
            resident.command_rules[0].scoped_functions.clone()
        };

        assert_eq!(projected(vec![]), None);
        assert_eq!(
            projected(vec![function_in(&[
                AgentFunction::Sales,
                AgentFunction::Marketing,
                AgentFunction::Sales,
            ])]),
            Some(vec![AgentFunction::Sales, AgentFunction::Marketing]),
            "sorted in enum order and deduplicated"
        );
        assert_eq!(
            projected(vec![
                function_in(&[AgentFunction::Marketing, AgentFunction::Sales]),
                function_in(&[AgentFunction::Sales, AgentFunction::Finance]),
            ]),
            Some(vec![AgentFunction::Sales])
        );
        assert_eq!(
            projected(vec![
                function_in(&[AgentFunction::Marketing]),
                function_in(&[AgentFunction::Finance]),
            ]),
            Some(vec![]),
            "disjoint conditions keep the rule, scoped to nobody"
        );
    }

    /// The cross-repo guard for the new shape, in the document the platform
    /// composes: a scoped deny decides from the resident bundle for the named
    /// function and steps aside for any other.
    #[test]
    fn scoped_rule_from_platform_bundle_denies_only_the_named_function() {
        let load = |function: &str| {
            // Written out in full rather than assembled from the fixture
            // helpers above: a field the client stops reading has to show up
            // here as a diff.
            let doc = serde_json::json!({
                "schema_version": 1,
                "organization_id": "0192f8a1-4c3b-7e2a-9f10-5d8c3b1a7e42",
                "revision": 78,
                "built_at": "2026-08-16T08:00:00Z",
                "enforcement_enabled": true,
                "rules": [{
                    "rule_id": "OL-CMD-002",
                    "kind": "command",
                    "match_pattern": "*psql*",
                    "conditions": [
                        {"field": "agent.function", "op": "in", "value": ["marketing", "sales"]}
                    ],
                    "action": "deny",
                    "mode": "enforce",
                    "severity": "high",
                    "reason": "Direct database access is not part of a marketing or sales workflow"
                }],
                "signature": null,
                "client_config": {
                    "capture_identity_signals": false,
                    "agent_context": {"function": function}
                }
            });
            let wire = parse_bundle_tolerant(doc).expect("the platform bundle parses");
            ResidentBundle::from_bundle(&wire)
        };

        let marketing = load("marketing");
        assert_eq!(marketing.agent_function, Some(AgentFunction::Marketing));
        assert_eq!(
            marketing.command_rules[0].scoped_functions,
            Some(vec![AgentFunction::Sales, AgentFunction::Marketing])
        );
        assert!(evaluate_command(&marketing, "psql -h prod").is_some_and(|m| !m.shadow));

        let engineering = load("engineering");
        assert_eq!(
            engineering.command_rules.len(),
            1,
            "the rule is held, not dropped"
        );
        assert!(evaluate_command(&engineering, "psql -h prod").is_none());
    }

    // -- the request plane --------------------------------------------------

    #[test]
    fn mixed_bundle_loads_both_planes() {
        let bundle = wire_bundle(
            vec![
                wire_rule(
                    "OL-CMD-001",
                    "*rm -rf*",
                    PolicyRuleMode::Enforce,
                    PolicyRuleSeverity::Critical,
                ),
                wire_request_rule("OL-REQ-001", "prefix_reorder", PolicyRuleMode::Observe),
                wire_request_rule("OL-REQ-002", "history_trim", PolicyRuleMode::Observe),
            ],
            true,
        );
        let resident = ResidentBundle::from_bundle(&bundle);

        assert_eq!(resident.command_rules.len(), 1);
        assert_eq!(resident.command_rules[0].rule_id, "OL-CMD-001");
        let request_ids: Vec<&str> = resident
            .request_rules
            .iter()
            .map(|r| r.rule_id.as_str())
            .collect();
        assert_eq!(request_ids, vec!["OL-REQ-001", "OL-REQ-002"]);
    }

    /// The deny must behave identically whether or not request rules ride
    /// alongside it — that is the whole deliverable boundary.
    #[test]
    fn command_plane_unchanged() {
        let deny = wire_rule(
            "OL-CMD-001",
            "*rm -rf*",
            PolicyRuleMode::Enforce,
            PolicyRuleSeverity::Critical,
        );

        let alone = ResidentBundle::from_bundle(&wire_bundle(vec![deny.clone()], true));
        let alongside = ResidentBundle::from_bundle(&wire_bundle(
            vec![
                deny,
                wire_request_rule("OL-REQ-001", "prompt_edit", PolicyRuleMode::Observe),
            ],
            true,
        ));

        assert_eq!(alone.command_rules, alongside.command_rules);
        assert_eq!(
            evaluate_command(&alone, "sudo rm -rf /tmp").map(|m| m.rule_id),
            evaluate_command(&alongside, "sudo rm -rf /tmp").map(|m| m.rule_id)
        );
        assert!(evaluate_command(&alongside, "sudo rm -rf /tmp").is_some_and(|m| !m.shadow));
    }

    /// The cross-repo guard. A command rule must survive projection when it
    /// arrives in the shape the platform actually serializes — and also if that
    /// serializer regresses into emitting explicit nulls for the request-plane
    /// keys, since a present-but-null key satisfies JSON Schema `required` and
    /// would otherwise trip the command branch's `not`.
    #[test]
    fn command_rule_from_platform_bundle_is_not_skipped() {
        let canonical = r#"{
            "schema_version": 1,
            "organization_id": "0192f8a1-4c3b-7e2a-9f10-5d8c3b1a7e42",
            "revision": 42,
            "built_at": "2026-07-21T09:00:00Z",
            "enforcement_enabled": true,
            "rules": [{
                "rule_id": "OL-CMD-001",
                "kind": "command",
                "match_pattern": "*rm -rf*",
                "action": "deny",
                "mode": "enforce",
                "severity": "critical",
                "reason": "Recursive delete of a root path"
            }],
            "signature": null
        }"#;
        let with_nulls = canonical.replace(
            r#""kind": "command","#,
            r#""kind": "command", "rule_version": null, "select": null, "params": null,"#,
        );

        for (label, body) in [("canonical", canonical), ("nulls", with_nulls.as_str())] {
            let wire: PolicyBundle =
                serde_json::from_str(body).unwrap_or_else(|e| panic!("{label} bundle parses: {e}"));
            let resident = ResidentBundle::from_bundle(&wire);
            assert_eq!(resident.command_rules.len(), 1, "{label}");
            let verdict = evaluate_command(&resident, "sudo rm -rf /tmp")
                .unwrap_or_else(|| panic!("{label} deny still matches"));
            assert_eq!(verdict.rule_id, "OL-CMD-001", "{label}");
            assert!(!verdict.shadow, "{label} deny still blocks");
        }
    }

    /// A bundle authored before the request plane existed must project exactly
    /// as it does today: every rule on the command plane, nothing on the other.
    #[test]
    fn old_shape_bundle_still_loads() {
        let body = r#"{
            "schema_version": 1,
            "organization_id": "0192f8a1-4c3b-7e2a-9f10-5d8c3b1a7e42",
            "revision": 7,
            "built_at": "2026-07-20T11:30:00Z",
            "enforcement_enabled": true,
            "rules": [
                {"rule_id": "OL-CMD-001", "kind": "command", "match_pattern": "*rm -rf*",
                 "action": "deny", "mode": "enforce", "severity": "critical", "reason": "no"},
                {"rule_id": "OL-CMD-002", "kind": "command", "match_pattern": "*curl*",
                 "action": "deny", "mode": "observe", "severity": "low", "reason": "watch"}
            ],
            "signature": null
        }"#;
        let wire: PolicyBundle = serde_json::from_str(body).expect("v1 bundle parses");
        let resident = ResidentBundle::from_bundle(&wire);

        assert_eq!(resident.command_rules.len(), 2);
        assert!(resident.request_rules.is_empty());
        assert_eq!(resident.revision, 7);
    }

    // -- skip postures ------------------------------------------------------

    /// Guards the helper the three assertions below depend on.
    ///
    /// `tracing` caches one `Interest` per callsite process-wide. Reaching a
    /// callsite for the first time from a thread with no subscriber used to
    /// cache it as `Interest::never()`, which silently disabled it for every
    /// other thread — including one inside [`capture_logs`]. This reproduces
    /// that interleaving directly: the capture's dispatcher already exists, and
    /// the callsite is first touched from a bare thread. Without
    /// [`keep_callsites_live`] the assertion below reads an empty string.
    #[test]
    fn a_callsite_first_reached_from_a_bare_thread_is_still_captured() {
        fn probe() {
            tracing::warn!(target: "policy", code = "OL-CAPTURE-PROBE", "probe");
        }

        let (_, logs) = capture_logs(|| {
            std::thread::spawn(probe)
                .join()
                .expect("probe thread joins");
            probe();
        });

        assert!(logs.contains("OL-CAPTURE-PROBE"), "{logs}");
    }

    /// One gate-violating rule costs only itself. The deny beside it still
    /// denies — the prime invariant, asserted rather than assumed.
    #[test]
    fn gate_violation_skips_one_rule() {
        let mut bad = wire_rule(
            "OL-CMD-002",
            "*curl*",
            PolicyRuleMode::Enforce,
            PolicyRuleSeverity::High,
        );
        // A command rule may carry nothing from the request plane.
        bad.select = Some(Default::default());

        let good = wire_rule(
            "OL-CMD-001",
            "*rm -rf*",
            PolicyRuleMode::Enforce,
            PolicyRuleSeverity::Critical,
        );

        let (resident, logs) =
            capture_logs(|| ResidentBundle::from_bundle(&wire_bundle(vec![bad, good], true)));

        let ids: Vec<&str> = resident
            .command_rules
            .iter()
            .map(|r| r.rule_id.as_str())
            .collect();
        assert_eq!(ids, vec!["OL-CMD-001"]);
        assert!(
            evaluate_command(&resident, "sudo rm -rf /tmp").is_some_and(|m| !m.shadow),
            "the surviving deny still blocks"
        );
        assert!(logs.contains(ERR_RULE_SKIPPED), "{logs}");
        assert!(logs.contains("gate_violation"), "{logs}");
        assert!(logs.contains("OL-CMD-002"), "{logs}");
    }

    #[test]
    fn prefix_reorder_with_min_messages_is_skipped() {
        let mut bad = wire_request_rule("OL-REQ-001", "prefix_reorder", PolicyRuleMode::Observe);
        bad.select
            .as_mut()
            .expect("fixture has a select")
            .min_messages = Some(4);

        let (resident, logs) = capture_logs(|| {
            ResidentBundle::from_bundle(&wire_bundle(
                vec![
                    bad,
                    wire_request_rule("OL-REQ-002", "prefix_reorder", PolicyRuleMode::Observe),
                ],
                true,
            ))
        });

        let ids: Vec<&str> = resident
            .request_rules
            .iter()
            .map(|r| r.rule_id.as_str())
            .collect();
        assert_eq!(ids, vec!["OL-REQ-002"]);
        assert!(logs.contains("gate_violation"), "{logs}");
    }

    #[test]
    fn history_trim_with_marker_is_skipped() {
        let mut bad = wire_request_rule("OL-REQ-001", "history_trim", PolicyRuleMode::Observe);
        bad.params.as_mut().expect("fixture has params").marker = Some("x".to_string());

        let (resident, logs) =
            capture_logs(|| ResidentBundle::from_bundle(&wire_bundle(vec![bad], true)));

        assert!(resident.request_rules.is_empty());
        assert!(logs.contains("gate_violation"), "{logs}");
    }

    /// D-U4: a request rule with no `rule_version` cannot be tied back to the
    /// text that produced a would-have report, so it is not held at all.
    #[test]
    fn request_rule_without_rule_version_is_skipped() {
        let mut bad = wire_request_rule("OL-REQ-001", "prompt_edit", PolicyRuleMode::Observe);
        bad.rule_version = None;

        let (resident, logs) =
            capture_logs(|| ResidentBundle::from_bundle(&wire_bundle(vec![bad], true)));

        assert!(resident.request_rules.is_empty());
        assert!(logs.contains(ERR_RULE_SKIPPED), "{logs}");
        assert!(logs.contains("gate_violation"), "{logs}");
    }

    /// D-U9's client half. Every request rule is coerced to observe by
    /// construction in this phase, so without a line naming each one the
    /// coercion would be invisible to the operator who authored `enforce`.
    #[test]
    fn authored_enforce_request_rule_logs_coercion() {
        let bundle = wire_bundle(
            vec![
                wire_request_rule("OL-REQ-001", "prefix_reorder", PolicyRuleMode::Enforce),
                wire_request_rule("OL-REQ-002", "history_trim", PolicyRuleMode::Observe),
            ],
            true,
        );

        let (resident, logs) = capture_logs(|| ResidentBundle::from_bundle(&bundle));

        assert_eq!(resident.request_rules.len(), 2, "neither rule is dropped");
        assert!(logs.contains("enforce_coerced"), "{logs}");
        assert!(logs.contains("OL-REQ-001"), "{logs}");
        assert!(
            !logs.contains("OL-REQ-002"),
            "an observe rule is not coerced: {logs}"
        );

        // The coercion is a property of the stored rule, not of the log line.
        // Asserting only the log would let a future consumer read `Enforce` and
        // act on a rule this phase promises is inert.
        assert!(
            resident
                .request_rules
                .iter()
                .all(|r| r.mode == PolicyRuleMode::Observe),
            "every resident request rule is held as observe"
        );
    }

    /// Per-rule validation must stay proportional to the rule count. The bound
    /// is loose on purpose — it is not a benchmark, it is a tripwire for the
    /// one regression that would matter: recompiling the validator per rule
    /// instead of once per process.
    #[test]
    fn a_large_bundle_loads_without_per_rule_compilation() {
        let rules: Vec<PolicyRule> = (0..500)
            .map(|i| {
                wire_rule(
                    &format!("OL-CMD-{i:03}"),
                    &format!("*pattern-{i}*"),
                    PolicyRuleMode::Enforce,
                    PolicyRuleSeverity::High,
                )
            })
            .collect();

        let started = std::time::Instant::now();
        let resident = ResidentBundle::from_bundle(&wire_bundle(rules, true));
        let elapsed = started.elapsed();

        assert_eq!(resident.command_rules.len(), 500);
        assert!(
            elapsed < Duration::from_secs(5),
            "500 rules took {elapsed:?}"
        );
    }

    /// The belt behind the gate. If the embedded subschema ever fails to
    /// compile, `validate_rule` accepts everything rather than disarm the fleet
    /// — and this check is then the only thing stopping a future command-plane
    /// action from being carried out as a `deny` nobody authored.
    #[test]
    fn unknown_command_action_is_skipped_by_projection() {
        let mut rule = wire_rule(
            "OL-CMD-002",
            "*sudo*",
            PolicyRuleMode::Enforce,
            PolicyRuleSeverity::High,
        );
        rule.action = "require_approval".to_string();

        assert!(project_rule(&rule).is_none());
    }
}