car-engine 0.47.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. Those are exact — they
//! cannot be wrong about a plan that would in fact have run.
//!
//! Everything state-dependent 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. Dependency cycles cannot occur at all (`car-ir`'s
//! DAG edges only point to lower indices).

use crate::admission::{AdmissionGate, GateContext, GateOutcome};
use car_ir::{ActionProposal, 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 verification issue is exact enough to refuse a whole proposal on.
///
/// Blocking 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.
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.
pub fn blocking_errors(
    proposal: &ActionProposal,
    state: Option<&HashMap<String, serde_json::Value>>,
    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)
    };
    if result.valid {
        return Vec::new();
    }
    result
        .issues
        .into_iter()
        .filter(is_blocking_issue)
        .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;
            blocking_errors(proposal, Some(ctx.state), &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"
        );
    }

    /// 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"
        );
    }
}