car-verify 0.50.0

//! Static plan verification for Agent IR.
//!
//! Deterministic graph and dataflow algorithms — no solver, no search, no proof
//! term. The checks differ in strength, and conflating them is how a green
//! verdict gets over-trusted:
//!
//! - **Decision procedures** over their fragment: STRIPS applicability
//!   ([`plan_check`]), workflow precedence ([`workflow_graph`]), and
//!   Denning-style lattice information flow ([`infoflow`]).
//! - **Deliberate heuristics**: loop detection fires on three identical calls,
//!   so it has both false positives (a legitimate 3× poll) and false negatives
//!   (semantically redundant calls with differing arguments).
//! - **Sampling**: [`equivalent`] probes the supplied test states — two trivial
//!   defaults if you pass none — and [`montecarlo::simulate_monte_carlo`]
//!   samples rollouts. Neither decides anything.
//!
//! Not all of it is static, either: [`trace_policy`] is runtime verification
//! over an execution trace (bounded LTL), and [`cwm`] scores recorded
//! trajectories with a model in the repair loop.
//!
//! **Known blind spots.** The forward walk applies only what each action
//! *declares* in `expected_effects`. So it is optimistic in one direction — a
//! declared effect is assumed to land, though the tool may fail at runtime — and
//! pessimistic in the other: a precondition or `state_dependency` reading a key
//! that an upstream tool really writes but never declared is reported as
//! unavailable, at `error` severity. [`StaticState::unknown_keys`] exists to
//! model "a tool wrote this, value unknown", but **nothing in the workspace ever
//! populates it**, so `is_unknown()` is always false and provides no relief.
//! Undeclared effects are the practical false-rejection source here.
//!
//! Write conflicts are reported as warnings, not errors: `valid` stays `true`.
//! And `dependency_edges` tracks writers last-writer-wins, so with two writers of
//! one key an intervening reader can be scheduled into the same execution level
//! as its writer — which the executor runs concurrently — while this crate
//! reports only a warning.
//!
//! [`VerificationEvidence`] on every result names what each check did and did
//! not establish. Read it rather than trusting `valid` alone. The taxonomy
//! above is also carried *in the data*: every [`VerifyIssue`] and
//! [`CheckRecord`] tags itself with an [`EvidenceTier`], so a caller can tell a
//! decision procedure's finding from a heuristic's without knowing which
//! function produced it. Single-tier modules expose the same thing as an
//! `evidence_tier()` on their report type.
//!
//! Given a state S and proposal P, you can:
//! 1. **verify**: Check P is satisfiable in S without executing
//! 2. **simulate**: Compute expected final state S' without tools
//! 3. **simulate_monte_carlo**: Sample N rollouts with tools allowed to fail,
//!    giving P(goal), a distribution over S', and per-action blast radius —
//!    see [`montecarlo`]
//! 4. **equivalent**: Sample whether two proposals produce identical state
//! 5. **optimize**: Reorder actions safely, per the checks above

use car_ir::precondition::{self, StateView};
use car_ir::{build_dag, Action, ActionProposal, ActionType, ToolSchema};
use serde_json::Value;
use std::collections::{HashMap, HashSet};

pub mod attempt;
pub mod concurrency;
pub mod cwm;
pub mod dag;
pub mod goal;
pub mod infoflow;
pub mod intent;
pub mod montecarlo;
pub mod plan_check;
pub mod trace_policy;
pub mod transaction;
pub mod verifier;
pub use attempt::{Attempt, AttemptAdvice, AttemptLedger, AttemptOutcome, Exclusion, FailureClass};
pub use goal::{
    anchor_directive, evaluate_goal, governor_check, run_goal_loop, GoalCondition, GoalGovernor,
    GoalHalt, GoalInputs, GoalRun, GoalRunState, GoalSpec, GoalStatus, GoalVerdict,
    IterationOutcome,
};
pub use intent::{
    check_intent, gate_intent, intent_actions_from, IntentAction, IntentDisposition,
    IntentGateDecision, IntentGatePolicy, IntentReport, IntentSpec, IntentViolation,
    IntentViolationKind,
};
pub use montecarlo::{
    simulate_monte_carlo, ActionOutcome, Distribution, KeyOutcome, MonteCarloConfig,
    MonteCarloResult, ValueFrequency,
};
pub use plan_check::{
    check_plan, PlanCheckReport, PlanCheckRequest, PlanDefect, PlanDefectKind, PlanStep,
};
pub use verifier::{
    admit, required_classes, AdmissionDecision, AdmissionOutcome, EvidenceRequirement, UnmetReason,
    UnmetRequirement, VerifierAuthority, VerifierCost, VerifierDescriptor, VerifierOutcome,
    VerifierVerdict,
};
pub mod workflow_graph;
pub use concurrency::{
    analyze as analyze_concurrency, gate_concurrency, AgentOp, AnomalyFinding, ConcurrencyAnomaly,
    ConcurrencyGate, ConcurrencyGatePolicy, ConcurrencyReport, ConsistencyLevel, Disposition,
    GatedRemediation, Remediation,
};
pub use cwm::{
    score, score_predictions, simulate_with_model, synthesize_cwm, CwmRequest, CwmResult,
    EffectModel, Failure, GatedEffectModel, GatedPrediction, ScoreReport, Transition,
};
pub use infoflow::{
    check_information_flow, gate_flow, Confidentiality, FlowAction, FlowGateDecision,
    FlowGatePolicy, FlowPolicy, FlowReport, FlowViolation, FlowViolationKind, ToolLabels,
    TrustLevel,
};
pub use transaction::{
    check_transaction, check_transaction_with_predictions, ConflictKind, TransactionConflict,
    TransactionReport,
};
pub use workflow_graph::{
    check_temporal_policies, verify_workflow_graph, PolicyReport, PolicyViolation, TemporalPolicy,
    WorkflowDefect, WorkflowDefectKind, WorkflowEdge, WorkflowGraph, WorkflowVerifyReport,
};

/// Symbolic state for static analysis.
#[derive(Debug, Clone)]
pub struct StaticState {
    pub known: HashMap<String, Value>,
    pub unknown_keys: HashSet<String>,
}

impl StaticState {
    pub fn new() -> Self {
        Self {
            known: HashMap::new(),
            unknown_keys: HashSet::new(),
        }
    }

    pub fn from_map(map: HashMap<String, Value>) -> Self {
        Self {
            known: map,
            unknown_keys: HashSet::new(),
        }
    }

    pub fn get(&self, key: &str) -> Option<&Value> {
        self.known.get(key)
    }

    pub fn exists(&self, key: &str) -> bool {
        self.known.contains_key(key)
    }

    pub fn is_unknown(&self, key: &str) -> bool {
        self.unknown_keys.contains(key)
    }

    pub fn set(&mut self, key: &str, value: Value) {
        self.known.insert(key.to_string(), value);
        self.unknown_keys.remove(key);
    }
}

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

impl StateView for StaticState {
    fn get_value(&self, key: &str) -> Option<Value> {
        self.known.get(key).cloned()
    }
    fn key_exists(&self, key: &str) -> bool {
        self.known.contains_key(key)
    }
    fn is_unknown(&self, key: &str) -> bool {
        self.unknown_keys.contains(key)
    }
}

/// What *kind* of check produced a finding — the crate's existing taxonomy
/// (decision procedures / heuristics / sampling), carried in the data instead
/// of only in the module docs.
///
/// The gap this closes is narrow and worth stating exactly: the taxonomy at the
/// top of this file has always been accurate, but a caller holding a
/// [`VerifyIssue`] could not tell which branch of it produced that issue
/// without recognising the message string. Two findings that read identically
/// in a log — one from an exact set-membership test, one from a `count >= 3`
/// rule of thumb — now differ in the data.
///
/// # What this axis is not
///
/// This tier classifies a check's relationship to **the property it reports**,
/// over **the inputs it was handed**. It is deliberately orthogonal to a second
/// question: whether those inputs describe what will actually happen at
/// runtime. That question is answered elsewhere — [`CheckRecord::cannot_verify`],
/// [`VerificationEvidence::assumptions`], [`VerificationEvidence::untested_regions`],
/// and the "Known blind spots" section of the module docs. Folding the two into
/// one ordering would be the same mistake as grading "who may authorize this"
/// on the same ladder as "can this be undone": correlated, distinct, and
/// misleading once collapsed.
///
/// So [`EvidenceTier::DecisionProcedure`] is **not** a proof, a soundness
/// claim, or a prediction that the plan will work. This crate depends on
/// `car-ir` and serde; there is no solver in it and nothing here proves
/// anything. The forward model applies only what an action *declares*, so an
/// exactly-decided finding can still be about a world the tools then
/// contradict. The tier's entire job is to stop three unlike kinds of check
/// from reading alike.
///
/// # No ordering, on purpose
///
/// This enum deliberately derives neither `PartialOrd` nor `Ord`. The three
/// variants are *kinds*, not grades: `Heuristic` and `Sampled` have no
/// defensible strength ranking against each other — a proxy signal over
/// complete inputs and an exact measurement over incomplete inputs fail in
/// different directions, and which is worse depends entirely on the question
/// being asked. An `Ord` derive would encode declaration order as if it meant
/// something, and would invite exactly the filter [`VerifyResult::issues_with_tier`]
/// warns against (`tier >= EvidenceTier::Heuristic`, i.e. "discard the
/// findings I trust least"), which discards the crate's only signal for the
/// things no decision procedure here covers. Compare by equality; if you need
/// per-tier handling, match exhaustively.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EvidenceTier {
    /// The check decides the property it reports, exactly, within a declared
    /// fragment: total, deterministic, and free of false positives and false
    /// negatives *with respect to its inputs*. Set membership, graph
    /// reachability, and the STRIPS-style forward walk are all of this kind.
    ///
    /// The fragment is the point. "This precondition is unsatisfied in the
    /// forward model" is decided; "this precondition will fail at runtime" is
    /// not, and the accompanying `cannot_verify` says which one you are
    /// holding.
    DecisionProcedure,
    /// The check reports a property it does **not** decide, using a proxy
    /// signal chosen because it is useful in practice. False positives and
    /// false negatives are expected *on the check's own inputs*, not merely on
    /// the gap between the model and runtime.
    ///
    /// Loop detection is the crate's example: the repeat count is exact, but
    /// the step from "three identical calls" to "this is a runaway loop" is
    /// the proxy — a legitimate 3× poll trips it, and three semantically
    /// redundant calls with differing arguments slip past.
    Heuristic,
    /// The check examines a subset of a space and reports what it found there.
    /// A finding is a witness from the sample; the *absence* of a finding is
    /// evidence about the sample only, and generalises no further.
    ///
    /// [`equivalent`] probes the supplied test states (two trivial defaults if
    /// you pass none), [`montecarlo::simulate_monte_carlo`] samples rollouts,
    /// and [`cwm::score`] measures accuracy over the transitions it was given.
    Sampled,
}

impl EvidenceTier {
    /// Stable lowercase label, matching the serde representation. Handy for
    /// log lines and for the FFI/JSON-RPC surfaces, which carry the tier as a
    /// string so they need no dependency on this crate.
    ///
    /// Matched exhaustively on purpose (project convention #2): a new tier
    /// must fail to compile here rather than silently acquire a label.
    pub const fn as_str(&self) -> &'static str {
        match self {
            EvidenceTier::DecisionProcedure => "decision_procedure",
            EvidenceTier::Heuristic => "heuristic",
            EvidenceTier::Sampled => "sampled",
        }
    }
}

/// A single verification finding.
#[derive(Debug, Clone, serde::Serialize)]
#[non_exhaustive]
pub struct VerifyIssue {
    pub action_id: String,
    pub severity: String, // "error", "warning", "info"
    pub message: String,
    /// Which kind of check produced this finding — see [`EvidenceTier`].
    ///
    /// Orthogonal to `severity`: severity says how bad the situation would be
    /// if the finding is right, the tier says how the finding was arrived at.
    /// An `error` from a heuristic and an `error` from a decision procedure are
    /// equally loud and not equally trustworthy.
    ///
    /// It is also *not* the same axis as "does this block execution".
    /// `car_engine::is_blocking_issue` blocks on state-independence, and two of
    /// the three findings it treats as advisory (`precondition will fail`,
    /// `not available at this point`) are `DecisionProcedure` findings —
    /// exactly decided over a forward model that only sees declared effects.
    /// Do not rewire that gate onto this field.
    pub tier: EvidenceTier,
}

/// Scope record for one verification check.
///
/// Survey "Code as Agent Harness" §5.2.2 argues a green check creates a
/// false sense of correctness unless the verifier declares *what it
/// verifies, what it cannot verify, and what confidence it provides*. A
/// `CheckRecord` makes that scope explicit per check so downstream
/// consumers (self-repair, harness evolution, human review) can reason
/// about *why* a proposal is `valid`, not merely that it is.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[non_exhaustive]
pub struct CheckRecord {
    /// Stable identifier, e.g. `"preconditions"`, `"tool_existence"`.
    pub name: String,
    /// Whether this check actually ran. Some checks are conditional —
    /// parameter-schema validation only runs when tool schemas are
    /// supplied; when skipped, `ran=false` and `cannot_verify` names the
    /// resulting blind spot.
    pub ran: bool,
    /// What a pass of this check establishes.
    pub verifies: String,
    /// The scope boundary: what a pass does *not* establish. The core
    /// anti-overconfidence signal.
    pub cannot_verify: String,
    /// Number of issues this check contributed to `issues`.
    pub findings: usize,
    /// What kind of check this is — see [`EvidenceTier`]. Every
    /// [`VerifyIssue`] this check contributed carries the same tier, so a
    /// consumer can read the strength of a whole check without walking its
    /// findings.
    pub tier: EvidenceTier,
}

/// Evidence bundle accompanying a verification result.
///
/// Makes the verifier's scope inspectable so a `valid` verdict is not
/// mistaken for a full-specification guarantee (survey §5.2.2: "every
/// accepted action [should] carry an evidence bundle containing the
/// checks run, the assumptions preserved, the untested regions, and the
/// remaining risks"). Static verification is sound only within its
/// declared scope; this bundle is that declaration.
#[derive(Debug, Clone, serde::Serialize)]
pub struct VerificationEvidence {
    /// Per-check scope records.
    pub checks: Vec<CheckRecord>,
    /// Assumptions the verdict relies on (e.g. registered tools behave
    /// per their schema; supplied state values are accurate).
    pub assumptions: Vec<String>,
    /// State keys / aspects static verification could not evaluate —
    /// unknown or dynamic keys, and runtime-only tool outputs.
    pub untested_regions: Vec<String>,
    /// Risks that persist even when `valid` is true — downgraded
    /// warnings, undeclared write conflicts, dynamically-resolved
    /// preconditions.
    pub residual_risks: Vec<String>,
    /// Heuristic 0.0–1.0 coverage confidence: how completely the
    /// applicable checks covered this proposal. 1.0 means every
    /// applicable check ran against fully-known state with no warnings;
    /// reduced by skipped checks, unknown/dynamic state, and warnings.
    /// This is a coverage signal, not a probability of success.
    pub confidence: f64,
}

/// Complete verification result.
#[derive(Debug, serde::Serialize)]
pub struct VerifyResult {
    pub valid: bool,
    pub issues: Vec<VerifyIssue>,
    pub simulated_state: HashMap<String, Value>,
    pub execution_levels: Vec<Vec<String>>,
    pub conflicts: Vec<(String, String, String)>, // (action1, action2, key)
    /// Inspectable scope of this verdict (survey §5.2.2). See
    /// [`VerificationEvidence`].
    pub evidence: VerificationEvidence,
}

impl VerifyResult {
    pub fn errors(&self) -> Vec<&VerifyIssue> {
        self.issues
            .iter()
            .filter(|i| i.severity == "error")
            .collect()
    }

    pub fn warnings(&self) -> Vec<&VerifyIssue> {
        self.issues
            .iter()
            .filter(|i| i.severity == "warning")
            .collect()
    }

    /// Findings produced by one kind of check — see [`EvidenceTier`].
    ///
    /// The intended use is triage, not filtering for correctness: "show me the
    /// heuristic findings separately so a reviewer can eyeball them" is a good
    /// reason to call this; "drop everything that isn't a decision procedure"
    /// is not, since a heuristic finding is the crate's only signal for the
    /// things no decision procedure here covers.
    pub fn issues_with_tier(&self, tier: EvidenceTier) -> Vec<&VerifyIssue> {
        self.issues.iter().filter(|i| i.tier == tier).collect()
    }
}

// --- Action effects (symbolic) ---

pub(crate) fn apply_action_effects(action: &Action, state: &mut StaticState) {
    if action.action_type == ActionType::StateWrite {
        if let Some(key) = action.parameters.get("key").and_then(|v| v.as_str()) {
            let value = action
                .parameters
                .get("value")
                .cloned()
                .unwrap_or(Value::Null);
            state.set(key, value);
        }
    }
    for (key, value) in &action.expected_effects {
        state.set(key, value.clone());
    }
}

// --- Conflict detection ---

fn detect_conflicts(actions: &[Action]) -> Vec<(String, String, String)> {
    let mut writers: HashMap<String, Vec<String>> = HashMap::new();

    for action in actions {
        let mut keys_written = HashSet::new();
        if action.action_type == ActionType::StateWrite {
            if let Some(k) = action.parameters.get("key").and_then(|v| v.as_str()) {
                keys_written.insert(k.to_string());
            }
        }
        for key in action.expected_effects.keys() {
            keys_written.insert(key.clone());
        }
        for key in keys_written {
            writers.entry(key).or_default().push(action.id.clone());
        }
    }

    let dep_map: HashMap<String, HashSet<String>> = actions
        .iter()
        .map(|a| (a.id.clone(), a.state_dependencies.iter().cloned().collect()))
        .collect();

    let mut conflicts = Vec::new();
    for (key, action_ids) in &writers {
        if action_ids.len() < 2 {
            continue;
        }
        for i in 0..action_ids.len() {
            for j in (i + 1)..action_ids.len() {
                let a1 = &action_ids[i];
                let a2 = &action_ids[j];
                let deps_a2 = dep_map.get(a2).cloned().unwrap_or_default();
                let deps_a1 = dep_map.get(a1).cloned().unwrap_or_default();
                if !deps_a2.contains(key) && !deps_a1.contains(key) {
                    conflicts.push((a1.clone(), a2.clone(), key.clone()));
                }
            }
        }
    }
    conflicts
}

// --- Tool-parameter schema validation ---

/// Friendly JSON type name for error messages.
fn json_type_name(v: &Value) -> &'static str {
    match v {
        Value::Null => "null",
        Value::Bool(_) => "boolean",
        Value::Number(_) => "number",
        Value::String(_) => "string",
        Value::Array(_) => "array",
        Value::Object(_) => "object",
    }
}

/// Does `v` satisfy a single JSON Schema `type` keyword?
fn value_matches_type(v: &Value, expected: &str) -> bool {
    match expected {
        "string" => v.is_string(),
        "number" => v.is_number(),
        // JSON Schema "integer": an integral number. Accept i64/u64,
        // plus a float with no fractional part (e.g. `5.0`).
        "integer" => {
            v.is_i64() || v.is_u64() || v.as_f64().map(|f| f.fract() == 0.0).unwrap_or(false)
        }
        "boolean" => v.is_boolean(),
        "array" => v.is_array(),
        "object" => v.is_object(),
        "null" => v.is_null(),
        // Unknown/unsupported type keyword: don't flag — we only
        // enforce the keywords we understand.
        _ => true,
    }
}

/// Validate a tool_call's `parameters` against the tool's JSON-Schema
/// `parameters` object. Intentionally a focused subset of JSON Schema
/// — the two checks that catch the overwhelming majority of malformed
/// model output: declared property `type`s and `required` presence.
/// Returns human-readable violation messages; empty when the schema
/// imposes no constraints (e.g. the default empty object `{}`).
fn validate_tool_params(params: &HashMap<String, Value>, schema: &Value) -> Vec<String> {
    let mut out = Vec::new();
    let Some(schema_obj) = schema.as_object() else {
        // Non-object schema: nothing we can enforce.
        return out;
    };

    // required: every named key must be present in params.
    if let Some(Value::Array(required)) = schema_obj.get("required") {
        for req in required {
            if let Some(name) = req.as_str() {
                if !params.contains_key(name) {
                    out.push(format!("missing required parameter '{name}'"));
                }
            }
        }
    }

    // property types: each supplied param whose key has a declared
    // `type` must match it. `type` may be a string or an array of
    // strings (JSON Schema union).
    if let Some(Value::Object(properties)) = schema_obj.get("properties") {
        for (key, val) in params {
            let Some(prop_schema) = properties.get(key).and_then(|s| s.as_object()) else {
                continue;
            };
            let ok = match prop_schema.get("type") {
                Some(Value::String(t)) => value_matches_type(val, t),
                Some(Value::Array(types)) => types
                    .iter()
                    .filter_map(|t| t.as_str())
                    .any(|t| value_matches_type(val, t)),
                // No declared type (or non-string/array): accept.
                _ => true,
            };
            if !ok {
                let expected = match prop_schema.get("type") {
                    Some(Value::String(t)) => t.clone(),
                    Some(Value::Array(types)) => types
                        .iter()
                        .filter_map(|t| t.as_str())
                        .collect::<Vec<_>>()
                        .join("|"),
                    _ => String::new(),
                };
                out.push(format!(
                    "parameter '{key}' has wrong type: expected {expected}, got {}",
                    json_type_name(val)
                ));
            }
        }
    }

    out
}

// --- Core verification ---

/// Statically verify a proposal against an initial state.
///
/// `registered_tools` carries tool *names* only, so tool-existence is
/// checked but `parameters` are not. To additionally validate each
/// `tool_call`'s parameters against the tool's registered JSON Schema
/// (type mismatches, missing required fields), use
/// [`verify_with_schemas`].
pub fn verify(
    proposal: &ActionProposal,
    initial_state: Option<&HashMap<String, Value>>,
    registered_tools: Option<&HashSet<String>>,
    max_actions: usize,
) -> VerifyResult {
    verify_inner(proposal, initial_state, registered_tools, None, max_actions)
}

/// Like [`verify`], but validates each `tool_call`'s `parameters`
/// against the registered [`ToolSchema`]'s `parameters` JSON Schema —
/// catching type mismatches (`{"path": 42}` for a `string` param) and
/// missing `required` fields before dispatch. Tool existence is
/// checked against the schema map's keys. This is the path the runtime
/// (`verify_proposal`) and daemon (`verify` JSON-RPC) use, where the
/// full schemas registered via `register_tool_schema` are available.
pub fn verify_with_schemas(
    proposal: &ActionProposal,
    initial_state: Option<&HashMap<String, Value>>,
    tool_schemas: Option<&HashMap<String, ToolSchema>>,
    max_actions: usize,
) -> VerifyResult {
    verify_inner(proposal, initial_state, None, tool_schemas, max_actions)
}

/// How the topological walk treats the effects of an action it has just found
/// a problem with.
///
/// The two callers want opposite things, and conflating them was
/// Parslee-ai/car#622.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum EffectMode {
    /// Apply `expected_effects` even when the action's preconditions fail or
    /// its state dependencies are missing.
    ///
    /// This is what [`verify`] wants. Its job is to report **every** problem in
    /// one pass, so it keeps walking as though each action had run. Withholding
    /// effects here would bury the real findings under a cascade of
    /// "dependency not available" issues that are artifacts of the first
    /// failure rather than independent defects.
    Optimistic,
    /// Skip the effects of an action that could not run.
    ///
    /// This is what [`simulate`] wants, because the executor rejects such an
    /// action *before* dispatch (`ActionStatus::Rejected`) and its effects
    /// never land. Downstream actions then find their dependencies missing and
    /// are skipped in turn, so the cascade emerges from the data dependencies —
    /// the same way the executor produces it — without modelling
    /// `failure_behavior` here.
    ExecutionFaithful,
}

fn verify_inner(
    proposal: &ActionProposal,
    initial_state: Option<&HashMap<String, Value>>,
    registered_tools: Option<&HashSet<String>>,
    tool_schemas: Option<&HashMap<String, ToolSchema>>,
    max_actions: usize,
) -> VerifyResult {
    verify_inner_with_effects(
        proposal,
        initial_state,
        registered_tools,
        tool_schemas,
        max_actions,
        EffectMode::Optimistic,
    )
}

fn verify_inner_with_effects(
    proposal: &ActionProposal,
    initial_state: Option<&HashMap<String, Value>>,
    registered_tools: Option<&HashSet<String>>,
    tool_schemas: Option<&HashMap<String, ToolSchema>>,
    max_actions: usize,
    effect_mode: EffectMode,
) -> VerifyResult {
    let mut state = match initial_state {
        Some(s) => StaticState::from_map(s.clone()),
        None => StaticState::new(),
    };
    let mut issues = Vec::new();

    // Per-check finding counters for the evidence bundle (§5.2.2). The
    // topo-walk checks below are interleaved per action, so they are
    // tallied inline rather than by issue-vector deltas.
    let mut precondition_findings = 0usize;
    let mut state_dependency_findings = 0usize;
    let mut tool_existence_findings = 0usize;
    let mut param_schema_findings = 0usize;
    let mut has_tool_calls = false;
    // A malformed `tool_call` with no tool named is a structural finding
    // the existence pass produces even without a registry — track it so
    // the check's `ran` flag and `findings` count can't contradict
    // (neo review m1).
    let mut saw_missing_tool = false;
    // Compensation resolution: a declared undo that names a missing tool or a
    // sibling action that isn't in the batch. Counted separately from
    // `tool_existence_findings` so the evidence bundle says which check fired.
    let mut compensation_findings = 0usize;
    // An `ActionRef` compensation is resolvable with no registry at all — the
    // referent is in the proposal — so the check can run even when existence
    // could not. Tracked so `ran` and `findings` cannot contradict.
    let mut saw_compensation_ref = false;
    // Which conditional checks actually ran, given the inputs we were
    // handed. Existence needs *some* tool registry; parameter-schema
    // validation needs the full schemas.
    let has_tool_registry = tool_schemas.is_some() || registered_tools.is_some();
    let param_schema_ran = tool_schemas.is_some();

    // Resource bounds
    let issues_before_bounds = issues.len();
    if proposal.actions.len() > max_actions {
        issues.push(VerifyIssue {
            action_id: proposal
                .actions
                .first()
                .map(|a| a.id.clone())
                .unwrap_or_default(),
            severity: "warning".to_string(),
            message: format!(
                "excessive actions: {} (limit {})",
                proposal.actions.len(),
                max_actions
            ),
            // `len > max_actions` is decided, not estimated. The *limit* is a
            // policy input supplied by the caller — choosing it well is a
            // judgement call, but the tier grades the check against its own
            // claim ("this plan exceeds the limit you gave me"), and that claim
            // is exact.
            tier: EvidenceTier::DecisionProcedure,
        });
    }

    let resource_bound_findings = issues.len() - issues_before_bounds;

    // Loop detection
    let issues_before_loop = issues.len();
    let mut seen_calls: HashMap<String, u32> = HashMap::new();
    for action in &proposal.actions {
        if action.action_type == ActionType::ToolCall {
            if let Some(ref tool) = action.tool {
                let params = serde_json::to_string(&action.parameters).unwrap_or_default();
                let key = format!("{}:{}", tool, params);
                *seen_calls.entry(key).or_insert(0) += 1;
            }
        }
    }
    for (call_key, count) in &seen_calls {
        let tool_name = call_key.split(':').next().unwrap_or("?");
        if *count >= 3 {
            issues.push(VerifyIssue {
                action_id: "proposal".to_string(),
                severity: "error".to_string(),
                message: format!(
                    "repeated identical tool call: {} ({}x) — likely loop",
                    tool_name, count
                ),
                // The count is exact; "likely loop" is not. Three legitimate
                // polls of the same endpoint produce this finding, and three
                // semantically redundant calls with differing arguments do not.
                tier: EvidenceTier::Heuristic,
            });
        } else if *count == 2 {
            issues.push(VerifyIssue {
                action_id: "proposal".to_string(),
                severity: "warning".to_string(),
                message: format!("duplicate tool call: {} ({}x)", tool_name, count),
                // Same proxy, one threshold lower: a duplicate call is
                // reported as suspicious, but a retry is a duplicate call.
                tier: EvidenceTier::Heuristic,
            });
        }
    }

    let loop_detection_findings = issues.len() - issues_before_loop;

    // Build DAG
    let levels = build_dag(&proposal.actions);
    let execution_levels: Vec<Vec<String>> = levels
        .iter()
        .map(|level| {
            level
                .iter()
                .map(|&i| proposal.actions[i].id.clone())
                .collect()
        })
        .collect();

    // Walk in topological order
    for level in &levels {
        for &idx in level {
            let action = &proposal.actions[idx];

            // Would the executor refuse to dispatch this action? A failing
            // precondition or a missing state dependency both produce
            // `ActionStatus::Rejected` *before* the tool runs, so under
            // `ExecutionFaithful` its effects must not land (car#622).
            let mut blocked = false;

            // Check preconditions
            for pre in &action.preconditions {
                if let Some(error) = precondition::check_precondition(pre, &state) {
                    precondition_findings += 1;
                    blocked = true;
                    issues.push(VerifyIssue {
                        action_id: action.id.clone(),
                        severity: "error".to_string(),
                        message: format!("precondition will fail: {}", error),
                        // `check_precondition` decides the predicate against
                        // the forward-simulated state — no guessing. That the
                        // forward state is built from *declared* effects, and
                        // so can disagree with runtime, is the separate
                        // fidelity axis: see `cannot_verify` on the
                        // "preconditions" CheckRecord and the crate's known
                        // blind spots.
                        tier: EvidenceTier::DecisionProcedure,
                    });
                }
            }

            // State dependencies
            for dep in &action.state_dependencies {
                if !state.exists(dep) && !state.is_unknown(dep) {
                    state_dependency_findings += 1;
                    blocked = true;
                    issues.push(VerifyIssue {
                        action_id: action.id.clone(),
                        severity: "error".to_string(),
                        message: format!("state dependency '{}' not available at this point", dep),
                        // Membership in the forward model's key set — decided.
                        // Undeclared writes are why the *model* can be wrong
                        // here, which is the fidelity axis, not this one.
                        tier: EvidenceTier::DecisionProcedure,
                    });
                }
            }

            // Tool existence + parameter-schema validation
            if action.action_type == ActionType::ToolCall {
                has_tool_calls = true;
                if let Some(ref tool) = action.tool {
                    // Existence: prefer the schema map's keys, fall
                    // back to the name set. When neither is provided
                    // (both None) existence isn't checked.
                    let registered = match (tool_schemas, registered_tools) {
                        (Some(schemas), _) => Some(schemas.contains_key(tool.as_str())),
                        (None, Some(names)) => Some(names.contains(tool.as_str())),
                        (None, None) => None,
                    };
                    if registered == Some(false) {
                        tool_existence_findings += 1;
                        issues.push(VerifyIssue {
                            action_id: action.id.clone(),
                            severity: "error".to_string(),
                            message: format!("tool '{}' is not registered", tool),
                            // Set membership in the supplied registry.
                            tier: EvidenceTier::DecisionProcedure,
                        });
                    }
                    // Parameters: validate against the registered
                    // schema when we have one. This is the check the
                    // `register_tool_schema` contract promises —
                    // type mismatches and missing required fields.
                    if let Some(schema) = tool_schemas.and_then(|s| s.get(tool.as_str())) {
                        for msg in validate_tool_params(&action.parameters, &schema.parameters) {
                            param_schema_findings += 1;
                            issues.push(VerifyIssue {
                                action_id: action.id.clone(),
                                severity: "error".to_string(),
                                message: format!("tool '{tool}': {msg}"),
                                // `validate_tool_params` implements a strict
                                // subset of JSON Schema (`required` +
                                // `type`) and decides that subset exactly —
                                // incomplete, but never approximate. What
                                // falls outside the subset is recorded in the
                                // check's `cannot_verify`, not hidden behind a
                                // weaker tier.
                                tier: EvidenceTier::DecisionProcedure,
                            });
                        }
                    }
                } else {
                    saw_missing_tool = true;
                    tool_existence_findings += 1;
                    issues.push(VerifyIssue {
                        action_id: action.id.clone(),
                        severity: "error".to_string(),
                        message: "tool_call action has no tool specified".to_string(),
                        // Structural: the field is absent or it isn't.
                        tier: EvidenceTier::DecisionProcedure,
                    });
                }
            }

            // Compensation resolution. A `Compensable` action's declared undo
            // is the whole basis for calling the effect recoverable, so a
            // compensation naming a tool that does not exist or an action that
            // is not in the batch is a rollback plan that cannot run — and the
            // moment anyone discovers it is the moment it is worth least.
            match &action.compensation {
                Some(car_ir::Compensation::Tool { tool, .. }) => {
                    let registered = match (tool_schemas, registered_tools) {
                        (Some(schemas), _) => Some(schemas.contains_key(tool.as_str())),
                        (None, Some(names)) => Some(names.contains(tool.as_str())),
                        (None, None) => None,
                    };
                    if registered == Some(false) {
                        compensation_findings += 1;
                        issues.push(VerifyIssue {
                            action_id: action.id.clone(),
                            severity: "error".to_string(),
                            message: format!(
                                "compensation names tool '{tool}', which is not registered"
                            ),
                            tier: EvidenceTier::DecisionProcedure,
                        });
                    }
                }
                Some(car_ir::Compensation::ActionRef { action_id }) => {
                    saw_compensation_ref = true;
                    if !proposal.actions.iter().any(|a| &a.id == action_id) {
                        compensation_findings += 1;
                        issues.push(VerifyIssue {
                            action_id: action.id.clone(),
                            severity: "error".to_string(),
                            message: format!(
                                "compensation references action '{action_id}', which is not in this proposal"
                            ),
                            tier: EvidenceTier::DecisionProcedure,
                        });
                    }
                }
                None => {}
            }

            // A `Compensable` contract with nothing declared to compensate
            // with. `Action::missing_required_compensation` owns the rule; this
            // is the surface that reports it.
            if action.missing_required_compensation() {
                compensation_findings += 1;
                issues.push(VerifyIssue {
                    action_id: action.id.clone(),
                    severity: "error".to_string(),
                    message: "action declares reversibility 'compensable' but no compensation"
                        .to_string(),
                    tier: EvidenceTier::DecisionProcedure,
                });
            }

            // `verify` applies effects regardless, so one early failure doesn't
            // bury the rest of the plan's real findings under a cascade of
            // knock-on "dependency not available" issues. `simulate` must not:
            // the executor rejects a blocked action before dispatch, so its
            // effects never land, and predicting otherwise is what made
            // `simulate` disagree with execution (car#622).
            if effect_mode == EffectMode::Optimistic || !blocked {
                apply_action_effects(action, &mut state);
            }
        }
    }

    // Conflicts
    let conflicts = detect_conflicts(&proposal.actions);
    for (a1, a2, key) in &conflicts {
        issues.push(VerifyIssue {
            action_id: a1.clone(),
            severity: "warning".to_string(),
            message: format!(
                "write conflict on '{}' with action {} (no dependency declared)",
                key, a2
            ),
            // `detect_conflicts` is exact over what the actions declare: two
            // writers of one key with no `state_dependencies` edge between
            // them. Whether the runtime interleaving actually hurts is a
            // different question, and the residual risk says so.
            tier: EvidenceTier::DecisionProcedure,
        });
    }

    let conflict_findings = conflicts.len();

    let has_errors = issues.iter().any(|i| i.severity == "error");
    let warning_count = issues.iter().filter(|i| i.severity == "warning").count();

    // --- Assemble the evidence bundle (§5.2.2) ---
    let checks = vec![
        CheckRecord {
            name: "resource_bounds".into(),
            ran: true,
            verifies: format!("action count is within the limit ({max_actions})"),
            cannot_verify: "per-action cost, wall-clock time, or memory at runtime".into(),
            findings: resource_bound_findings,
            tier: EvidenceTier::DecisionProcedure,
        },
        CheckRecord {
            name: "loop_detection".into(),
            ran: true,
            verifies: "no identical tool call is repeated enough to look like a loop".into(),
            cannot_verify: "semantically redundant calls with differing arguments".into(),
            findings: loop_detection_findings,
            // The only heuristic among `verify`'s checks — see the
            // `EvidenceTier::Heuristic` docs for why the repeat count doesn't
            // decide the property it reports.
            tier: EvidenceTier::Heuristic,
        },
        CheckRecord {
            name: "preconditions".into(),
            ran: true,
            verifies: "declared preconditions hold against the statically-known state".into(),
            cannot_verify: "preconditions over keys whose values are only known at runtime".into(),
            findings: precondition_findings,
            tier: EvidenceTier::DecisionProcedure,
        },
        CheckRecord {
            name: "state_dependencies".into(),
            ran: true,
            verifies: "each declared state dependency is produced before it is read".into(),
            cannot_verify: "undeclared reads — state a tool consumes without listing it".into(),
            findings: state_dependency_findings,
            tier: EvidenceTier::DecisionProcedure,
        },
        CheckRecord {
            // The existence pass "ran" if a registry let us check names,
            // or if it caught a structurally malformed tool_call (no tool
            // named) even without one — so `ran` and `findings` agree.
            name: "tool_existence".into(),
            ran: has_tool_registry || saw_missing_tool,
            verifies: if has_tool_registry {
                "every tool_call names a registered tool".into()
            } else if saw_missing_tool {
                "tool_call structural well-formedness (a tool is named); registry not supplied so existence unchecked".into()
            } else {
                "(skipped — no tool registry supplied)".into()
            },
            cannot_verify: "whether the registered tool behaves as its name/description implies"
                .into(),
            findings: tool_existence_findings,
            // Set membership plus a structural field test. The tier describes
            // the check, so it stays `DecisionProcedure` even when the check
            // was skipped for want of a registry — `ran: false` is how a skip
            // is reported, not a weaker tier.
            tier: EvidenceTier::DecisionProcedure,
        },
        CheckRecord {
            name: "param_schema".into(),
            ran: param_schema_ran,
            verifies: if param_schema_ran {
                "tool_call parameters match the registered JSON Schema (types + required)".into()
            } else {
                "(skipped — no tool schemas supplied; existence only)".into()
            },
            cannot_verify:
                "value-level constraints beyond type/required (ranges, formats, cross-field)".into(),
            findings: param_schema_findings,
            tier: EvidenceTier::DecisionProcedure,
        },
        CheckRecord {
            name: "compensation_resolution".into(),
            // Resolvable without a registry when the compensation is an
            // `ActionRef` (the referent is in the proposal), and the
            // declared-but-missing rule needs no inputs at all.
            ran: has_tool_registry || saw_compensation_ref || compensation_findings > 0,
            verifies: "a declared compensation names a registered tool or an action in this \
                       proposal, and a `compensable` action declares one at all"
                .into(),
            cannot_verify: "whether the named compensation actually undoes the effect — that it \
                            is the right inverse, and that it will still work later"
                .into(),
            findings: compensation_findings,
            // Set membership and an id lookup over the batch.
            tier: EvidenceTier::DecisionProcedure,
        },
        CheckRecord {
            name: "write_conflicts".into(),
            ran: true,
            verifies: "concurrent writers to the same key declare an ordering dependency".into(),
            cannot_verify:
                "semantic conflicts — two actions whose effects are logically incompatible".into(),
            findings: conflict_findings,
            tier: EvidenceTier::DecisionProcedure,
        },
    ];

    // Untested regions: values the static pass cannot pin down because
    // they are only determined at runtime. A tool_call's return value is
    // opaque to static analysis, and any state key the tool is declared
    // to write holds a runtime-determined value (the declared effect is a
    // placeholder, not the real value). We source these from the IR
    // directly rather than from `StaticState`, which only tracks
    // statically-known values (neo review M1).
    let mut untested_regions: Vec<String> = Vec::new();
    for action in &proposal.actions {
        if action.action_type == ActionType::ToolCall {
            if let Some(ref tool) = action.tool {
                untested_regions.push(format!(
                    "runtime output of tool '{tool}' (action {})",
                    action.id
                ));
            }
            for key in action.expected_effects.keys() {
                untested_regions.push(format!(
                    "state key '{key}' (value set at runtime by action {})",
                    action.id
                ));
            }
        }
    }
    untested_regions.sort();
    untested_regions.dedup();

    let mut assumptions = vec![
        "supplied initial-state values are accurate".to_string(),
        "tool implementations honor their declared effects and side effects".to_string(),
    ];
    if !param_schema_ran && has_tool_calls {
        assumptions.push(
            "tool_call parameters are well-formed (no schemas supplied to check them)".to_string(),
        );
    }

    let mut residual_risks = Vec::new();
    if !conflicts.is_empty() {
        residual_risks.push(format!(
            "{} undeclared write conflict(s) — last-writer-wins at runtime",
            conflicts.len()
        ));
    }
    if warning_count > 0 {
        residual_risks.push(format!(
            "{warning_count} warning(s) not blocking the verdict"
        ));
    }
    if !untested_regions.is_empty() {
        residual_risks.push(
            "outcomes depending on runtime tool output or runtime-set state are unverified"
                .to_string(),
        );
    }

    // Coverage confidence: start full, dock for skipped applicable
    // checks, unknown/dynamic state, and warnings. A coverage signal,
    // not a probability — documented on the field.
    let mut confidence: f64 = 1.0;
    if has_tool_calls && !has_tool_registry {
        confidence -= 0.15;
    }
    if has_tool_calls && !param_schema_ran {
        confidence -= 0.20;
    }
    confidence -= (untested_regions.len() as f64 * 0.02).min(0.25);
    confidence -= (warning_count as f64 * 0.05).min(0.20);
    let confidence = confidence.clamp(0.0, 1.0);

    let evidence = VerificationEvidence {
        checks,
        assumptions,
        untested_regions,
        residual_risks,
        confidence,
    };

    VerifyResult {
        valid: !has_errors,
        issues,
        simulated_state: state.known,
        execution_levels,
        conflicts,
        evidence,
    }
}

/// Simulate a proposal's state effects without executing tools.
///
/// Predicts the state the **executor** would leave behind: an action whose
/// preconditions fail, or whose state dependencies aren't available, is
/// rejected before dispatch and contributes no effects. Downstream actions then
/// find their own dependencies missing and drop out in turn, so the cascade
/// follows the data dependencies exactly as it does at runtime.
///
/// This deliberately differs from [`verify`], which keeps applying effects past
/// a failure so it can report every problem in one pass. Sharing that
/// optimism made `simulate` claim `deployed: true` for a deploy whose
/// `tests_passed` precondition provably could not hold (Parslee-ai/car#622).
///
/// Scope: models per-action gating, not `failure_behavior`. An *independent*
/// action alongside a blocked one still contributes its effects here, whereas
/// the executor's default `FailureBehavior::Abort` may stop the run before
/// reaching it. So this is the state assuming execution proceeds as far as the
/// dependency graph allows — never a claim that a provably-blocked action ran.
pub fn simulate(
    proposal: &ActionProposal,
    initial_state: Option<&HashMap<String, Value>>,
) -> HashMap<String, Value> {
    verify_inner_with_effects(
        proposal,
        initial_state,
        None,
        None,
        usize::MAX,
        EffectMode::ExecutionFaithful,
    )
    .simulated_state
}

/// Test if two proposals produce identical state transitions.
///
/// [`EvidenceTier::Sampled`]: this probes the states in `test_states` and
/// nothing else — two trivial defaults (empty, and `{x:1, y:2}`) when you pass
/// none. `false` is a witness: some supplied state separates the two proposals.
/// `true` means only that none of the sampled states did, which is why the
/// return type is a bare `bool` with no result object to hang a tier on — read
/// this doc comment as the tier.
pub fn equivalent(
    p1: &ActionProposal,
    p2: &ActionProposal,
    test_states: Option<&[HashMap<String, Value>]>,
) -> bool {
    let defaults = vec![
        HashMap::new(),
        [
            ("x".to_string(), Value::from(1)),
            ("y".to_string(), Value::from(2)),
        ]
        .into(),
    ];
    let states = test_states.unwrap_or(&defaults);

    for state in states {
        let s1 = simulate(p1, Some(state));
        let s2 = simulate(p2, Some(state));
        if s1 != s2 {
            return false;
        }
    }
    true
}

/// Optimize a proposal: remove phantom dependencies to enable more parallelism.
pub fn optimize(proposal: &ActionProposal) -> ActionProposal {
    // Find which keys are actually written
    let mut written_keys = HashSet::new();
    for action in &proposal.actions {
        if action.action_type == ActionType::StateWrite {
            if let Some(k) = action.parameters.get("key").and_then(|v| v.as_str()) {
                written_keys.insert(k.to_string());
            }
        }
        for key in action.expected_effects.keys() {
            written_keys.insert(key.clone());
        }
    }

    let optimized_actions: Vec<Action> = proposal
        .actions
        .iter()
        .map(|action| {
            let pruned: Vec<String> = action
                .state_dependencies
                .iter()
                .filter(|d| written_keys.contains(d.as_str()))
                .cloned()
                .collect();

            if pruned.len() != action.state_dependencies.len() {
                let mut new_action = action.clone();
                new_action.state_dependencies = pruned;
                new_action
            } else {
                action.clone()
            }
        })
        .collect();

    ActionProposal {
        id: proposal.id.clone(),
        source: proposal.source.clone(),
        actions: optimized_actions,
        timestamp: proposal.timestamp,
        context: proposal.context.clone(),
    }
}

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

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

    fn state_write(id: &str, key: &str, value: Value) -> Action {
        {
            let mut a = Action::new(ActionType::StateWrite);
            a.id = id.to_string();
            a.parameters = [
                ("key".to_string(), Value::from(key)),
                ("value".to_string(), value),
            ]
            .into();
            a
        }
    }

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

    #[test]
    fn verify_valid_proposal() {
        let p = prop(vec![state_write("a1", "x", Value::from(1)), {
            let mut a = tool_call("a2", "search");
            a.state_dependencies = vec!["x".to_string()];
            a
        }]);
        let r = verify(&p, None, Some(&["search".to_string()].into()), 30);
        assert!(r.valid);
    }

    // --- tool-parameter schema validation (car-releases#56) ---

    fn echo_schema_parameters() -> Value {
        serde_json::json!({
            "type": "object",
            "properties": { "msg": { "type": "string" } },
            "required": ["msg"],
        })
    }

    fn schema_map(parameters: Value) -> HashMap<String, ToolSchema> {
        [(
            "echo".to_string(),
            ToolSchema {
                name: "echo".to_string(),
                description: String::new(),
                parameters,
                returns: None,
                idempotent: true,
                cache_ttl_secs: None,
                rate_limit: None,
            },
        )]
        .into()
    }

    fn echo_call(params: HashMap<String, Value>) -> ActionProposal {
        let mut a = tool_call("a1", "echo");
        a.parameters = params;
        prop(vec![a])
    }

    #[test]
    fn schema_verify_accepts_well_typed_params() {
        let p = echo_call([("msg".to_string(), Value::from("hi"))].into());
        let r = verify_with_schemas(&p, None, Some(&schema_map(echo_schema_parameters())), 30);
        assert!(r.valid, "{:?}", r.issues);
    }

    #[test]
    fn schema_verify_rejects_type_mismatch() {
        let p = echo_call([("msg".to_string(), Value::from(42))].into());
        let r = verify_with_schemas(&p, None, Some(&schema_map(echo_schema_parameters())), 30);
        assert!(!r.valid);
        assert!(r
            .issues
            .iter()
            .any(|i| i.message.contains("wrong type") && i.message.contains("msg")));
    }

    #[test]
    fn schema_verify_rejects_missing_required() {
        let p = echo_call(HashMap::new());
        let r = verify_with_schemas(&p, None, Some(&schema_map(echo_schema_parameters())), 30);
        assert!(!r.valid);
        assert!(
            r.issues
                .iter()
                .any(|i| i.message.contains("missing required parameter")
                    && i.message.contains("msg"))
        );
    }

    #[test]
    fn schema_verify_rejects_unknown_tool() {
        let mut a = tool_call("a1", "nope");
        a.parameters = [("msg".to_string(), Value::from("hi"))].into();
        let r = verify_with_schemas(
            &prop(vec![a]),
            None,
            Some(&schema_map(echo_schema_parameters())),
            30,
        );
        assert!(!r.valid);
        assert!(r
            .issues
            .iter()
            .any(|i| i.message.contains("not registered")));
    }

    #[test]
    fn name_only_verify_still_skips_param_validation() {
        // Back-compat: verify() with names checks existence only. A
        // bad parameter type must NOT be flagged when no schema is
        // supplied — that path has no schema to validate against.
        let p = echo_call([("msg".to_string(), Value::from(42))].into());
        let r = verify(&p, None, Some(&["echo".to_string()].into()), 30);
        assert!(
            r.valid,
            "name-only verify must not validate params: {:?}",
            r.issues
        );
    }

    #[test]
    fn schema_verify_accepts_integer_and_union_types() {
        let parameters = serde_json::json!({
            "type": "object",
            "properties": {
                "n": { "type": "integer" },
                "maybe": { "type": ["string", "null"] },
            },
            "required": ["n"],
        });
        let p = echo_call(
            [
                ("n".to_string(), Value::from(7)),
                ("maybe".to_string(), Value::Null),
            ]
            .into(),
        );
        let r = verify_with_schemas(&p, None, Some(&schema_map(parameters)), 30);
        assert!(r.valid, "{:?}", r.issues);
    }

    #[test]
    fn schema_verify_empty_schema_imposes_no_constraints() {
        // Default `{}` parameters schema -> existence only, no param
        // checks (preserves behavior for tools registered without a
        // detailed schema).
        let p = echo_call([("anything".to_string(), Value::from(42))].into());
        let r = verify_with_schemas(&p, None, Some(&schema_map(serde_json::json!({}))), 30);
        assert!(r.valid, "{:?}", r.issues);
    }

    #[test]
    fn verify_catches_unsatisfied_precondition() {
        let mut a = tool_call("a1", "deploy");
        a.preconditions = vec![Precondition {
            key: "tests_passed".to_string(),
            operator: "eq".to_string(),
            value: Value::Bool(true),
            description: String::new(),
        }];
        let r = verify(&prop(vec![a]), None, None, 30);
        assert!(!r.valid);
    }

    #[test]
    fn verify_precondition_satisfied_by_earlier_action() {
        let mut a2 = tool_call("a2", "deploy");
        a2.preconditions = vec![Precondition {
            key: "ready".to_string(),
            operator: "eq".to_string(),
            value: Value::Bool(true),
            description: String::new(),
        }];
        a2.state_dependencies = vec!["ready".to_string()];

        let p = prop(vec![state_write("a1", "ready", Value::Bool(true)), a2]);
        let r = verify(&p, None, None, 30);
        assert!(r.valid);
    }

    #[test]
    fn verify_missing_state_dependency() {
        let mut a = tool_call("a1", "x");
        a.state_dependencies = vec!["nonexistent".to_string()];
        let r = verify(&prop(vec![a]), None, None, 30);
        assert!(!r.valid);
    }

    #[test]
    fn verify_tool_not_registered() {
        let a = tool_call("a1", "quantum");
        let r = verify(&prop(vec![a]), None, Some(&HashSet::new()), 30);
        assert!(!r.valid);
    }

    #[test]
    fn compensation_naming_an_unregistered_tool_is_a_finding() {
        // A `compensable` contract is only worth what its declared undo is
        // worth. A compensation naming a tool that does not exist is a
        // rollback plan that cannot run, and the moment anyone finds out is
        // the moment it is worth least.
        let mut a = tool_call("a1", "poll");
        a.reversibility = car_ir::Reversibility::Compensable;
        a.compensation = Some(car_ir::Compensation::Tool {
            tool: "db.delet".into(), // typo
            parameters: Default::default(),
        });
        let r = verify(&prop(vec![a]), None, Some(&["poll".to_string()].into()), 30);
        assert!(!r.valid);
        assert!(r
            .issues
            .iter()
            .any(|i| i.message.contains("compensation names tool 'db.delet'")));

        // The same declaration against a registry that has the tool is fine.
        let mut a = tool_call("a1", "poll");
        a.reversibility = car_ir::Reversibility::Compensable;
        a.compensation = Some(car_ir::Compensation::Tool {
            tool: "undo".into(),
            parameters: Default::default(),
        });
        let reg = ["poll".to_string(), "undo".to_string()].into();
        assert!(verify(&prop(vec![a]), None, Some(&reg), 30).valid);
    }

    #[test]
    fn compensation_action_ref_must_resolve_within_the_proposal() {
        // Resolvable with no registry at all — the referent is in the batch.
        let mut a = tool_call("a1", "deploy");
        a.reversibility = car_ir::Reversibility::Compensable;
        a.compensation = Some(car_ir::Compensation::ActionRef {
            action_id: "rollback-1".into(),
        });
        let r = verify(&prop(vec![a.clone()]), None, None, 30);
        assert!(!r.valid);
        assert!(r
            .issues
            .iter()
            .any(|i| i.message.contains("references action 'rollback-1'")));

        // With the referenced action actually present, it resolves.
        let mut undo = tool_call("rollback-1", "rollback");
        undo.id = "rollback-1".into();
        let r = verify(&prop(vec![a, undo]), None, None, 30);
        assert!(
            !r.issues
                .iter()
                .any(|i| i.message.contains("references action")),
            "{:?}",
            r.issues
        );
    }

    #[test]
    fn compensable_with_no_compensation_declared_is_a_finding() {
        let mut a = tool_call("a1", "poll");
        a.reversibility = car_ir::Reversibility::Compensable;
        a.compensation = None;
        let r = verify(&prop(vec![a]), None, None, 30);
        assert!(!r.valid);
        assert!(r.issues.iter().any(|i| i
            .message
            .contains("declares reversibility 'compensable' but no compensation")));
        // Every compensation finding is an exact lookup, never a guess.
        assert!(r
            .issues_with_tier(EvidenceTier::DecisionProcedure)
            .iter()
            .any(|i| i.message.contains("no compensation")));
        // ...and the check reports itself as having run.
        let rec = r
            .evidence
            .checks
            .iter()
            .find(|c| c.name == "compensation_resolution")
            .expect("compensation_resolution check is recorded");
        assert!(rec.ran);
        assert_eq!(rec.findings, 1);
    }

    #[test]
    fn verify_no_tool_specified() {
        let mut a = tool_call("a1", "x");
        a.tool = None;
        let r = verify(&prop(vec![a]), None, None, 30);
        assert!(!r.valid);
    }

    #[test]
    fn detect_write_conflict() {
        let p = prop(vec![
            state_write("a1", "x", Value::from(1)),
            state_write("a2", "x", Value::from(2)),
        ]);
        let r = verify(&p, None, None, 30);
        assert!(!r.conflicts.is_empty());
    }

    #[test]
    fn simulate_state_writes() {
        let p = prop(vec![
            state_write("a1", "x", Value::from(10)),
            state_write("a2", "y", Value::from(20)),
        ]);
        let s = simulate(&p, None);
        assert_eq!(s.get("x"), Some(&Value::from(10)));
        assert_eq!(s.get("y"), Some(&Value::from(20)));
    }

    /// Parslee-ai/car#622 — the reported case. A deploy gated on
    /// `tests_passed == true`, simulated from a state where it is `false`, used
    /// to come back `deployed: true`: `simulate` shared `verify`'s optimistic
    /// effect application, so it predicted the effects of an action the
    /// executor would reject before dispatch.
    #[test]
    fn simulate_skips_effects_of_a_provably_blocked_action() {
        let mut deploy = tool_call("deploy", "deploy");
        deploy.preconditions = vec![Precondition {
            key: "tests_passed".to_string(),
            operator: "eq".to_string(),
            value: Value::Bool(true),
            description: String::new(),
        }];
        deploy
            .expected_effects
            .insert("deployed".to_string(), Value::Bool(true));
        let p = prop(vec![deploy]);

        let failing: HashMap<String, Value> =
            [("tests_passed".to_string(), Value::Bool(false))].into();
        let s = simulate(&p, Some(&failing));
        assert_eq!(
            s.get("deployed"),
            None,
            "a deploy whose precondition provably fails must not appear deployed: {s:?}"
        );

        // And it still predicts the effects when the precondition holds.
        let passing: HashMap<String, Value> =
            [("tests_passed".to_string(), Value::Bool(true))].into();
        let s = simulate(&p, Some(&passing));
        assert_eq!(s.get("deployed"), Some(&Value::Bool(true)));
    }

    /// `verify` keeps applying effects past a failure on purpose: it reports
    /// every problem in one pass, and withholding effects would bury the real
    /// findings under knock-on "dependency not available" issues. Its behaviour
    /// must not change with the simulate fix.
    #[test]
    fn verify_stays_optimistic_so_it_reports_every_finding() {
        let mut deploy = tool_call("deploy", "deploy");
        deploy.preconditions = vec![Precondition {
            key: "tests_passed".to_string(),
            operator: "eq".to_string(),
            value: Value::Bool(true),
            description: String::new(),
        }];
        deploy
            .expected_effects
            .insert("deployed".to_string(), Value::Bool(true));
        let mut notify = tool_call("notify", "notify");
        notify.state_dependencies = vec!["deployed".to_string()];
        let p = prop(vec![deploy, notify]);

        let failing: HashMap<String, Value> =
            [("tests_passed".to_string(), Value::Bool(false))].into();
        let r = verify(&p, Some(&failing), None, 30);

        assert!(!r.valid);
        // Exactly one finding: the precondition. `notify` must NOT also be
        // flagged for a missing `deployed`, which is the cascade the optimism
        // exists to suppress.
        assert_eq!(
            r.errors().len(),
            1,
            "expected only the precondition finding, got {:?}",
            r.issues
        );
        assert!(r.issues[0].message.contains("precondition will fail"));
    }

    /// The block propagates along data dependencies, the way the executor
    /// produces it — no `failure_behavior` modelling needed.
    #[test]
    fn simulate_cascade_follows_data_dependencies() {
        let mut build = tool_call("build", "build");
        build.preconditions = vec![Precondition {
            key: "ready".to_string(),
            operator: "eq".to_string(),
            value: Value::Bool(true),
            description: String::new(),
        }];
        build
            .expected_effects
            .insert("artifact".to_string(), Value::from("app.tar.gz"));
        let mut deploy = tool_call("deploy", "deploy");
        deploy.state_dependencies = vec!["artifact".to_string()];
        deploy
            .expected_effects
            .insert("deployed".to_string(), Value::Bool(true));

        let s = simulate(&prop(vec![build, deploy]), None);
        assert_eq!(
            s.get("artifact"),
            None,
            "blocked build produced no artifact"
        );
        assert_eq!(
            s.get("deployed"),
            None,
            "deploy depends on the artifact that never appeared: {s:?}"
        );
    }

    /// `equivalent` compares `simulate` output, so it inherited the bug: two
    /// proposals differing only in a precondition that gates one of them read
    /// as equivalent.
    #[test]
    fn equivalent_distinguishes_a_gated_proposal_from_an_ungated_one() {
        let mut gated = tool_call("a", "deploy");
        gated.preconditions = vec![Precondition {
            key: "tests_passed".to_string(),
            operator: "eq".to_string(),
            value: Value::Bool(true),
            description: String::new(),
        }];
        gated
            .expected_effects
            .insert("deployed".to_string(), Value::Bool(true));

        let mut ungated = tool_call("b", "deploy");
        ungated
            .expected_effects
            .insert("deployed".to_string(), Value::Bool(true));

        let failing: Vec<HashMap<String, Value>> =
            vec![[("tests_passed".to_string(), Value::Bool(false))].into()];
        assert!(
            !equivalent(&prop(vec![gated]), &prop(vec![ungated]), Some(&failing)),
            "a gate that blocks one proposal and not the other is a real difference"
        );
    }

    #[test]
    fn equivalent_proposals() {
        let p1 = prop(vec![
            state_write("a1", "x", Value::from(1)),
            state_write("a2", "y", Value::from(2)),
        ]);
        let p2 = prop(vec![
            state_write("b1", "y", Value::from(2)),
            state_write("b2", "x", Value::from(1)),
        ]);
        assert!(equivalent(&p1, &p2, None));
    }

    #[test]
    fn non_equivalent_proposals() {
        let p1 = prop(vec![state_write("a1", "x", Value::from(1))]);
        let p2 = prop(vec![state_write("b1", "x", Value::from(99))]);
        assert!(!equivalent(&p1, &p2, None));
    }

    #[test]
    fn optimize_removes_phantom_deps() {
        let mut a = tool_call("a1", "search");
        a.state_dependencies = vec!["phantom".to_string()];
        let p = prop(vec![a]);
        let optimized = optimize(&p);
        assert!(optimized.actions[0].state_dependencies.is_empty());
    }

    #[test]
    fn optimize_preserves_real_deps() {
        let mut a2 = tool_call("a2", "x");
        a2.state_dependencies = vec!["x".to_string()];
        let p = prop(vec![state_write("a1", "x", Value::from(1)), a2]);
        let optimized = optimize(&p);
        assert_eq!(optimized.actions[1].state_dependencies, vec!["x"]);
    }

    #[test]
    fn loop_detection_duplicates() {
        let p = prop(vec![tool_call("a1", "search"), tool_call("a2", "search")]);
        let r = verify(&p, None, None, 30);
        assert!(r.issues.iter().any(|i| i.message.contains("duplicate")));
    }

    #[test]
    fn loop_detection_triple() {
        let p = prop(vec![
            tool_call("a1", "search"),
            tool_call("a2", "search"),
            tool_call("a3", "search"),
        ]);
        let r = verify(&p, None, None, 30);
        assert!(!r.valid);
        assert!(r.issues.iter().any(|i| i.message.contains("likely loop")));
    }

    #[test]
    fn resource_bounds() {
        let actions: Vec<Action> = (0..35)
            .map(|i| tool_call(&format!("a{}", i), &format!("t{}", i)))
            .collect();
        let r = verify(&prop(actions), None, None, 30);
        assert!(r.issues.iter().any(|i| i.message.contains("excessive")));
    }

    // --- evidence bundle (§5.2.2) ---

    #[test]
    fn evidence_declares_all_check_scopes() {
        let p = echo_call([("msg".to_string(), Value::from("hi"))].into());
        let r = verify_with_schemas(&p, None, Some(&schema_map(echo_schema_parameters())), 30);
        // Every check category is present with a non-empty scope.
        for want in [
            "resource_bounds",
            "loop_detection",
            "preconditions",
            "state_dependencies",
            "tool_existence",
            "param_schema",
            "write_conflicts",
        ] {
            let rec = r
                .evidence
                .checks
                .iter()
                .find(|c| c.name == want)
                .unwrap_or_else(|| panic!("missing check record {want}"));
            assert!(!rec.verifies.is_empty());
            assert!(!rec.cannot_verify.is_empty());
        }
        // With schemas supplied, both conditional checks ran.
        let by = |n: &str| r.evidence.checks.iter().find(|c| c.name == n).unwrap();
        assert!(by("param_schema").ran);
        assert!(by("tool_existence").ran);
    }

    #[test]
    fn evidence_marks_param_schema_skipped_without_schemas() {
        // Tool call but no schemas: param_schema can't run; confidence
        // is docked and the blind spot is recorded as an assumption.
        let p = prop(vec![tool_call("a1", "search")]);
        let r = verify(&p, None, None, 30);
        let param = r
            .evidence
            .checks
            .iter()
            .find(|c| c.name == "param_schema")
            .unwrap();
        assert!(!param.ran);
        assert!(
            r.evidence.confidence < 1.0,
            "skipped check should dock coverage"
        );
        assert!(r
            .evidence
            .assumptions
            .iter()
            .any(|a| a.contains("well-formed")));
    }

    #[test]
    fn evidence_full_confidence_for_pure_state_writes() {
        // No tool calls, fully-known state, no warnings: coverage is 1.0.
        let p = prop(vec![state_write("a1", "x", Value::from(1))]);
        let r = verify(&p, None, None, 30);
        assert!(r.valid);
        assert_eq!(r.evidence.confidence, 1.0);
        assert!(r.evidence.untested_regions.is_empty());
    }

    #[test]
    fn evidence_conflicts_become_residual_risk() {
        // Two undeclared writers to the same key: warning, not error, so
        // it must surface as a residual risk rather than vanish.
        let p = prop(vec![
            state_write("a1", "k", Value::from(1)),
            state_write("a2", "k", Value::from(2)),
        ]);
        let r = verify(&p, None, None, 30);
        assert!(r.valid, "conflicts are warnings, not errors");
        assert!(!r.conflicts.is_empty());
        assert!(r
            .evidence
            .residual_risks
            .iter()
            .any(|s| s.contains("write conflict")));
        let wc = r
            .evidence
            .checks
            .iter()
            .find(|c| c.name == "write_conflicts")
            .unwrap();
        assert_eq!(wc.findings, r.conflicts.len());
    }

    #[test]
    fn evidence_untested_includes_runtime_set_effect_keys() {
        // A tool whose declared effect writes `out`: the *key* exists
        // statically but its *value* is runtime-determined, so it is an
        // untested region — not just the tool's opaque return (neo M1).
        let mut a = tool_call("a1", "fetch");
        a.expected_effects = [("out".to_string(), Value::from("placeholder"))].into();
        let r = verify(
            &prop(vec![a]),
            None,
            Some(&["fetch".to_string()].into()),
            30,
        );
        assert!(r
            .evidence
            .untested_regions
            .iter()
            .any(|s| s.contains("state key 'out'")));
        assert!(r
            .evidence
            .untested_regions
            .iter()
            .any(|s| s.contains("runtime output of tool 'fetch'")));
    }

    #[test]
    fn evidence_tool_existence_ran_consistent_with_findings() {
        // Malformed tool_call (no tool named) with no registry supplied:
        // the existence record must not claim ran:false while reporting a
        // finding (neo m1).
        let mut a = tool_call("a1", "x");
        a.tool = None;
        let r = verify(&prop(vec![a]), None, None, 30);
        assert!(!r.valid);
        let te = r
            .evidence
            .checks
            .iter()
            .find(|c| c.name == "tool_existence")
            .unwrap();
        assert!(te.findings >= 1);
        assert!(
            te.ran,
            "ran must be true whenever the check produced a finding"
        );
    }

    // --- evidence tiers ---

    /// The loop rule is the one heuristic in `verify`, and the tier is how a
    /// caller learns that without recognising the message. Pinning it here
    /// means a later refactor that mislabels it fails a test rather than
    /// quietly presenting a rule of thumb as an exact result.
    #[test]
    fn loop_detection_findings_are_heuristic_and_the_rest_are_not() {
        let p = prop(vec![
            tool_call("a1", "poll"),
            tool_call("a2", "poll"),
            tool_call("a3", "poll"),
            tool_call("a4", "ghost"),
        ]);
        let r = verify(&p, None, Some(&["poll".to_string()].into()), 30);

        let heuristic = r.issues_with_tier(EvidenceTier::Heuristic);
        assert_eq!(
            heuristic.len(),
            1,
            "only the repeated-call finding is heuristic: {:?}",
            r.issues
        );
        assert!(heuristic[0]
            .message
            .contains("repeated identical tool call"));

        // The unregistered tool is set membership — exactly decided.
        let decided = r.issues_with_tier(EvidenceTier::DecisionProcedure);
        assert!(decided
            .iter()
            .any(|i| i.message.contains("'ghost' is not registered")));

        // Nothing in `verify` samples anything.
        assert!(r.issues_with_tier(EvidenceTier::Sampled).is_empty());
    }

    /// Every issue's tier must agree with the tier on the check that reported
    /// it — the two are the same claim at different granularity, and a caller
    /// reading either must get the same answer.
    ///
    /// The correlation is done by counting, not by name: for each tier, the
    /// `findings` declared by the checks carrying that tier must equal the
    /// number of issues actually carrying it, and the totals must account for
    /// every issue. That catches the failure a per-name assertion misses — a
    /// future finding site emitting, say, a `Heuristic` issue from under the
    /// `write_conflicts` record, which would leave the two views of the same
    /// verdict disagreeing.
    #[test]
    fn check_records_and_issues_agree_on_tier() {
        let p = prop(vec![
            // Two identical calls to a registered tool: loop_detection
            // (Heuristic) reports one duplicate.
            tool_call("a1", "poll"),
            tool_call("a2", "poll"),
            // Unregistered tool reading a key nobody writes: tool_existence
            // and state_dependencies, one finding each (DecisionProcedure).
            {
                let mut a = tool_call("a3", "ghost");
                a.state_dependencies = vec!["missing".to_string()];
                a
            },
            // Two undeclared-order writers of one key: write_conflicts
            // (DecisionProcedure).
            state_write("a4", "x", Value::from(1)),
            state_write("a5", "x", Value::from(2)),
        ]);
        let r = verify(&p, None, Some(&["poll".to_string()].into()), 30);

        // Listed explicitly rather than iterated: `EvidenceTier` has no `Ord`
        // and no variant count, so a new tier has to be added here by hand —
        // which is the intended nudge to decide what it means for this
        // invariant.
        for tier in [
            EvidenceTier::DecisionProcedure,
            EvidenceTier::Heuristic,
            EvidenceTier::Sampled,
        ] {
            let declared: usize = r
                .evidence
                .checks
                .iter()
                .filter(|c| c.tier == tier)
                .map(|c| c.findings)
                .sum();
            let actual = r.issues_with_tier(tier).len();
            assert_eq!(
                declared,
                actual,
                "checks at tier {} declare {declared} findings but {actual} issues carry it: {:?}",
                tier.as_str(),
                r.issues
            );
        }

        // …and between them the checks account for every issue, so a mismatch
        // can't hide as an issue no check claims.
        let total: usize = r.evidence.checks.iter().map(|c| c.findings).sum();
        assert_eq!(total, r.issues.len(), "unaccounted issues: {:?}", r.issues);

        // Non-vacuity: the fixture really does exercise both tiers that
        // `verify` can produce.
        assert_eq!(
            r.issues_with_tier(EvidenceTier::Heuristic).len(),
            1,
            "expected exactly the duplicate-call finding: {:?}",
            r.issues
        );
        assert!(
            r.issues_with_tier(EvidenceTier::DecisionProcedure).len() >= 3,
            "expected the unregistered tool, the missing dependency, and the \
             write conflict: {:?}",
            r.issues
        );
    }

    /// The tier travels over the wire as a stable snake_case string; the FFI
    /// and JSON-RPC surfaces depend on these exact labels.
    #[test]
    fn tier_serializes_as_stable_snake_case() {
        let p = prop(vec![tool_call("a1", "ghost")]);
        let r = verify(&p, None, Some(&HashSet::new()), 30);
        let json = serde_json::to_value(&r.issues[0]).expect("issue serializes");
        assert_eq!(json["tier"], Value::from("decision_procedure"));
        assert_eq!(
            json["tier"],
            Value::from(r.issues[0].tier.as_str()),
            "as_str and the serde representation must not drift"
        );
    }
}