car-engine 0.62.0

//! Static plan verification as an admission gate.
//!
//! `car-verify` could already reject a bad plan before any tool ran — but the
//! only call site that did so lived inside the executor's *replan* loop
//! (`ReplanConfig::verify_before_execute`). That loop needs a registered
//! [`crate::ReplanCallback`], and it is bounded by `max_replans`, which defaults
//! to `0`. No production path registered a callback, so the check never fired on
//! real work: the README promised verification before execution and the runtime
//! delivered it only for a replan that never happened.
//!
//! [`StaticVerificationGate`] moves the same check to the seam that *does* run on
//! every proposal — [`crate::admission::AdmissionGate`], the same place the
//! information-flow gate already sits.
//!
//! **What this is worth, precisely.** `validate_action` already checks tool
//! existence, parameter schemas, preconditions, and state dependencies — per
//! action, and with a *stronger* schema validator than car-verify's (the full
//! `jsonschema` crate versus a hand-rolled `required`+`type` subset). So on a
//! **single-action** proposal this gate adds no coverage at all: rejecting "the
//! whole proposal" and rejecting "the one action" are the same thing, and the
//! validator gets there anyway.
//!
//! The value is entirely on **multi-action** proposals, and it is about *timing*,
//! not coverage. `validate_action` runs as execution reaches each action, so a bad
//! tool name in action 5 is discovered after actions 1–4 have already had their
//! side effects. This gate sees the whole proposal up front and refuses it before
//! the first dispatch, so nothing partial happens. Register it on any runtime that
//! accepts externally-authored multi-action proposals; on a runtime that only ever
//! submits one action at a time it is close to inert.
//!
//! **What blocks, and what is only advisory.** Blocking is limited to
//! *state-independent* findings — an unregistered tool, parameters that violate the
//! registered schema, and a `tool_call` with no tool — plus one state-dependent
//! finding narrow enough to share their footing: an **unmeetable dependency**
//! (below), which is decided against the executor's own ordering. The state-independent three are exact — they cannot be wrong
//! about a plan that would in fact have run.
//!
//! Every other state-dependent finding is advisory, because the forward model is
//! incomplete: it applies only the effects an action *declares* in
//! `expected_effects`. An action that really writes a key without declaring it is
//! invisible, so a downstream precondition or `state_dependency` reading that key
//! is reported as failing even though execution would have succeeded. Blocking a
//! whole proposal on that would be a false rejection, and `StaticState::is_unknown`
//! does not save us — nothing in the workspace ever populates `unknown_keys`, so it
//! is always false. The loop heuristic (`count >= 3`) is advisory for the same
//! reason: three legitimate polls are indistinguishable from a runaway loop.
//!
//! Preconditions still get enforced — by `validate_action`, at execution time,
//! against live state, where the answer is accurate. Write conflicts are warnings
//! upstream and never blocked.
//!
//! **Unmeetable dependencies.** `car-ir`'s DAG edges only point to lower
//! indices, so a dependency cycle cannot be drawn there; it surfaces as unmet
//! dependencies, and left advisory that meant partial execution (car#3060): the
//! validator refused the cycle's actions at run time after the rest of the
//! proposal had already run. The gate now refuses the whole proposal when an
//! action is certain to be refused that way, which takes all of:
//!
//! - it needs (through `state_dependencies`, or a precondition that fails when
//!   its key is absent) a key that is **absent from live state at admission**
//!   — read in the namespace the action executes in (car#3121): under a
//!   tenant-scoped proposal, a `state_write` / `state_read` / `assertion`
//!   reads its tenant's keys, so a bare shared key does not count for it,
//!   while a `tool_call` runs on the unscoped path and reads shared keys;
//! - no declared writer of that key (in `expected_effects`, or as a
//!   `state_write`'s key) that is not itself certain to be refused **finishes
//!   before it** — "before" meaning the executor's own order: an
//!   earlier `build_dag` level, or earlier in a level that runs sequentially
//!   (one action, or any action with `FailureBehavior::Abort`, the default);
//! - no such **tool call** finishes before it either, because a tool's handler
//!   can write state it never declares: a WebSocket client's `tools.execute`
//!   handler sending `state.set` on its connection, a NAPI / PyO3 / UniFFI
//!   callback calling `state_set`, a car-multi delegate or spawned subtask
//!   writing the parent's shared store. A streaming or long-running tool call
//!   returns a handle before its handler is done, so it does not count.
//!
//! A cycle whose members wait only on each other is the main case; a key
//! whose only declared writer runs later, or that nothing in the proposal
//! declares writing, is the same defect without one. Such a proposal cannot
//! complete through its declared effects and ordering.
//!
//! What stays advisory, still refused per action at run time while the other
//! actions run:
//!
//! - **A tool call that finishes first.** The issue's own shape, `[a0, a1, a2]`
//!   with a0 a `tool_call` and the default failure behavior, is one: the level
//!   holding a0 and a1 runs sequentially, so a0's handler may set a1's key, and
//!   the plan may complete.
//! - **A needed key already in live state.**
//!
//! The residual is a race. Two actions in a level that runs concurrently are
//! ordered neither way, so a tool call there could still set a waiting
//! action's key before it is validated; so could a streaming or long-running
//! tool call ordered earlier whose handler happens to finish in time; and a
//! write from outside the proposal (another client sending `state.set` on the
//! same session) can land at any time — for a key nothing declares writing,
//! that is the only way it could be set. The gate refuses those plans, which
//! complete only by timing. The
//! refusal names the key, so the fix is to declare the write on an action that
//! runs earlier, or to set the key before submitting. Without live state
//! (`blocking_errors` given `None`) the check does not run.

use crate::admission::{AdmissionGate, GateContext, GateOutcome};
use car_ir::precondition::{check_precondition, StateView};
use car_ir::{ActionProposal, ActionType, ToolSchema};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use tokio::sync::RwLock as TokioRwLock;

/// Default ceiling on how many actions the gate will walk. Matches the
/// executor's replan-path budget so both verification points agree.
const DEFAULT_MAX_ACTIONS: usize = 100;

/// Message fragments identifying the findings that are *state-dependent* and
/// therefore advisory (see the module docs). `advisory_findings_do_not_block`
/// pins the coupling: change any of these messages upstream and the test fails,
/// rather than the gate silently starting to reject legitimate proposals.
///
/// **`VerifyIssue::tier` is not the structured kind this wanted.** The tier says
/// how a finding was *derived* — decision procedure, heuristic, or sampling —
/// and only the loop finding here is a heuristic. `precondition will fail` and
/// `not available at this point` are `EvidenceTier::DecisionProcedure`: exactly
/// decided, over a forward model that sees only *declared* effects. That gap,
/// not the derivation, is why they can't block. Keying this list on
/// `tier != Heuristic` would start rejecting proposals on the two findings the
/// module docs above spend a paragraph explaining must stay advisory.
const ADVISORY_ISSUE_FRAGMENTS: &[&str] = &[
    // `count >= 3` heuristic — three legitimate polls look identical to a loop.
    "repeated identical tool call",
    // Both depend on the forward model, which only sees *declared* effects.
    "precondition will fail",
    "not available at this point",
];

/// Whether a car-verify issue is exact enough to refuse a whole proposal on.
///
/// Blocking car-verify findings are state-independent: they are true regardless of
/// what the tools actually do, so they cannot false-reject a plan that would have
/// run. The one state-dependent refusal, an unmeetable dependency, is not a
/// car-verify finding; [`blocking_errors`] adds it.
pub fn is_blocking_issue(issue: &car_verify::VerifyIssue) -> bool {
    issue.severity == "error"
        && !ADVISORY_ISSUE_FRAGMENTS
            .iter()
            .any(|frag| issue.message.contains(frag))
}

/// Run static verification and return the blocking errors, if any.
///
/// Shared by [`StaticVerificationGate`] and the executor's replan quality gate so
/// the two verification points cannot disagree about what counts as fatal — they
/// previously did, with the replan path blocking on the loop heuristic this gate
/// deliberately treats as advisory.
///
/// An **empty** `tools` map means "this runtime doesn't declare its tools", not
/// "no tool exists". Passing it through would flag every `tool_call` as
/// unregistered and reject every proposal, turning a safety check into a denial of
/// service for any embedder that executes via callback without registering
/// schemas. In that case tool-existence and parameter-schema checking are skipped.
///
/// Note `Runtime::restore_checkpoint` rebuilds the registry with name-only schemas
/// (`parameters: {}`), so the map stays non-empty and existence still checks, but
/// parameter validation silently becomes a no-op after a restore.
///
/// `state` is the keys visible to `tenant`
/// (`StateStore::versioned_snapshot_visible_to`), or the whole store when
/// unscoped, and `tenant` the proposal's tenant, if any: each action's keys
/// are resolved in the namespace it executes in (car#3121, see the module
/// docs).
pub fn blocking_errors(
    proposal: &ActionProposal,
    state: Option<&HashMap<String, serde_json::Value>>,
    tenant: Option<&str>,
    tools: &HashMap<String, ToolSchema>,
    max_actions: usize,
) -> Vec<car_verify::VerifyIssue> {
    let result = if tools.is_empty() {
        car_verify::verify(proposal, state, None, max_actions)
    } else {
        car_verify::verify_with_schemas(proposal, state, Some(tools), max_actions)
    };
    let mut errors: Vec<car_verify::VerifyIssue> = if result.valid {
        Vec::new()
    } else {
        result
            .issues
            .into_iter()
            .filter(is_blocking_issue)
            .collect()
    };
    if let Some(state) = state {
        errors.extend(unmeetable_dependencies(proposal, state, tenant));
    }
    errors
}

/// A state view in which no key exists: asks whether a precondition fails when
/// its key is absent. A precondition reads only its own key, so the answer is
/// exact for that key.
struct NoKeys;

impl StateView for NoKeys {
    fn get_value(&self, _key: &str) -> Option<serde_json::Value> {
        None
    }

    fn key_exists(&self, _key: &str) -> bool {
        false
    }
}

/// The actions of `proposal` that the validator is certain to refuse, because a
/// key each one needs cannot be written before it is validated — one error
/// finding each (see the module docs for the rule and its residual).
///
/// An action *needs* a key when it lists it in `state_dependencies`, or has a
/// precondition on it that fails when the key is absent; only keys absent from
/// `state` count, whether or not any action declares writing them. "Before" is the
/// executor's order: an earlier `build_dag` level, or earlier in a level that
/// runs sequentially ([`crate::executor::level_runs_sequentially`]); two
/// actions in a concurrent level are ordered neither way.
///
/// Every action starts stuck. An action is freed when every key it needs has a
/// free declared writer ordered before it (trivially, when it needs nothing),
/// or when a free, non-detached tool call is ordered before it: that call's
/// handler may set the key without declaring it. What is still stuck when
/// nothing more can be freed is reported. Each free action is processed once:
/// per key, the needers sorted by order are freed from the latest down as the
/// key's earliest free writer moves earlier, and the tool rule does the same
/// over all actions.
///
/// Keys are compared as the store holds them. Under a tenant, a state action's
/// keys resolve to `tenant:<id>:<key>` and a tool call's stay shared
/// ([`crate::executor::action_state_tenant`]), so a tenant `state_write` does
/// not free a tool call's shared need. A tool call's declared effect frees the
/// key in both namespaces, because its handler can write either: callback
/// `state_changes` land in the tenant's namespace, a `state.set` in the shared
/// one.
fn unmeetable_dependencies(
    proposal: &ActionProposal,
    state: &HashMap<String, serde_json::Value>,
    tenant: Option<&str>,
) -> Vec<car_verify::VerifyIssue> {
    let actions = &proposal.actions;
    let n = actions.len();

    // Execution order as the executor runs it: `rank[j] < rank[i]` exactly when
    // j finishes before i is validated. Within a concurrent level every action
    // shares one rank, because none is guaranteed to finish first.
    let mut rank: Vec<(usize, usize)> = vec![(0, 0); n];
    for (level_no, level) in car_ir::build_dag(actions).iter().enumerate() {
        let sequential = crate::executor::level_runs_sequentially(level, actions);
        for (pos, &i) in level.iter().enumerate() {
            rank[i] = (level_no, if sequential { pos } else { 0 });
        }
    }
    let by_rank = |mut v: Vec<usize>| {
        v.sort_by_key(|&i| rank[i]);
        v
    };

    // An action's keys as the store holds them (car#3121).
    let store_key = |i: usize, key: &str| {
        car_state::tenant_key(
            crate::executor::action_state_tenant(&actions[i], tenant),
            key,
        )
    };

    // The keys each action declares writing, then declared writers per key.
    let writes: Vec<Vec<String>> = actions
        .iter()
        .map(|action| crate::executor::action_write_store_keys(action, tenant))
        .collect();
    let mut writers: HashMap<&str, Vec<usize>> = HashMap::new();
    for (i, keys) in writes.iter().enumerate() {
        for key in keys {
            writers.entry(key.as_str()).or_default().push(i);
        }
    }

    // Each needed key as (store key, the name the action uses).
    let needs: Vec<Vec<(String, &str)>> = actions
        .iter()
        .enumerate()
        .map(|(i, action)| {
            let mut keys: Vec<(String, &str)> = action
                .state_dependencies
                .iter()
                .map(String::as_str)
                .chain(
                    action
                        .preconditions
                        .iter()
                        .filter(|pre| check_precondition(pre, &NoKeys).is_some())
                        .map(|pre| pre.key.as_str()),
                )
                .map(|key| (store_key(i, key), key))
                .filter(|(stored, _)| !state.contains_key(stored))
                .collect();
            keys.sort_unstable();
            keys.dedup();
            keys
        })
        .collect();

    // Per key: its needers in execution order, and how many of them (a suffix)
    // already have a free writer ordered before them.
    let mut needers: HashMap<&str, Vec<usize>> = HashMap::new();
    for (i, keys) in needs.iter().enumerate() {
        for (key, _) in keys {
            needers.entry(key.as_str()).or_default().push(i);
        }
    }
    let mut needers: HashMap<&str, (Vec<usize>, usize)> = needers
        .into_iter()
        .map(|(key, v)| {
            let v = by_rank(v);
            let cut = v.len();
            (key, (v, cut))
        })
        .collect();
    let all_by_rank = by_rank((0..n).collect());
    let mut tool_cut = n;
    let can_write_undeclared = |j: usize| {
        actions[j].action_type == ActionType::ToolCall && !actions[j].invocation_mode.is_detached()
    };

    let mut stuck = vec![true; n];
    let mut unmet: Vec<usize> = needs.iter().map(Vec::len).collect();
    let mut worklist: Vec<usize> = (0..n).filter(|&i| needs[i].is_empty()).collect();
    for &i in &worklist {
        stuck[i] = false;
    }
    while let Some(i) = worklist.pop() {
        for key in &writes[i] {
            let Some((order, cut)) = needers.get_mut(key.as_str()) else {
                continue;
            };
            while *cut > 0 && rank[order[*cut - 1]] > rank[i] {
                *cut -= 1;
                let m = order[*cut];
                unmet[m] -= 1;
                if stuck[m] && unmet[m] == 0 {
                    stuck[m] = false;
                    worklist.push(m);
                }
            }
        }
        if can_write_undeclared(i) {
            while tool_cut > 0 && rank[all_by_rank[tool_cut - 1]] > rank[i] {
                tool_cut -= 1;
                let m = all_by_rank[tool_cut];
                if stuck[m] {
                    stuck[m] = false;
                    worklist.push(m);
                }
            }
        }
    }

    (0..n)
        .filter(|&i| stuck[i])
        .filter_map(|i| {
            // A key this action still waits on: no free writer is ordered
            // before it, so it is not in this key's freed suffix.
            let (stored, key) = needs[i].iter().find(|(stored, _)| {
                needers
                    .get(stored.as_str())
                    .is_some_and(|(order, cut)| !order[*cut..].contains(&i))
            })?;
            let writers_note = writers
                .get(stored.as_str())
                .map(|ws| {
                    ws.iter()
                        .map(|&j| format!("'{}'", actions[j].id))
                        .collect::<Vec<_>>()
                        .join(", ")
                })
                .map(|ids| format!("its declared writers ({ids}) are themselves refused or run after it"))
                .unwrap_or_else(|| "no action declares writing it".to_string());
            Some(car_verify::VerifyIssue::new(
                actions[i].id.clone(),
                "error",
                format!(
                    "unmeetable dependency: action '{}' needs '{key}', which is not in state; \
                     {writers_note}, and no tool call finishes before it to set '{key}', so the proposal \
                     cannot complete through its declared effects and ordering. Declare the \
                     write on an action that runs earlier, or set '{key}' before submitting",
                    actions[i].id
                ),
                // Decided over declared effects, the executor's ordering, and
                // live state at admission; the residual (a race inside a
                // concurrent level) is in the module docs.
                car_verify::EvidenceTier::DecisionProcedure,
            ))
        })
        .collect()
}

/// An admission gate that statically verifies a proposal before any action runs.
///
/// Holds a handle to the runtime's tool registry rather than a back-reference to
/// the `Runtime` itself — gates are stored *on* the runtime, so an `Arc<Runtime>`
/// here would be a cycle. The registry is already `Arc`-shared, so cloning it is
/// both cheap and always current: tools registered after this gate is built are
/// visible to it.
pub struct StaticVerificationGate {
    tools: Arc<TokioRwLock<HashMap<String, ToolSchema>>>,
    max_actions: usize,
}

impl StaticVerificationGate {
    /// Build a gate reading tool schemas from the runtime's live registry.
    pub fn new(tools: Arc<TokioRwLock<HashMap<String, ToolSchema>>>) -> Self {
        Self {
            tools,
            max_actions: DEFAULT_MAX_ACTIONS,
        }
    }

    /// Override the action ceiling (proposals longer than this are not walked).
    pub fn with_max_actions(mut self, max_actions: usize) -> Self {
        self.max_actions = max_actions;
        self
    }
}

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

    async fn check(&self, proposal: &ActionProposal, ctx: &GateContext<'_>) -> GateOutcome {
        let errors = {
            // The read guard is held across the synchronous verify call and
            // dropped before returning; car-verify does no .await, so nothing
            // can deadlock behind it.
            let tools = self.tools.read().await;
            let tenant = ctx.scope.and_then(|s| s.tenant_id.as_deref());
            blocking_errors(proposal, Some(ctx.state), tenant, &tools, self.max_actions)
        };
        if errors.is_empty() {
            return GateOutcome::Allow;
        }

        // Only real action ids go in `blocked`. car-verify attributes some
        // findings to synthetic ids (loop detection uses "proposal"), and an id
        // matching no action would make the executor fall back to the generic
        // "blocked by admission gate" message for every action, hiding the
        // actual reason.
        let real_ids: HashSet<&str> = proposal.actions.iter().map(|a| a.id.as_str()).collect();
        let blocked: HashSet<String> = errors
            .iter()
            .map(|i| i.action_id.clone())
            .filter(|id| real_ids.contains(id.as_str()))
            .collect();
        let reason = format!(
            "static verification failed: {}",
            errors
                .iter()
                .map(|i| i.message.as_str())
                .collect::<Vec<_>>()
                .join("; ")
        );
        GateOutcome::Reject { blocked, reason }
    }
}

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

    /// Build via serde rather than a struct literal: `Action` has 16 fields and
    /// no `Default`, so a literal here would need editing every time an
    /// unrelated field lands. The `#[serde(default)]` attributes are the
    /// contract we actually care about.
    fn schema(name: &str) -> ToolSchema {
        serde_json::from_value(json!({ "name": name })).expect("tool schema fixture")
    }

    fn registry(names: &[&str]) -> Arc<TokioRwLock<HashMap<String, ToolSchema>>> {
        let map = names
            .iter()
            .map(|n| (n.to_string(), schema(n)))
            .collect::<HashMap<_, _>>();
        Arc::new(TokioRwLock::new(map))
    }

    fn proposal_of(actions: serde_json::Value) -> ActionProposal {
        serde_json::from_value(json!({
            "id": "p1", "source": "test", "actions": actions,
        }))
        .expect("proposal fixture")
    }

    fn one(action_id: &str, tool: &str) -> ActionProposal {
        proposal_of(json!([{
            "id": action_id, "type": "tool_call", "tool": tool, "parameters": {},
        }]))
    }

    async fn check(
        gate: &StaticVerificationGate,
        proposal: &ActionProposal,
        state: &HashMap<String, serde_json::Value>,
    ) -> GateOutcome {
        let versions = HashMap::new();
        let ctx = GateContext {
            session_id: None,
            scope: None,
            state,
            versions: &versions,
        };
        gate.check(proposal, &ctx).await
    }

    /// The regression this gate exists to prevent: on a multi-action proposal a
    /// bad tool in a *later* action must stop the whole thing before the first
    /// action dispatches, so no partial side effects happen.
    #[tokio::test]
    async fn rejects_before_any_action_when_a_later_tool_is_unregistered() {
        let gate = StaticVerificationGate::new(registry(&["echo"]));
        let proposal = proposal_of(json!([
            { "id": "a1", "type": "tool_call", "tool": "echo", "parameters": {} },
            { "id": "a2", "type": "tool_call", "tool": "ghost", "parameters": {} },
        ]));

        match check(&gate, &proposal, &HashMap::new()).await {
            GateOutcome::Reject { blocked, reason } => {
                assert!(blocked.contains("a2"), "the offending action must be named");
                assert!(
                    !blocked.contains("a1"),
                    "the valid action is collateral, not the cause"
                );
                assert!(
                    reason.contains("static verification failed"),
                    "got: {reason}"
                );
            }
            other => panic!("expected Reject, got {other:?}"),
        }
    }

    /// Parameter-schema violations block: a missing required field is exact and
    /// state-independent, so refusing early cannot be a false rejection.
    #[tokio::test]
    async fn rejects_parameter_schema_violation() {
        let strict: ToolSchema = serde_json::from_value(json!({
            "name": "write",
            "parameters": {
                "type": "object",
                "properties": { "path": { "type": "string" } },
                "required": ["path"],
            },
        }))
        .expect("schema");
        let reg = registry(&[]);
        reg.write().await.insert("write".to_string(), strict);
        let gate = StaticVerificationGate::new(reg);

        match check(&gate, &one("a1", "write"), &HashMap::new()).await {
            GateOutcome::Reject { reason, .. } => {
                assert!(
                    reason.contains("path"),
                    "reason should name the field: {reason}"
                )
            }
            other => panic!("expected Reject for a missing required param, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn allows_a_valid_proposal() {
        let gate = StaticVerificationGate::new(registry(&["echo"]));
        assert!(
            matches!(
                check(&gate, &one("a1", "echo"), &HashMap::new()).await,
                GateOutcome::Allow
            ),
            "a valid proposal must not be blocked"
        );
    }

    /// An embedder that registers no schemas executes through its own callback.
    /// Blocking every such proposal would be a denial of service.
    #[tokio::test]
    async fn empty_registry_does_not_block_every_proposal() {
        let gate = StaticVerificationGate::new(registry(&[]));
        assert!(
            matches!(
                check(&gate, &one("a1", "anything_at_all"), &HashMap::new()).await,
                GateOutcome::Allow
            ),
            "an empty tool registry means 'undeclared', not 'nothing exists'"
        );
    }

    /// The gate holds the live registry, not a snapshot.
    #[tokio::test]
    async fn sees_tools_registered_after_construction() {
        let reg = registry(&[]);
        let gate = StaticVerificationGate::new(reg.clone());
        reg.write().await.insert("echo".to_string(), schema("echo"));
        assert!(
            matches!(
                check(&gate, &one("a1", "ghost"), &HashMap::new()).await,
                GateOutcome::Reject { .. }
            ),
            "the registry is now non-empty, so an unknown tool must be caught"
        );
    }

    /// State-dependent findings are advisory, and this pins the message
    /// coupling in `ADVISORY_ISSUE_FRAGMENTS`: it asserts car-verify still
    /// *reports* each one at error severity (so the fragments are still live)
    /// while the gate still allows. Change a message upstream and this fails,
    /// rather than the gate silently starting to reject valid proposals.
    #[tokio::test]
    async fn advisory_findings_do_not_block() {
        let gate = StaticVerificationGate::new(registry(&["poll"]));

        // 3 identical calls -> loop heuristic; unmet state_dependency -> the
        // "not available at this point" finding; failing precondition -> the
        // "precondition will fail" finding.
        let proposal = proposal_of(json!([
            { "id": "a1", "type": "tool_call", "tool": "poll", "parameters": {} },
            { "id": "a2", "type": "tool_call", "tool": "poll", "parameters": {} },
            { "id": "a3", "type": "tool_call", "tool": "poll", "parameters": {},
              "state_dependencies": ["written_but_undeclared"] },
            { "id": "a4", "type": "tool_call", "tool": "poll", "parameters": {},
              "preconditions": [{ "key": "missing", "op": "exists" }] },
        ]));

        let schemas: HashMap<String, ToolSchema> =
            [("poll".to_string(), schema("poll"))].into_iter().collect();
        let raw = car_verify::verify_with_schemas(&proposal, None, Some(&schemas), 100);
        for frag in ADVISORY_ISSUE_FRAGMENTS {
            assert!(
                raw.errors().iter().any(|i| i.message.contains(frag)),
                "car-verify no longer reports an error containing {frag:?} — update \
                 ADVISORY_ISSUE_FRAGMENTS, or the gate will start blocking on it"
            );
        }

        assert!(
            matches!(
                check(&gate, &proposal, &HashMap::new()).await,
                GateOutcome::Allow
            ),
            "state-dependent and heuristic findings must not reject a proposal"
        );
    }

    /// a1 and a2 each need a key only the other declares writing, neither key
    /// is in live state, and the only other action is a state_write of an
    /// unrelated key. The whole proposal is refused, and `blocked` names the
    /// cycle, not a0.
    #[tokio::test]
    async fn rejects_a_pure_declared_effect_cycle() {
        let gate = StaticVerificationGate::new(registry(&["step"]));
        let proposal = proposal_of(json!([
            { "id": "a0", "type": "state_write", "parameters": { "key": "k0", "value": 0 } },
            { "id": "a1", "type": "tool_call", "tool": "step", "parameters": {},
              "state_dependencies": ["k2"], "expected_effects": { "k1": 1 } },
            { "id": "a2", "type": "tool_call", "tool": "step", "parameters": {},
              "state_dependencies": ["k1"], "expected_effects": { "k2": 2 } },
        ]));

        match check(&gate, &proposal, &HashMap::new()).await {
            GateOutcome::Reject { blocked, reason } => {
                let expected: HashSet<String> = ["a1", "a2"].map(String::from).into();
                assert_eq!(blocked, expected);
                assert!(
                    reason.contains("cannot complete") && reason.contains("'k2'"),
                    "the refusal names the key it waits on: {reason}"
                );
            }
            other => panic!("expected Reject, got {other:?}"),
        }
    }

    /// An action that needs a key only it declares writing cannot complete
    /// through its declared effects and ordering either.
    #[tokio::test]
    async fn rejects_an_action_waiting_on_its_own_declared_write() {
        let gate = StaticVerificationGate::new(registry(&["step"]));
        let proposal = proposal_of(json!([
            { "id": "a1", "type": "tool_call", "tool": "step", "parameters": {},
              "preconditions": [{ "key": "k", "operator": "exists" }],
              "expected_effects": { "k": 1 } },
        ]));
        assert!(
            matches!(
                check(&gate, &proposal, &HashMap::new()).await,
                GateOutcome::Reject { .. }
            ),
            "a self-dependency on an absent key is a cycle of one"
        );
    }

    /// Each of these is the cycle above with one thing changed that lets the
    /// plan start, so the gate must allow it and leave the rest to the
    /// per-action validator.
    #[tokio::test]
    async fn allows_cycles_the_plan_can_break() {
        let gate = StaticVerificationGate::new(registry(&["step"]));
        let cycle = |a0: serde_json::Value| {
            proposal_of(json!([
                a0,
                { "id": "a1", "type": "tool_call", "tool": "step", "parameters": {},
                  "state_dependencies": ["k2"], "expected_effects": { "k1": 1 } },
                { "id": "a2", "type": "tool_call", "tool": "step", "parameters": {},
                  "state_dependencies": ["k1"], "expected_effects": { "k2": 2 } },
            ]))
        };
        let plain = json!({ "id": "a0", "type": "state_write",
                            "parameters": { "key": "k0", "value": 0 } });

        let live: HashMap<String, serde_json::Value> = [("k2".to_string(), json!(2))].into();
        assert!(
            matches!(check(&gate, &cycle(plain), &live).await, GateOutcome::Allow),
            "k2 is already in live state, so a1 can run"
        );

        let state_write = json!({ "id": "a0", "type": "state_write",
                                  "parameters": { "key": "k2", "value": 0 } });
        assert!(
            matches!(
                check(&gate, &cycle(state_write), &HashMap::new()).await,
                GateOutcome::Allow
            ),
            "a state_write of k2 is a declared writer outside the cycle"
        );
    }

    /// The issue's `[a0 tool_call, a1, a2]` three ways. Which one the gate
    /// refuses turns on whether a tool call finishes before a1 is validated.
    #[tokio::test]
    async fn a_tool_call_breaks_a_cycle_only_if_it_finishes_first() {
        let gate = StaticVerificationGate::new(registry(&["step"]));
        let shape = |failure: &str, mode: &str| {
            proposal_of(json!([
                { "id": "a0", "type": "tool_call", "tool": "step", "parameters": {},
                  "failure_behavior": failure, "invocation_mode": mode },
                { "id": "a1", "type": "tool_call", "tool": "step", "parameters": {},
                  "failure_behavior": failure,
                  "state_dependencies": ["k2"], "expected_effects": { "k1": 1 } },
                { "id": "a2", "type": "tool_call", "tool": "step", "parameters": {},
                  "failure_behavior": failure,
                  "state_dependencies": ["k1"], "expected_effects": { "k2": 2 } },
            ]))
        };

        // Default `abort`: the level runs sequentially, a0's handler finishes
        // before a1 is validated and may set k2 (`state.set`, FFI
        // `state_set`), so the plan may complete.
        assert!(
            matches!(
                check(&gate, &shape("abort", "one_shot"), &HashMap::new()).await,
                GateOutcome::Allow
            ),
            "a sequential tool call before the cycle may break it"
        );

        // No `abort` in the level: a0 and a1 run concurrently, so a0 could set
        // k2 in time only by winning a race. a2 comes after a0 and is freed;
        // a1 is certain to be refused.
        match check(&gate, &shape("skip", "one_shot"), &HashMap::new()).await {
            GateOutcome::Reject { blocked, reason } => {
                assert_eq!(blocked, ["a1".to_string()].into());
                assert!(reason.contains("'k2'"), "{reason}");
            }
            other => panic!("expected Reject, got {other:?}"),
        }

        // A detached a0 returns a handle before its handler is done, so it is
        // not guaranteed to have written anything when a1 is validated.
        assert!(
            matches!(
                check(&gate, &shape("abort", "long_running"), &HashMap::new()).await,
                GateOutcome::Reject { .. }
            ),
            "a detached tool call does not finish before the next action"
        );
    }

    /// Only a precondition that fails when its key is absent makes an action
    /// wait on that key: `not_exists` passes on an absent key.
    #[tokio::test]
    async fn not_exists_precondition_does_not_wait_on_its_key() {
        let gate = StaticVerificationGate::new(registry(&["step"]));
        let proposal = proposal_of(json!([
            { "id": "a1", "type": "tool_call", "tool": "step", "parameters": {},
              "preconditions": [{ "key": "k2", "operator": "not_exists" }],
              "expected_effects": { "k1": 1 } },
            { "id": "a2", "type": "tool_call", "tool": "step", "parameters": {},
              "state_dependencies": ["k1"], "expected_effects": { "k2": 2 } },
        ]));
        assert!(
            matches!(
                check(&gate, &proposal, &HashMap::new()).await,
                GateOutcome::Allow
            ),
            "a1 can run first, so this is a chain, not a cycle"
        );
    }

    /// A key nothing declares writing, with no tool call before the action that
    /// needs it, could be set only from outside the proposal: refused. With a
    /// tool call finishing first, whose handler may set it, it stays advisory.
    #[tokio::test]
    async fn undeclared_key_is_refused_unless_a_tool_call_finishes_first() {
        let gate = StaticVerificationGate::new(registry(&["step"]));
        let waiting = json!({ "id": "a1", "type": "tool_call", "tool": "step", "parameters": {},
                              "state_dependencies": ["written_but_undeclared"] });

        match check(
            &gate,
            &proposal_of(json!([waiting.clone()])),
            &HashMap::new(),
        )
        .await
        {
            GateOutcome::Reject { blocked, reason } => {
                assert_eq!(blocked, ["a1".to_string()].into());
                assert!(reason.contains("no action declares writing it"), "{reason}");
            }
            other => panic!("expected Reject, got {other:?}"),
        }

        let tool_first = proposal_of(json!([
            { "id": "a0", "type": "tool_call", "tool": "step", "parameters": {} },
            waiting,
        ]));
        assert!(matches!(
            check(&gate, &tool_first, &HashMap::new()).await,
            GateOutcome::Allow
        ));
    }

    /// The replan quality gate calls `blocking_errors` too, so the cycle refusal
    /// reaches both verification points.
    #[test]
    fn blocking_errors_reports_the_cycle_for_the_replan_path() {
        let proposal = proposal_of(json!([
            { "id": "a1", "type": "tool_call", "tool": "step", "parameters": {},
              "state_dependencies": ["k2"], "expected_effects": { "k1": 1 } },
            { "id": "a2", "type": "tool_call", "tool": "step", "parameters": {},
              "state_dependencies": ["k1"], "expected_effects": { "k2": 2 } },
        ]));
        let errors = blocking_errors(&proposal, Some(&HashMap::new()), None, &HashMap::new(), 100);
        let ids: HashSet<&str> = errors.iter().map(|i| i.action_id.as_str()).collect();
        assert_eq!(ids, ["a1", "a2"].into());
    }

    /// Under a tenant, a tool call's declared effect frees the key in both
    /// namespaces, because its handler can write either (callback
    /// `state_changes` land in the tenant's, a `state.set` in the shared one):
    /// a detached tool call declaring `k` frees a later tenant `state_read` of
    /// `k` and a later tool call needing the shared `k`. A tenant
    /// `state_write` of `k` frees only the tenant's `k` (car#3121).
    #[test]
    fn tenant_keys_resolve_per_action_namespace() {
        let tenant_errors = |actions: serde_json::Value| {
            blocking_errors(
                &proposal_of(actions),
                Some(&HashMap::new()),
                Some("acme"),
                &HashMap::new(),
                100,
            )
            .into_iter()
            .map(|i| i.action_id)
            .collect::<HashSet<String>>()
        };
        let detached_writer = json!({ "id": "t0", "type": "tool_call", "tool": "step",
            "parameters": {}, "invocation_mode": "long_running", "expected_effects": { "k": 1 } });
        let tenant_read = json!({ "id": "r1", "type": "state_read",
            "parameters": { "key": "k" }, "state_dependencies": ["k"] });
        let tool_need = json!({ "id": "t1", "type": "tool_call", "tool": "step",
            "parameters": {}, "state_dependencies": ["k"] });
        let tenant_write = json!({ "id": "w0", "type": "state_write",
            "parameters": { "key": "k", "value": 1 } });

        assert!(tenant_errors(json!([detached_writer, tenant_read])).is_empty());
        assert!(tenant_errors(json!([detached_writer, tool_need])).is_empty());
        assert!(tenant_errors(json!([tenant_write, tenant_read])).is_empty());
        assert_eq!(
            tenant_errors(json!([tenant_write, tool_need])),
            ["t1".to_string()].into()
        );
    }

    /// Warnings never block — a write conflict is reported at `warning`
    /// severity and `valid` stays true, so the plan still runs.
    #[tokio::test]
    async fn warnings_do_not_block() {
        let gate = StaticVerificationGate::new(registry(&["w"]));
        let proposal = proposal_of(json!([
            { "id": "a1", "type": "tool_call", "tool": "w", "parameters": {},
              "expected_effects": { "k": 1 } },
            { "id": "a2", "type": "tool_call", "tool": "w", "parameters": {},
              "expected_effects": { "k": 2 } },
        ]));
        assert!(
            matches!(
                check(&gate, &proposal, &HashMap::new()).await,
                GateOutcome::Allow
            ),
            "a write conflict is a warning, not a refusal"
        );
    }
}