rego_engine/

engine.rs

use crate::eval_context::{EvaluationContext, EvaluationScope, ReferenceRendering, ReferenceRenderingScope};
use crate::policies;
use data_source::embedded;
use data_source::types::KnownResourceTypes;
use diagnostics::{Diagnostic, PhaseMetric, phase_metric};
use log::{debug, info};
use rules::{RuleInfo, RuleMetadataEntry, RuleOrigin, build_rule_metadata_map};
use schema_validator::{
    OverlayCatalog, SchemaMetadataCatalog, SchemaValidator, getatt_data_with_overlays,
    schema_metadata_catalog_with_overlays,
};
use std::collections::{HashMap, HashSet};
use std::str::from_utf8;
use std::sync::{Arc, LazyLock, Mutex};
use template_model::SemanticModel;
use validation_engine::{
    EngineConfig, GuardRuleSet, ValidateConfig, ValidationEngine, ValidationError, build_rule_list,
    extract_diagnostics_from_value, semantic_model_to_input_json,
};

static REGORUS_DATA: LazyLock<Vec<(&str, &[u8])>> = LazyLock::new(|| {
    vec![
        (KNOWN_RESOURCE_TYPES_PATH, &*embedded::KNOWN_RESOURCE_TYPES_BYTES),
        (PRIMARY_IDENTIFIERS_PATH, &*embedded::PRIMARY_IDENTIFIERS_BYTES),
        ("data/iam_action_resource_patterns", &*embedded::IAM_ACTION_RESOURCE_PATTERNS_BYTES),
        ("data/stateful_resource_types", &*embedded::STATEFUL_RESOURCE_TYPES_BYTES),
        ("data/aws_rds_dbinstance_dbinstanceclass_enum", &*embedded::AWS_RDS_DBINSTANCE_DBINSTANCECLASS_ENUM_BYTES),
        ("data/aws_ec2_instance_instancetype_enum", &*embedded::AWS_EC2_INSTANCE_INSTANCETYPE_ENUM_BYTES),
        (
            "data/aws_emr_cluster_instancetypeconfig_instancetype_enum",
            &*embedded::AWS_EMR_CLUSTER_INSTANCETYPECONFIG_INSTANCETYPE_ENUM_BYTES,
        ),
        ("data/aws_gamelift_fleet_ec2instancetype_enum", &*embedded::AWS_GAMELIFT_FLEET_EC2INSTANCETYPE_ENUM_BYTES),
        ("data/aws_appstream_fleet_instancetype_enum", &*embedded::AWS_APPSTREAM_FLEET_INSTANCETYPE_ENUM_BYTES),
        ("data/aws_dax_cluster_nodetype_enum", &*embedded::AWS_DAX_CLUSTER_NODETYPE_ENUM_BYTES),
        ("data/aws_docdb_dbinstance_dbinstanceclass_enum", &*embedded::AWS_DOCDB_DBINSTANCE_DBINSTANCECLASS_ENUM_BYTES),
        (
            "data/aws_elasticache_cachecluster_cachenodetype_enum",
            &*embedded::AWS_ELASTICACHE_CACHECLUSTER_CACHENODETYPE_ENUM_BYTES,
        ),
        (
            "data/aws_managedblockchain_node_nodeconfiguration_instancetype_enum",
            &*embedded::AWS_MANAGEDBLOCKCHAIN_NODE_NODECONFIGURATION_INSTANCETYPE_ENUM_BYTES,
        ),
        (
            "data/aws_neptune_dbinstance_dbinstanceclass_enum",
            &*embedded::AWS_NEPTUNE_DBINSTANCE_DBINSTANCECLASS_ENUM_BYTES,
        ),
        (
            "data/aws_rds_dbcluster_dbclusterinstanceclass_enum",
            &*embedded::AWS_RDS_DBCLUSTER_DBCLUSTERINSTANCECLASS_ENUM_BYTES,
        ),
        ("data/aws_rds_dbinstance_db_instance_class", &*embedded::AWS_RDS_DBINSTANCE_DB_INSTANCE_CLASS_BYTES),
        ("data/aws_redshift_cluster_nodetype_enum", &*embedded::AWS_REDSHIFT_CLUSTER_NODETYPE_ENUM_BYTES),
        ("data/aws_amazonmq_broker_instancetype_enum", &*embedded::AWS_AMAZONMQ_BROKER_INSTANCETYPE_ENUM_BYTES),
        (
            "data/aws_sagemaker_processing_instancetype_enum",
            &*embedded::AWS_SAGEMAKER_PROCESSING_INSTANCETYPE_ENUM_BYTES,
        ),
        ("data/aws_sagemaker_hosting_instancetype_enum", &*embedded::AWS_SAGEMAKER_HOSTING_INSTANCETYPE_ENUM_BYTES),
        ("data/aws_sagemaker_transform_instancetype_enum", &*embedded::AWS_SAGEMAKER_TRANSFORM_INSTANCETYPE_ENUM_BYTES),
        ("data/aws_sagemaker_cluster_instancetype_enum", &*embedded::AWS_SAGEMAKER_CLUSTER_INSTANCETYPE_ENUM_BYTES),
        (
            "data/aws_elasticsearch_domain_elasticsearchclusterconfig_instancetype_enum",
            &*embedded::AWS_ELASTICSEARCH_DOMAIN_ELASTICSEARCHCLUSTERCONFIG_INSTANCETYPE_ENUM_BYTES,
        ),
        (
            "data/aws_opensearchservice_domain_clusterconfig_instancetype_enum",
            &*embedded::AWS_OPENSEARCHSERVICE_DOMAIN_CLUSTERCONFIG_INSTANCETYPE_ENUM_BYTES,
        ),
        (GETATT_ATTRIBUTES_PATH, &*embedded::GETATT_ATTRIBUTES_BYTES),
        ("data/codepipeline_action_artifact_counts", &*embedded::CODEPIPELINE_ACTION_ARTIFACT_COUNTS_BYTES),
        ("data/deprecated_resource_types", &*embedded::DEPRECATED_RESOURCE_TYPES_BYTES),
        ("data/retention_period_requirements", &*embedded::RETENTION_PERIOD_REQUIREMENTS_BYTES),
        ("data/sensitive_ports", &*embedded::SENSITIVE_PORTS_BYTES),
        ("data/secretsmanager_arn_fields", &*embedded::SECRETSMANAGER_ARN_FIELDS_BYTES),
        ("data/rule_data", &*embedded::RULE_DATA_BYTES),
        ("data/rule_tables", &*embedded::RULE_TABLES_BYTES),
    ]
});

/// The built-in rule packages, one query each. A package groups rules by
/// implementation area; the registry, not the package, decides a rule's category.
const CORE_PACKAGES: &[&str] = &[
    "data.structure.violation",
    "data.intrinsics.violation",
    "data.references.violation",
    "data.best_practices.violation",
    "data.resources.violation",
];

/// The [`REGORUS_DATA`] entry holding the catalog of resource types the rules
/// treat as existing.
const KNOWN_RESOURCE_TYPES_PATH: &str = "data/known_resource_types";

/// The [`REGORUS_DATA`] entry holding GetAtt attributes and attribute types.
const GETATT_ATTRIBUTES_PATH: &str = "data/getatt_attributes";

/// The [`REGORUS_DATA`] entry holding primary identifiers per type.
const PRIMARY_IDENTIFIERS_PATH: &str = "data/primary_identifiers";

/// Re-serializes the known-resource-type catalog with `extra_types` appended, or
/// returns `None` when there is nothing to add so the embedded bytes are used
/// verbatim.
fn extend_known_resource_types(extra_types: &[String]) -> anyhow::Result<Option<String>> {
    if extra_types.is_empty() {
        return Ok(None);
    }
    let mut catalog: KnownResourceTypes = serde_json::from_slice(&embedded::KNOWN_RESOURCE_TYPES_BYTES)
        .map_err(|e| anyhow::anyhow!("Failed to parse the embedded known_resource_types data: {e}"))?;
    anyhow::ensure!(!catalog.known_resource_types.is_empty(), "Embedded known_resource_types data must not be empty");
    for type_name in extra_types {
        if !catalog.known_resource_types.contains(type_name) {
            catalog.known_resource_types.push(type_name.clone());
        }
    }
    Ok(Some(serde_json::to_string(&catalog)?))
}

/// Extends the embedded getatt_attributes data with overlay catalog entries.
/// Returns `None` when there is nothing to add.
fn extend_getatt_data(catalog: &OverlayCatalog) -> anyhow::Result<Option<String>> {
    if catalog.getatt_attributes.is_empty() && catalog.getatt_attribute_types.is_empty() {
        return Ok(None);
    }
    let mut data: serde_json::Value = serde_json::from_slice(&embedded::GETATT_ATTRIBUTES_BYTES)
        .map_err(|e| anyhow::anyhow!("Failed to parse the embedded getatt_attributes data: {e}"))?;

    // Merge getatt_attributes (sort/dedup after merging)
    if let Some(attrs_obj) = data.get_mut("getatt_attributes").and_then(|v| v.as_object_mut()) {
        for (type_name, attrs) in &catalog.getatt_attributes {
            let entry = attrs_obj.entry(type_name.clone()).or_insert_with(|| serde_json::Value::Array(Vec::new()));
            if let Some(arr) = entry.as_array_mut() {
                for attr in attrs {
                    let val = serde_json::Value::String(attr.clone());
                    if !arr.contains(&val) {
                        arr.push(val);
                    }
                }
                arr.sort_by(|a, b| a.as_str().unwrap_or("").cmp(b.as_str().unwrap_or("")));
                arr.dedup();
            }
        }
    }

    // Merge getatt_attribute_types
    if let Some(types_obj) = data.get_mut("getatt_attribute_types").and_then(|v| v.as_object_mut()) {
        for (type_name, attr_types) in &catalog.getatt_attribute_types {
            let entry =
                types_obj.entry(type_name.clone()).or_insert_with(|| serde_json::Value::Object(serde_json::Map::new()));
            if let Some(obj) = entry.as_object_mut() {
                for (attr, atype) in attr_types {
                    obj.insert(attr.clone(), serde_json::Value::String(atype.clone()));
                }
            }
        }
    }

    Ok(Some(serde_json::to_string(&data)?))
}

/// Extends the embedded primary_identifiers data with overlay catalog entries.
/// Returns `None` when there is nothing to add.
fn extend_primary_identifiers_data(catalog: &OverlayCatalog) -> anyhow::Result<Option<String>> {
    if catalog.primary_identifiers.is_empty() {
        return Ok(None);
    }
    let mut data: serde_json::Value = serde_json::from_slice(&embedded::PRIMARY_IDENTIFIERS_BYTES)
        .map_err(|e| anyhow::anyhow!("Failed to parse the embedded primary_identifiers data: {e}"))?;

    if let Some(pids_obj) = data.get_mut("primary_identifiers").and_then(|v| v.as_object_mut()) {
        for (type_name, pids) in &catalog.primary_identifiers {
            pids_obj.insert(type_name.clone(), serde_json::json!(pids));
        }
    }

    Ok(Some(serde_json::to_string(&data)?))
}

/// Whether an engine instance loads and evaluates the handwritten built-in
/// policies, or serves only caller-supplied external rules.
///
/// In [`BuiltinRuleMode::ExternalOnly`] the built-in policies are neither loaded
/// nor advertised through registry metadata, so the engine evaluates only custom
/// and Guard rules. Embedded data tables, custom builtins, and external-rule
/// metadata are retained in both modes.
#[derive(Clone, Copy, PartialEq, Eq)]
enum BuiltinRuleMode {
    Enabled,
    ExternalOnly,
}

/// Who authored a Rego package, which fixes the contract its rules were written
/// against: how its diagnostics are attributed and how the resolution builtins
/// render a reference while it evaluates.
#[derive(Clone, Copy)]
enum PolicyPackageKind {
    /// A handwritten built-in policy shipped with the engine.
    BuiltIn,
    /// A caller-supplied custom rule.
    Custom,
}

impl PolicyPackageKind {
    fn source_label(self) -> &'static str {
        match self {
            Self::BuiltIn => "Core",
            Self::Custom => "Custom",
        }
    }

    /// The origin stamped on the package's diagnostics; a built-in rule keeps the
    /// origin the registry records for it.
    fn rule_origin_override(self) -> Option<&'static RuleOrigin> {
        match self {
            Self::BuiltIn => None,
            Self::Custom => Some(&RuleOrigin::Custom),
        }
    }

    /// Built-in policies validate literal content and must never see a logical ID
    /// where a value belongs. Custom rules keep the logical-ID string they were
    /// written against, so upgrading the engine does not change what they report.
    fn reference_rendering(self) -> ReferenceRendering {
        match self {
            Self::BuiltIn => ReferenceRendering::Marker,
            Self::Custom => ReferenceRendering::TargetId,
        }
    }
}

pub struct RegoEngine {
    base_rego: regorus::Engine,
    /// Whether the handwritten built-in policies are evaluated, or only the
    /// caller-supplied external rules.
    builtin_mode: BuiltinRuleMode,
    /// Built-in rule metadata from the rules registry only.
    registry_metadata: HashMap<String, RuleMetadataEntry>,
    /// Custom rego rule metadata discovered from evaluation output.
    /// Rego rules embed metadata in their output objects, so it can only
    /// be extracted after evaluation. May be incomplete if not all rules fired.
    discovered_custom_metadata: Mutex<HashMap<String, RuleMetadataEntry>>,
    custom_packages: Vec<String>,
    /// Guard rules are evaluated by the shared Guard evaluator, not by Rego, so
    /// every engine reports the same Guard findings.
    guard_rules: Option<GuardRuleSet>,
    init_metric: PhaseMetric,
}

impl RegoEngine {
    pub fn new(config: EngineConfig) -> anyhow::Result<Self> {
        let overlay_catalog =
            config.build_overlay_catalog().map_err(|e| anyhow::anyhow!("Failed to build overlay catalog: {e}"))?;
        let start = web_time::Instant::now();
        let schema_metadata = schema_metadata_catalog_with_overlays(&overlay_catalog)?;
        Self::new_from_parts(config, &overlay_catalog, schema_metadata, BuiltinRuleMode::Enabled, start)
    }

    /// Constructs the engine reusing metadata from an already-built
    /// [`SchemaValidator`](schema_validator::SchemaValidator). The validator's
    /// overlay catalog is treated as authoritative - the engine does not
    /// re-resolve overlay schemas.
    ///
    /// This entry point is intended for language bindings and the CLI, which
    /// construct a `SchemaValidator` once and share it with the engine. The
    /// validator's shared schema-metadata catalog is reused rather than rebuilt,
    /// so every engine built from the same validator shares one catalog rather
    /// than rebuilding it.
    #[doc(hidden)]
    pub fn new_with_schema_validator(config: EngineConfig, validator: &SchemaValidator) -> anyhow::Result<Self> {
        let start = web_time::Instant::now();
        let schema_metadata = validator.schema_metadata_catalog()?;
        Self::new_from_parts(config, validator.overlay_catalog(), schema_metadata, BuiltinRuleMode::Enabled, start)
    }

    /// Constructs an engine that evaluates only the caller-supplied custom and
    /// Guard rules, without loading or advertising the handwritten built-in
    /// policies. Embedded data tables and custom builtins remain available so
    /// external rules can call them.
    ///
    /// This backs the composite engine, whose built-in rules are owned by a
    /// separate engine; loading them here too would double-evaluate them.
    pub fn new_external_only(config: EngineConfig) -> anyhow::Result<Self> {
        let overlay_catalog =
            config.build_overlay_catalog().map_err(|e| anyhow::anyhow!("Failed to build overlay catalog: {e}"))?;
        let start = web_time::Instant::now();
        let schema_metadata = schema_metadata_catalog_with_overlays(&overlay_catalog)?;
        Self::new_from_parts(config, &overlay_catalog, schema_metadata, BuiltinRuleMode::ExternalOnly, start)
    }

    /// External-only counterpart to [`RegoEngine::new_with_schema_validator`],
    /// reusing an already-built validator's overlay catalog and shared
    /// schema-metadata catalog.
    #[doc(hidden)]
    pub fn new_external_only_with_schema_validator(
        config: EngineConfig,
        validator: &SchemaValidator,
    ) -> anyhow::Result<Self> {
        let start = web_time::Instant::now();
        let schema_metadata = validator.schema_metadata_catalog()?;
        Self::new_from_parts(config, validator.overlay_catalog(), schema_metadata, BuiltinRuleMode::ExternalOnly, start)
    }

    /// Internal constructor that accepts a pre-built overlay catalog and the
    /// shared schema-metadata catalog resolved for it.
    fn new_from_parts(
        config: EngineConfig,
        overlay_catalog: &OverlayCatalog,
        schema_metadata: Arc<SchemaMetadataCatalog>,
        builtin_mode: BuiltinRuleMode,
        start: web_time::Instant,
    ) -> anyhow::Result<Self> {
        let mut rego = regorus::Engine::new();
        rego.set_strict_builtin_errors(false);

        // Single-pass merge avoids per-file JSON parsing overhead.
        {
            // Resource types introduced by an overlay schema are legitimate
            // targets, so the type catalog the rules consult must include them
            // rather than reporting them as nonexistent.
            let overlay_types: Vec<String> = overlay_catalog.type_names.clone();
            let extended_known_types = extend_known_resource_types(&overlay_types)?;

            // Extend getatt_attributes data with overlay entries
            let extended_getatt = extend_getatt_data(overlay_catalog)?;
            // Extend primary_identifiers data with overlay entries
            let extended_primary_ids = extend_primary_identifiers_data(overlay_catalog)?;

            // Embedded lengths closely size the common no-overlay path; extended
            // overlay documents can grow the buffer normally when they are larger.
            let merged_capacity: usize = REGORUS_DATA.iter().map(|(_, json_bytes)| json_bytes.len()).sum();
            let mut merged = String::with_capacity(merged_capacity);
            merged.push('{');
            for (i, (path, json_bytes)) in REGORUS_DATA.iter().enumerate() {
                let json_str = match (
                    *path,
                    extended_known_types.as_deref(),
                    extended_getatt.as_deref(),
                    extended_primary_ids.as_deref(),
                ) {
                    (KNOWN_RESOURCE_TYPES_PATH, Some(extended), _, _) => extended,
                    (GETATT_ATTRIBUTES_PATH, _, Some(extended), _) => extended,
                    (PRIMARY_IDENTIFIERS_PATH, _, _, Some(extended)) => extended,
                    _ => from_utf8(json_bytes).map_err(|error| {
                        anyhow::anyhow!("Embedded JSON data '{}' is not valid UTF-8: {}", path, error)
                    })?,
                };
                let inner = json_str
                    .trim()
                    .strip_prefix('{')
                    .and_then(|value| value.strip_suffix('}'))
                    .ok_or_else(|| anyhow::anyhow!("Embedded JSON data '{}' must be a top-level object", path))?;
                anyhow::ensure!(!inner.trim().is_empty(), "Embedded JSON data '{}' must not be empty", path);
                if i > 0 {
                    merged.push(',');
                }
                merged.push_str(inner);
            }
            merged.push('}');
            rego.add_data(regorus::Value::from_json_str(&merged)?)?;
        }

        // In external-only mode the built-in policies are owned by a separate
        // engine, so they are neither loaded nor evaluated here. Embedded data,
        // custom builtins, custom Rego, and Guard rules are still loaded below
        // regardless of mode.
        let handwritten_rule_count = match builtin_mode {
            BuiltinRuleMode::Enabled => {
                for (path, source) in policies::HANDWRITTEN_REGO_POLICIES {
                    rego.add_policy(path.to_string(), source.to_string())?;
                }
                policies::HANDWRITTEN_REGO_POLICIES.len()
            }
            BuiltinRuleMode::ExternalOnly => 0,
        };
        debug!("Loaded {} data files, {} handwritten rules", REGORUS_DATA.len(), handwritten_rule_count);

        let guard_rules = GuardRuleSet::compile(&config.guard_rules).map_err(|error| anyhow::anyhow!(error))?;

        for entry in &config.custom_rules {
            rego.add_policy(entry.name.clone(), entry.content.clone())?;
        }
        let mut custom_packages = Vec::new();
        for entry in &config.custom_rules {
            for line in entry.content.lines() {
                let trimmed = line.trim();
                if let Some(pkg) = trimmed.strip_prefix("package ") {
                    let eval_path = format!("data.{}.violation", pkg.trim());
                    if !custom_packages.contains(&eval_path) {
                        custom_packages.push(eval_path);
                    }
                }
            }
        }

        // The GetAtt return-type table is the process-wide one the schema store
        // shares, layered with this engine's overlays; resolving it here keeps the
        // parse off the first template without giving the engine its own copy.
        let getatt = getatt_data_with_overlays(overlay_catalog)?;
        crate::builtins::register_all(&mut rego, schema_metadata, getatt)?;

        // External-only construction advertises no built-in rules; their
        // metadata belongs to the engine that owns the built-in policies.
        let registry_metadata = match builtin_mode {
            BuiltinRuleMode::Enabled => build_rule_metadata_map(),
            BuiltinRuleMode::ExternalOnly => HashMap::new(),
        };

        // Every evaluation clones `rego`, and a clone inherits the prepared state
        // (analysis, scheduling, and loop hoisting of every loaded policy). Preparing
        // once here keeps that cost out of every evaluation; without it an engine
        // that loads only external rules would re-analyze the whole custom rule set
        // for each template. A trivial query prepares the engine without evaluating
        // any rule body, so it applies equally to both built-in modes. A preparation
        // failure is not swallowed: the first real evaluation reports it.
        let _ = rego.eval_query("true".to_string(), false);

        info!(
            "RegoEngine initialized: {} handwritten rules, {} data files, {} registry + {} Guard metadata entries",
            handwritten_rule_count,
            REGORUS_DATA.len(),
            registry_metadata.len(),
            guard_rules.as_ref().map_or(0, |guard| guard.rule_metadata().len())
        );
        let init_metric = phase_metric(start);
        Ok(RegoEngine {
            base_rego: rego,
            builtin_mode,
            registry_metadata,
            discovered_custom_metadata: Mutex::new(HashMap::new()),
            custom_packages,
            guard_rules,
            init_metric,
        })
    }

    /// Evaluates a single Rego package and appends its diagnostics to `out`.
    ///
    /// The package's [`ReferenceRendering`] is in effect only while its query
    /// runs: Regorus discards memoized rule and builtin results between queries,
    /// so a value rendered for one package never reaches the next.
    ///
    /// Any evaluation or serialization failure is returned as a structured
    /// [`ValidationError`] - an exception the caller can handle - and is never
    /// converted into a diagnostic. A rule that fails to run must surface as an
    /// error, not masquerade as a finding.
    fn eval_package_into(
        &self,
        rego: &mut regorus::Engine,
        package: &str,
        kind: PolicyPackageKind,
        model: &SemanticModel,
        out: &mut Vec<Diagnostic>,
    ) -> Result<(), ValidationError> {
        let source_label = kind.source_label();
        let _rendering = ReferenceRenderingScope::enter(kind.reference_rendering());
        let value = rego.eval_rule(package.to_string()).map_err(|e| {
            ValidationError::Engine(format!("{source_label} rule package '{package}' failed to evaluate: {e}"))
        })?;
        let diagnostics_json = serde_json::to_value(&value).map_err(|e| {
            ValidationError::Engine(format!(
                "{source_label} rule package '{package}' produced a result that could not be \
                 serialized to JSON: {e}"
            ))
        })?;
        extract_diagnostics_from_value(&diagnostics_json, model, out, kind.rule_origin_override())
            .map_err(ValidationError::from)
    }

    /// The built-in rule IDs that global filtering proves cannot survive under
    /// `config`, so their handwritten clauses can stop at the `cfn_rule_active`
    /// guard before doing any work. Only rules the registry defines are eligible;
    /// custom and guard rules are never globally suppressed here.
    fn globally_suppressed_builtin_rules(&self, config: &ValidateConfig) -> Arc<HashSet<String>> {
        let suppressed: HashSet<String> = self
            .registry_metadata
            .iter()
            .filter(|(rule_id, entry)| {
                config.filters.globally_suppresses_builtin_rule(
                    rule_id,
                    entry.category.as_deref(),
                    entry.severity,
                    config.strict,
                    config.severity_level,
                )
            })
            .map(|(rule_id, _)| rule_id.clone())
            .collect();
        Arc::new(suppressed)
    }
}

impl ValidationEngine for RegoEngine {
    fn engine_name(&self) -> &str {
        "rego"
    }

    fn evaluate_rules(
        &self,
        model: &Arc<SemanticModel>,
        config: &ValidateConfig,
    ) -> Result<Vec<Diagnostic>, ValidationError> {
        let context = EvaluationContext::new(
            model.clone(),
            config.pseudo_parameter_overrides.region.clone(),
            self.globally_suppressed_builtin_rules(config),
        );
        let _scope = EvaluationScope::enter(context);

        let mut rego = self.base_rego.clone();

        let input_json = semantic_model_to_input_json(model)?;
        let input_value = crate::builtins::serde_json_to_rego_value(&input_json);
        rego.set_input(input_value);

        let mut diagnostics = Vec::new();

        // Core built-in policies run only when this engine owns them and the
        // caller has not disabled built-ins. The custom and Guard evaluation
        // below runs unconditionally so external rules are always evaluated.
        let evaluate_builtins = matches!(self.builtin_mode, BuiltinRuleMode::Enabled) && !config.disable_builtin_rules;
        if evaluate_builtins {
            // Every package is evaluated whatever the category filters say. The
            // packages group rules by implementation area, not by registry
            // category - security and parameter rules live in `best_practices`
            // and `structure` - so skipping a package for one excluded category
            // would drop its rules of every other category. Each clause instead
            // stops at its `cfn_rule_active` guard, which the evaluation context
            // above feeds from the registry category of every rule, so an
            // excluded rule costs one builtin call. The packages share no rules,
            // so one query per package costs the same as one aggregated query.
            for package in CORE_PACKAGES {
                self.eval_package_into(&mut rego, package, PolicyPackageKind::BuiltIn, model, &mut diagnostics)?;
            }
        }

        for pkg in &self.custom_packages {
            self.eval_package_into(&mut rego, pkg, PolicyPackageKind::Custom, model, &mut diagnostics)?;
        }

        if let Some(guard_rules) = &self.guard_rules {
            diagnostics.extend(guard_rules.evaluate(model)?);
        }

        if !self.custom_packages.is_empty() {
            let mut discovered = self.discovered_custom_metadata.lock().unwrap_or_else(|e| e.into_inner());
            for d in &diagnostics {
                if d.source == RuleOrigin::Custom {
                    discovered.entry(d.rule_id.clone()).or_insert_with(|| RuleMetadataEntry {
                        category: d.category.clone(),
                        description: d.message.clone(),
                        severity: d.severity,
                        origin: RuleOrigin::Custom,
                    });
                }
            }
        }

        Ok(diagnostics)
    }

    fn list_rules(&self) -> Vec<RuleInfo> {
        build_rule_list(&self.registry_metadata, &self.external_rule_metadata())
    }

    fn rule_metadata(&self) -> &HashMap<String, RuleMetadataEntry> {
        &self.registry_metadata
    }

    /// Returns Guard rule metadata merged with any custom Rego rule metadata
    /// discovered from prior evaluations. Custom diagnostics already carry their
    /// current evaluation's rule description, so this accumulated metadata cannot
    /// overwrite a later report.
    fn external_rule_metadata(&self) -> HashMap<String, RuleMetadataEntry> {
        let mut merged = self.guard_rules.as_ref().map(|guard| guard.rule_metadata().clone()).unwrap_or_default();
        if !self.custom_packages.is_empty() {
            let discovered = self.discovered_custom_metadata.lock().unwrap_or_else(|e| e.into_inner());
            merged.extend(discovered.iter().map(|(k, v)| (k.clone(), v.clone())));
        }
        merged
    }

    fn init_metric(&self) -> &PhaseMetric {
        &self.init_metric
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use diagnostics::DetailLevel;
    use rules::{Category, FilterConfig, RuleFilterConfig, Severity, lookup_rule};
    use std::sync::Barrier;
    use std::thread;
    use template_model::SemanticModel;
    use validation_engine::{EngineConfig, ExternalRuleSource, ValidateConfig, ValidationEngine, validate_bytes};

    fn make_engine() -> RegoEngine {
        RegoEngine::new(EngineConfig::default()).unwrap()
    }

    fn make_model_from_yaml(yaml: &str) -> Arc<SemanticModel> {
        let model = SemanticModel::from_bytes(yaml.as_bytes()).unwrap();
        Arc::new(model)
    }

    #[test]
    fn new_default_config_succeeds() {
        let engine = make_engine();
        assert_eq!(engine.engine_name(), "rego");
    }

    #[test]
    fn list_rules_returns_nonempty() {
        let engine = make_engine();
        let rules = engine.list_rules();
        assert!(!rules.is_empty(), "should have built-in rules");
    }

    #[test]
    fn rule_metadata_contains_entries() {
        let engine = make_engine();
        assert!(!engine.rule_metadata().is_empty());
    }

    #[test]
    fn init_metric_has_positive_duration() {
        let engine = make_engine();
        let metric = engine.init_metric();
        assert!(metric.duration_ms > 0.0, "init duration should be > 0, got {}", metric.duration_ms);
    }

    #[test]
    fn evaluate_minimal_template_no_errors() {
        let engine = make_engine();
        let model = make_model_from_yaml(
            r#"
AWSTemplateFormatVersion: "2010-09-09"
Resources:
  MyBucket:
    Type: AWS::S3::Bucket
"#,
        );
        let diags = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();
        assert!(
            diags.iter().all(|d| d.severity != Severity::Fatal),
            "rego engine should not produce Fatal diagnostics"
        );
    }

    #[test]
    fn evaluate_empty_resources_does_not_panic() {
        let engine = make_engine();
        let model = make_model_from_yaml(
            r#"
AWSTemplateFormatVersion: "2010-09-09"
Resources: {}
"#,
        );
        // CloudFormation rejects templates with empty Resources, so the empty-resources diagnostic is expected.
        let diags = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();
        assert!(
            diags.iter().any(|d| d.rule_id == "F0001"),
            "expected F0001 Fatal for empty Resources section, got: {:?}",
            diags.iter().map(|d| &d.rule_id).collect::<Vec<_>>(),
        );
    }

    /// Excluding a category must remove exactly the rules the registry files
    /// under it. The `best_practices` package also hosts security rules, so a
    /// package-level skip would wrongly drop them along with the best-practice
    /// rules; the template fires one of each from that package.
    #[test]
    fn evaluate_with_excluded_categories() {
        let engine = make_engine();
        let model = make_model_from_yaml(
            r#"
AWSTemplateFormatVersion: "2010-09-09"
Resources:
  Permission:
    Type: AWS::Lambda::Permission
    Properties:
      Action: lambda:InvokeFunction
      FunctionName: my-function
      Principal: sns.amazonaws.com
      SourceArn: arn:aws:sns:us-east-1:123456789012:my-topic
"#,
        );
        fn rule_ids(diags: &[Diagnostic]) -> HashSet<&str> {
            diags.iter().map(|d| d.rule_id.as_str()).collect()
        }
        let diags_all = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();
        let all_ids = rule_ids(&diags_all);
        assert!(all_ids.contains("W9002"), "sanity: the best-practice hardcoded-ARN rule fires: {all_ids:?}");
        assert!(all_ids.contains("W9013"), "sanity: the security hardcoded-account-ID rule fires: {all_ids:?}");

        let excluded = Category::BestPractice;
        let config = ValidateConfig {
            filters: FilterConfig::new(
                RuleFilterConfig::default(),
                RuleFilterConfig { categories: vec![excluded.as_str().to_string()], ..Default::default() },
            ),
            ..Default::default()
        };
        let diags_filtered = engine.evaluate_rules(&model, &config).unwrap();
        let filtered_ids = rule_ids(&diags_filtered);

        let expected_ids: HashSet<&str> =
            all_ids.iter().copied().filter(|id| lookup_rule(id).is_none_or(|rule| rule.category != excluded)).collect();
        assert_eq!(filtered_ids, expected_ids, "only rules the registry files under {excluded:?} may disappear");
        assert!(!filtered_ids.contains("W9002"), "the excluded best-practice rule is gone");
        assert!(filtered_ids.contains("W9013"), "the security rule hosted in the same package survives");
    }

    #[test]
    fn custom_rego_rule_produces_diagnostics() {
        let custom_rego = r#"
package custom_test
import rego.v1

violation contains v if {
    some name, res in input.resources
    res.resourceType == "AWS::S3::Bucket"
    v := {"rule_id": "CUSTOM001", "severity": "error", "message": "Custom rule fired", "resource_id": name}
}
"#;
        let config = EngineConfig {
            custom_rules: vec![ExternalRuleSource { name: "custom_test.rego".into(), content: custom_rego.into() }],
            guard_rules: vec![],
            ..Default::default()
        };
        let engine = RegoEngine::new(config).unwrap();
        let model = make_model_from_yaml(
            r#"
AWSTemplateFormatVersion: "2010-09-09"
Resources:
  MyBucket:
    Type: AWS::S3::Bucket
"#,
        );
        let diags = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();
        let custom = diags.iter().find(|d| d.rule_id == "CUSTOM001");
        assert!(custom.is_some(), "custom rule should fire");
    }

    #[test]
    fn guard_rule_produces_diagnostics() {
        let guard_source = r#"
rule check_bucket_name {
    AWS::S3::Bucket {
        Properties.BucketName EXISTS
        <<BucketName must be specified>>
    }
}
"#;
        let config = EngineConfig {
            custom_rules: vec![],
            guard_rules: vec![ExternalRuleSource { name: "test.guard".into(), content: guard_source.into() }],
            ..Default::default()
        };
        let engine = RegoEngine::new(config).unwrap();
        let model = make_model_from_yaml(
            r#"
AWSTemplateFormatVersion: "2010-09-09"
Resources:
  MyBucket:
    Type: AWS::S3::Bucket
"#,
        );
        let diags = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();
        let guard_diag = diags.iter().find(|d| d.rule_id == "check_bucket_name").expect("guard rule fires");
        assert_eq!(guard_diag.source, RuleOrigin::Guard);
        assert_eq!(guard_diag.category.as_deref(), Some("guard:test"));
        assert_eq!(guard_diag.message, "BucketName must be specified");
        assert_eq!(guard_diag.resource_logical_id(), Some("MyBucket"));
    }

    #[test]
    fn guard_rule_exists_check_does_not_fire_when_property_is_present() {
        let guard_source = r#"
rule check_bucket_name {
    AWS::S3::Bucket {
        Properties.BucketName EXISTS
        <<BucketName must be specified>>
    }
}
"#;
        let config = EngineConfig {
            custom_rules: vec![],
            guard_rules: vec![ExternalRuleSource { name: "test.guard".into(), content: guard_source.into() }],
            ..Default::default()
        };
        let engine = RegoEngine::new(config).unwrap();
        let model = make_model_from_yaml(
            r#"
AWSTemplateFormatVersion: "2010-09-09"
Resources:
  MyBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: my-bucket
"#,
        );
        let diags = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();
        let guard_diag = diags.iter().find(|d| d.rule_id == "check_bucket_name");
        assert!(
            guard_diag.is_none(),
            "guard EXISTS must be satisfied when BucketName is present, so the rule must not fire; got: {:?}",
            diags.iter().map(|d| &d.rule_id).collect::<Vec<_>>()
        );
    }

    #[test]
    fn external_only_engine_evaluates_external_rules_but_skips_builtins() {
        let custom_rego = r#"
package external_only_test
import rego.v1

violation contains v if {
    v := {"rule_id": "EXT_ONLY", "severity": "error", "message": "external rule fired", "resource_id": ""}
}
"#;
        let config = EngineConfig {
            custom_rules: vec![ExternalRuleSource {
                name: "external_only_test.rego".into(),
                content: custom_rego.into(),
            }],
            guard_rules: vec![],
            ..Default::default()
        };
        let engine = RegoEngine::new_external_only(config).unwrap();
        assert!(engine.rule_metadata().is_empty(), "an external-only engine must advertise no built-in registry rules");

        // An empty Resources section makes a normal engine emit a built-in
        // structural diagnostic, so its absence proves the built-ins were
        // skipped rather than simply not triggered.
        let model = make_model_from_yaml(
            r#"
AWSTemplateFormatVersion: "2010-09-09"
Resources: {}
"#,
        );
        let normal_diags = RegoEngine::new(EngineConfig::default())
            .unwrap()
            .evaluate_rules(&model, &ValidateConfig::default())
            .unwrap();
        assert!(
            normal_diags.iter().any(|d| d.rule_id == "F0001"),
            "a normal engine must emit the built-in F0001 for empty Resources"
        );

        let diags = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();
        let external = diags.iter().find(|d| d.rule_id == "EXT_ONLY").expect("the external rule must still evaluate");
        assert_eq!(external.severity, Severity::Error);
        assert_eq!(external.source, RuleOrigin::Custom);
        assert!(
            diags.iter().all(|d| d.rule_id == "EXT_ONLY"),
            "external-only evaluation must produce no built-in diagnostics, got: {:?}",
            diags.iter().map(|d| &d.rule_id).collect::<Vec<_>>()
        );
        assert!(
            !engine.list_rules().iter().any(|r| r.id == "F0001"),
            "external-only list_rules must not include built-in rule IDs"
        );
    }

    #[test]
    fn external_only_engine_registers_custom_builtin_helpers() {
        let custom_rego = r#"
package external_builtin_test
import rego.v1

violation contains make_diag("EXT_BUILTINS", "error", name, "custom builtins available") if {
    some name in resources_of_type("AWS::S3::Bucket")
    has_property(name, "BucketName")
    resolve(name, "Properties.BucketName") == "custom-builtins-bucket"
    "BucketName" in schema_properties("AWS::S3::Bucket")
}
"#;
        let engine = RegoEngine::new_external_only(EngineConfig {
            custom_rules: vec![ExternalRuleSource {
                name: "external_builtin_test.rego".into(),
                content: custom_rego.into(),
            }],
            ..Default::default()
        })
        .expect("external-only engine must initialize with custom builtins");
        let model = make_model_from_yaml(
            r#"
Resources:
  Bucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: custom-builtins-bucket
"#,
        );

        let diagnostics = engine.evaluate_rules(&model, &ValidateConfig::default()).expect("custom rule must evaluate");
        let finding = diagnostics
            .iter()
            .find(|diagnostic| diagnostic.rule_id == "EXT_BUILTINS")
            .expect("custom rule using resource, resolution, schema, and diagnostic helpers must fire");
        assert_eq!(finding.source, RuleOrigin::Custom);
        assert_eq!(finding.resource_logical_id(), Some("Bucket"));
    }

    #[test]
    fn evaluate_rules_can_be_called_repeatedly() {
        let engine = make_engine();
        let model = make_model_from_yaml(
            r#"
AWSTemplateFormatVersion: "2010-09-09"
Resources:
  A:
    Type: AWS::S3::Bucket
"#,
        );
        // The per-thread evaluation context is installed and torn down each call,
        // so repeated evaluations on one thread neither deadlock nor leak state.
        let _ = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();
        let diags = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();
        assert!(
            diags.iter().all(|d| d.severity != Severity::Fatal),
            "rego engine should not produce Fatal diagnostics"
        );
    }

    #[test]
    fn build_rule_metadata_map_returns_entries() {
        let index = build_rule_metadata_map();
        assert!(!index.is_empty());
    }

    const BUILTIN_TEST_TEMPLATE: &str = r#"
AWSTemplateFormatVersion: "2010-09-09"
Parameters:
  Env:
    Type: String
    AllowedValues:
      - prod
      - dev
  Port:
    Type: Number
Conditions:
  IsProd:
    Fn::Equals:
      - !Ref Env
      - prod
Mappings:
  RegionMap:
    us-east-1:
      AMI: ami-12345678
    us-west-2:
      AMI: ami-abcdefgh
Resources:
  MyBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: my-bucket
      Tags:
        - Key: Env
          Value: !If [IsProd, "production", "development"]
  MyFunc:
    Type: AWS::Lambda::Function
    DependsOn: MyBucket
    Condition: IsProd
    Properties:
      FunctionName: my-func
      Runtime: python3.12
      Handler: index.handler
      Code:
        S3Bucket: !Ref MyBucket
        S3Key: code.zip
      VpcConfig:
        SecurityGroupIds:
          - sg-0123456789abcdef0
        SubnetIds:
          - !Ref MySubnet
          - subnet-0123456789abcdef0
  MyVpc:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: 10.0.0.0/16
  MySubnet:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref MyVpc
      CidrBlock: 10.0.0.0/24
  MyQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: my-queue
Transform: AWS::Serverless-2016-10-31
"#;

    fn eval_builtin_policy(rego_source: &str, rule_id: &str) -> Vec<Diagnostic> {
        let config = EngineConfig {
            custom_rules: vec![ExternalRuleSource { name: "builtin_test.rego".into(), content: rego_source.into() }],
            guard_rules: vec![],
            ..Default::default()
        };
        let engine = RegoEngine::new(config).unwrap();
        let model = make_model_from_yaml(BUILTIN_TEST_TEMPLATE);
        let diags = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();
        diags.into_iter().filter(|d| d.rule_id == rule_id).collect()
    }

    #[test]
    fn builtin_resolve_returns_property_value() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    val := resolve("MyBucket", "BucketName")
    val == "my-bucket"
    v := {"rule_id": "B_RESOLVE", "severity": "error", "message": "ok", "resource_id": "MyBucket"}
}
"#,
            "B_RESOLVE",
        );
        assert_eq!(diags.len(), 1, "resolve should return 'my-bucket'");
    }

    #[test]
    fn builtin_resolve_all_returns_conditional_values() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    vals := resolve_all("MyBucket", "Tags.0.Value")
    count(vals) >= 1
    v := {"rule_id": "B_RESOLVE_ALL", "severity": "error", "message": "ok", "resource_id": "MyBucket"}
}
"#,
            "B_RESOLVE_ALL",
        );
        assert_eq!(diags.len(), 1, "resolve_all should return at least one value");
    }

    /// The contract custom rules were written against: `resolve` on a `Ref` or
    /// `Fn::GetAtt` yields the target's logical ID as a string, so a rule can look
    /// the target up in `input.resources` or compare it with another logical ID.
    const LEGACY_REFERENCE_CUSTOM_RULE: &str = r#"
package legacy_reference_test
import rego.v1

violation contains make_diag("LEGACY_LOOKUP", "error", name, sprintf("code bucket is %s", [target])) if {
    some name in resources_of_type("AWS::Lambda::Function")
    target := resolve(name, "Properties.Code.S3Bucket")
    is_string(target)
    input.resources[target].resourceType == "AWS::S3::Bucket"
}

violation contains make_diag("LEGACY_EQUALS", "error", name, "code lives in MyBucket") if {
    some name in resources_of_type("AWS::Lambda::Function")
    resolve(name, "Properties.Code.S3Bucket") == "MyBucket"
}

violation contains make_diag("LEGACY_MARKER_SEEN", "error", name, "resolve returned the marker object") if {
    some name in resources_of_type("AWS::Lambda::Function")
    is_object(resolve(name, "Properties.Code.S3Bucket"))
}
"#;

    fn assert_legacy_reference_contract(diags: &[Diagnostic]) {
        let lookup = diags
            .iter()
            .find(|d| d.rule_id == "LEGACY_LOOKUP")
            .expect("resolve() must hand back a logical ID a rule can look up in input.resources");
        assert_eq!(lookup.message, "code bucket is MyBucket");
        assert_eq!(lookup.resource_logical_id(), Some("MyFunc"));
        assert!(
            diags.iter().any(|d| d.rule_id == "LEGACY_EQUALS"),
            "resolve() must compare equal to the target's logical ID"
        );
        assert!(
            !diags.iter().any(|d| d.rule_id == "LEGACY_MARKER_SEEN"),
            "a custom rule must never see the marker object from resolve()"
        );
    }

    #[test]
    fn custom_rule_resolve_keeps_the_legacy_target_id_string_for_a_reference() {
        let engine = RegoEngine::new(EngineConfig {
            custom_rules: vec![ExternalRuleSource {
                name: "legacy_reference_test.rego".into(),
                content: LEGACY_REFERENCE_CUSTOM_RULE.into(),
            }],
            ..Default::default()
        })
        .unwrap();
        let model = make_model_from_yaml(BUILTIN_TEST_TEMPLATE);

        let diags = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();

        assert_legacy_reference_contract(&diags);
    }

    #[test]
    fn external_only_engine_keeps_the_legacy_target_id_string_for_a_reference() {
        let engine = RegoEngine::new_external_only(EngineConfig {
            custom_rules: vec![ExternalRuleSource {
                name: "legacy_reference_test.rego".into(),
                content: LEGACY_REFERENCE_CUSTOM_RULE.into(),
            }],
            ..Default::default()
        })
        .unwrap();
        let model = make_model_from_yaml(BUILTIN_TEST_TEMPLATE);

        let diags = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();

        assert_legacy_reference_contract(&diags);
    }

    #[test]
    fn custom_rule_resolve_all_list_items_keep_the_legacy_target_id_string() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains make_diag("LEGACY_LIST_ITEM", "error", "MyFunc", sprintf("subnets: %v", [subnets])) if {
    some subnets in resolve_all("MyFunc", "Properties.VpcConfig.SubnetIds")
    subnets == ["MySubnet", "subnet-0123456789abcdef0"]
}
"#,
            "LEGACY_LIST_ITEM",
        );
        assert_eq!(diags.len(), 1, "a referenced list item must come back as the target's logical ID");
    }

    /// Both kinds of package evaluate on one Regorus engine instance, so the
    /// rendering must switch per package: the built-in policies fixed to skip a
    /// reference they would otherwise judge as a literal must keep doing so while
    /// the custom package beside them still receives the legacy string.
    #[test]
    fn builtin_policies_keep_the_marker_rendering_while_a_custom_rule_gets_the_legacy_string() {
        let engine = RegoEngine::new(EngineConfig {
            custom_rules: vec![ExternalRuleSource {
                name: "legacy_reference_test.rego".into(),
                content: LEGACY_REFERENCE_CUSTOM_RULE.into(),
            }],
            ..Default::default()
        })
        .unwrap();
        let model = make_model_from_yaml(
            r#"
AWSTemplateFormatVersion: "2010-09-09"
Resources:
  Store:
    Type: AWS::SSM::Parameter
    Properties:
      Type: String
      Value: placeholder
  MyBucket:
    Type: AWS::S3::Bucket
  Distribution:
    Type: AWS::CloudFront::Distribution
    Properties:
      DistributionConfig:
        Enabled: true
        Aliases:
          - !GetAtt Store.Value
        DefaultCacheBehavior:
          TargetOriginId: primary
          ViewerProtocolPolicy: redirect-to-https
          ForwardedValues:
            QueryString: false
        Origins:
          - Id: primary
            DomainName: origin.example.com
            CustomOriginConfig:
              OriginProtocolPolicy: https-only
  MyFunc:
    Type: AWS::Lambda::Function
    Properties:
      Runtime: python3.12
      Handler: index.handler
      Role: arn:aws:iam::123456789012:role/lambda-role
      Code:
        S3Bucket: !Ref MyBucket
        S3Key: code.zip
"#,
        );

        let diags = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();

        assert!(
            !diags.iter().any(|d| d.rule_id == "E3013"),
            "the built-in alias check must not judge the logical ID 'Store' as a domain name: {:?}",
            diags.iter().map(|d| (&d.rule_id, &d.message)).collect::<Vec<_>>()
        );
        assert_legacy_reference_contract(&diags);
    }

    #[test]
    fn builtin_is_dynamic_on_conditional_property() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    is_dynamic("MyFunc", "Code.S3Bucket")
    v := {"rule_id": "B_IS_DYN", "severity": "error", "message": "ok", "resource_id": "MyFunc"}
}
"#,
            "B_IS_DYN",
        );
        assert_eq!(diags.len(), 1, "Code.S3Bucket is a Ref → dynamic");
    }

    #[test]
    fn builtin_is_dynamic_false_for_static() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    not is_dynamic("MyBucket", "BucketName")
    v := {"rule_id": "B_NOT_DYN", "severity": "error", "message": "ok", "resource_id": "MyBucket"}
}
"#,
            "B_NOT_DYN",
        );
        assert_eq!(diags.len(), 1, "BucketName is static");
    }

    #[test]
    fn builtin_resources_of_type() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    ids := resources_of_type("AWS::S3::Bucket")
    count(ids) == 1
    v := {"rule_id": "B_ROT", "severity": "error", "message": "ok", "resource_id": ""}
}
"#,
            "B_ROT",
        );
        assert_eq!(diags.len(), 1, "should find 1 S3 bucket");
    }

    #[test]
    fn builtin_has_property_true() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    has_property("MyBucket", "BucketName")
    v := {"rule_id": "B_HAS_P", "severity": "error", "message": "ok", "resource_id": "MyBucket"}
}
"#,
            "B_HAS_P",
        );
        assert_eq!(diags.len(), 1, "MyBucket has BucketName");
    }

    #[test]
    fn builtin_has_property_false() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    not has_property("MyBucket", "NonExistent")
    v := {"rule_id": "B_NO_P", "severity": "error", "message": "ok", "resource_id": "MyBucket"}
}
"#,
            "B_NO_P",
        );
        assert_eq!(diags.len(), 1, "MyBucket does not have NonExistent");
    }

    #[test]
    fn builtin_has_transform() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    has_transform("AWS::Serverless-2016-10-31")
    v := {"rule_id": "B_HAS_T", "severity": "error", "message": "ok", "resource_id": ""}
}
"#,
            "B_HAS_T",
        );
        assert_eq!(diags.len(), 1, "template has Serverless transform");
    }

    #[test]
    fn builtin_param_allowed_values() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    vals := param_allowed_values("Env")
    count(vals) == 2
    v := {"rule_id": "B_PAV", "severity": "error", "message": "ok", "resource_id": ""}
}
"#,
            "B_PAV",
        );
        assert_eq!(diags.len(), 1, "Env has 2 allowed values");
    }

    #[test]
    fn builtin_param_type() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    t := param_type("Port")
    t == "Number"
    v := {"rule_id": "B_PT", "severity": "error", "message": "ok", "resource_id": ""}
}
"#,
            "B_PT",
        );
        assert_eq!(diags.len(), 1, "Port param type is Number");
    }

    #[test]
    fn builtin_mapping_value() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    val := mapping_value("RegionMap", "us-east-1", "AMI")
    val == "ami-12345678"
    v := {"rule_id": "B_MAP", "severity": "error", "message": "ok", "resource_id": ""}
}
"#,
            "B_MAP",
        );
        assert_eq!(diags.len(), 1, "mapping lookup should return ami-12345678");
    }

    #[test]
    fn builtin_resource_condition() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    c := resource_condition("MyFunc")
    c == "IsProd"
    v := {"rule_id": "B_RC", "severity": "error", "message": "ok", "resource_id": "MyFunc"}
}
"#,
            "B_RC",
        );
        assert_eq!(diags.len(), 1, "MyFunc has condition IsProd");
    }

    #[test]
    fn builtin_resource_condition_null_when_none() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    c := resource_condition("MyBucket")
    is_null(c)
    v := {"rule_id": "B_RC_NULL", "severity": "error", "message": "ok", "resource_id": "MyBucket"}
}
"#,
            "B_RC_NULL",
        );
        assert_eq!(diags.len(), 1, "MyBucket has no condition");
    }

    #[test]
    fn builtin_get_resource() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    res := get_resource("MyBucket")
    res.resourceType == "AWS::S3::Bucket"
    v := {"rule_id": "B_GR", "severity": "error", "message": "ok", "resource_id": "MyBucket"}
}
"#,
            "B_GR",
        );
        assert_eq!(diags.len(), 1, "get_resource should return bucket data");
    }

    #[test]
    fn builtin_depends_on() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    depends_on("MyFunc", "MyBucket")
    v := {"rule_id": "B_DEP", "severity": "error", "message": "ok", "resource_id": "MyFunc"}
}
"#,
            "B_DEP",
        );
        assert_eq!(diags.len(), 1, "MyFunc depends on MyBucket");
    }

    #[test]
    fn builtin_ref_targets() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    targets := ref_targets("MyFunc")
    count(targets) > 0
    v := {"rule_id": "B_RT", "severity": "error", "message": "ok", "resource_id": "MyFunc"}
}
"#,
            "B_RT",
        );
        assert_eq!(diags.len(), 1, "MyFunc should have ref targets (MyBucket)");
    }

    #[test]
    fn builtin_ref_sources() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    sources := ref_sources("MyBucket")
    count(sources) > 0
    v := {"rule_id": "B_RS", "severity": "error", "message": "ok", "resource_id": "MyBucket"}
}
"#,
            "B_RS",
        );
        assert_eq!(diags.len(), 1, "MyBucket should have ref sources (MyFunc)");
    }

    #[test]
    fn builtin_edges_from() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    edges := edges_from("MyFunc")
    count(edges) > 0
    v := {"rule_id": "B_EF", "severity": "error", "message": "ok", "resource_id": "MyFunc"}
}
"#,
            "B_EF",
        );
        assert_eq!(diags.len(), 1, "MyFunc should have outgoing edges");
    }

    #[test]
    fn builtin_edges_to() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    edges := edges_to("MyBucket")
    count(edges) > 0
    v := {"rule_id": "B_ET", "severity": "error", "message": "ok", "resource_id": "MyBucket"}
}
"#,
            "B_ET",
        );
        assert_eq!(diags.len(), 1, "MyBucket should have incoming edges");
    }

    #[test]
    fn builtin_make_diag_at() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    v := make_diag_at("B_MDA", "error", "MyBucket", "BucketName", "test diag at")
}
"#,
            "B_MDA",
        );
        assert_eq!(diags.len(), 1);
        assert_eq!(diags[0].message, "test diag at");
    }

    #[test]
    fn builtin_make_diag_full() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    v := make_diag_full("B_MDF", "error", "MyBucket", "BucketName", "full diag", "fix it", "https://example.com")
}
"#,
            "B_MDF",
        );
        assert_eq!(diags.len(), 1);
        assert_eq!(diags[0].message, "full diag");
    }

    #[test]
    fn builtin_schema_properties_known_type() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    props := schema_properties("AWS::S3::Bucket")
    # Just verify the builtin is callable and returns an array
    is_array(props)
    v := {"rule_id": "B_SP", "severity": "error", "message": "ok", "resource_id": ""}
}
"#,
            "B_SP",
        );
        assert_eq!(diags.len(), 1, "schema_properties should be callable and return array");
    }

    #[test]
    fn builtin_schema_required_known_type() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    req := schema_required("AWS::Lambda::Function")
    is_array(req)
    v := {"rule_id": "B_SR", "severity": "error", "message": "ok", "resource_id": ""}
}
"#,
            "B_SR",
        );
        assert_eq!(diags.len(), 1, "schema_required should be callable and return array");
    }

    #[test]
    fn builtin_conditions_compatible() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    conditions_compatible("MyBucket", "MyQueue")
    v := {"rule_id": "B_CC", "severity": "error", "message": "ok", "resource_id": ""}
}
"#,
            "B_CC",
        );
        assert_eq!(diags.len(), 1, "unconditional resources are always compatible");
    }

    #[test]
    fn builtin_resolve_type() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    t := resolve_type("MyBucket", "BucketName")
    t == "string"
    v := {"rule_id": "B_RTYPE", "severity": "error", "message": "ok", "resource_id": "MyBucket"}
}
"#,
            "B_RTYPE",
        );
        assert_eq!(diags.len(), 1, "BucketName should resolve to string type");
    }

    #[test]
    fn region_override_available_to_builtins() {
        let custom_rego = r#"
package region_test
import rego.v1
violation contains v if {
    r := input_region()
    r == "eu-west-1"
    v := {"rule_id": "B_REGION", "severity": "error", "message": "ok", "resource_id": ""}
}
"#;
        let config = EngineConfig {
            custom_rules: vec![ExternalRuleSource { name: "region_test.rego".into(), content: custom_rego.into() }],
            guard_rules: vec![],
            ..Default::default()
        };
        let engine = RegoEngine::new(config).unwrap();
        let model = make_model_from_yaml(BUILTIN_TEST_TEMPLATE);
        let mut vc = ValidateConfig::default();
        vc.pseudo_parameter_overrides.region = Some("eu-west-1".to_string());
        let diags = engine.evaluate_rules(&model, &vc).unwrap();
        let found = diags.iter().find(|d| d.rule_id == "B_REGION");
        assert!(found.is_some(), "input_region() should return eu-west-1");
    }

    #[test]
    fn multiple_custom_packages_all_evaluated() {
        let pkg_a = r#"
package custom_a
import rego.v1
violation contains v if {
    v := {"rule_id": "PKG_A", "severity": "error", "message": "from A", "resource_id": ""}
}
"#;
        let pkg_b = r#"
package custom_b
import rego.v1
violation contains v if {
    v := {"rule_id": "PKG_B", "severity": "error", "message": "from B", "resource_id": ""}
}
"#;
        let config = EngineConfig {
            custom_rules: vec![
                ExternalRuleSource { name: "a.rego".into(), content: pkg_a.into() },
                ExternalRuleSource { name: "b.rego".into(), content: pkg_b.into() },
            ],
            guard_rules: vec![],
            ..Default::default()
        };
        let engine = RegoEngine::new(config).unwrap();
        let model = make_model_from_yaml(BUILTIN_TEST_TEMPLATE);
        let diags = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();
        assert!(diags.iter().any(|d| d.rule_id == "PKG_A"), "package A should fire");
        assert!(diags.iter().any(|d| d.rule_id == "PKG_B"), "package B should fire");
    }

    #[test]
    fn duplicate_custom_package_deduplicated() {
        let source = r#"
package custom_dedup
import rego.v1
violation contains v if {
    v := {"rule_id": "DEDUP1", "severity": "error", "message": "first", "resource_id": ""}
}
"#;
        let config = EngineConfig {
            custom_rules: vec![
                ExternalRuleSource { name: "a.rego".into(), content: source.into() },
                ExternalRuleSource { name: "b.rego".into(), content: source.into() },
            ],
            guard_rules: vec![],
            ..Default::default()
        };
        let engine = RegoEngine::new(config).unwrap();
        assert_eq!(
            engine.custom_packages.iter().filter(|p| p.contains("custom_dedup")).count(),
            1,
            "duplicate package should be deduplicated"
        );
    }

    #[test]
    fn builtin_follow_ref_returns_target_resource() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    target := follow_ref("MyFunc", "Code.S3Bucket")
    target == "MyBucket"
    v := {"rule_id": "B_FREF", "severity": "error", "message": "ok", "resource_id": "MyFunc"}
}
"#,
            "B_FREF",
        );
        assert_eq!(diags.len(), 1, "follow_ref should resolve Ref to MyBucket");
    }

    #[test]
    fn builtin_is_from_parameter_true_for_ref_param() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    is_from_parameter("MyBucket", "Tags.0.Value")
    v := {"rule_id": "B_IFP", "severity": "error", "message": "ok", "resource_id": "MyBucket"}
}
"#,
            "B_IFP",
        );
        // Tags.0.Value uses Fn::If with a condition based on a parameter, so it's from a parameter
        assert!(diags.len() <= 1, "expected at most 1 diagnostic, got {}", diags.len());
    }

    #[test]
    fn builtin_flatten_list_returns_items() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    items := flatten_list("MyBucket", "Tags")
    count(items) >= 1
    v := {"rule_id": "B_FL", "severity": "error", "message": "ok", "resource_id": "MyBucket"}
}
"#,
            "B_FL",
        );
        assert_eq!(diags.len(), 1, "flatten_list should return tag items");
    }

    #[test]
    fn builtin_resolve_scenarios_returns_values() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    scenarios := resolve_scenarios("MyBucket", "Tags.0.Value")
    count(scenarios) >= 1
    v := {"rule_id": "B_RS2", "severity": "error", "message": "ok", "resource_id": "MyBucket"}
}
"#,
            "B_RS2",
        );
        assert_eq!(diags.len(), 1, "resolve_scenarios should return at least one scenario");
    }

    #[test]
    fn builtin_is_satisfiable_with_compatible_assumptions() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    is_satisfiable({"IsProd": true})
    v := {"rule_id": "B_SAT", "severity": "error", "message": "ok", "resource_id": ""}
}
"#,
            "B_SAT",
        );
        assert_eq!(diags.len(), 1, "IsProd=true should be satisfiable");
    }

    #[test]
    fn builtin_make_diag_produces_diagnostic() {
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    v := make_diag("B_MD", "error", "MyBucket", "test message")
}
"#,
            "B_MD",
        );
        assert_eq!(diags.len(), 1);
        assert_eq!(diags[0].message, "test message");
    }

    #[test]
    fn builtin_estimated_string_length_bounds_returns_both_bounds() {
        // `!If [IsProd, "production", "development"]` - the deployment picks one of
        // two known values, so the length is bounded but not fixed.
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    bounds := estimated_string_length_bounds("MyBucket", "Tags.0.Value")
    bounds.shortest == 10
    bounds.longest == 11
    v := {"rule_id": "B_ESL", "severity": "error", "message": "ok", "resource_id": "MyBucket"}
}
"#,
            "B_ESL",
        );
        assert_eq!(diags.len(), 1, "a value chosen between two known strings must report both bounds");
    }

    #[test]
    fn builtin_estimated_string_length_bounds_is_undefined_for_a_literal() {
        // Schema validation checks a literal against the constraint, so this rule
        // is not given one to estimate.
        let diags = eval_builtin_policy(
            r#"
package builtin_test
import rego.v1
violation contains v if {
    bounds := estimated_string_length_bounds("MyBucket", "BucketName")
    v := {"rule_id": "B_ESL_LITERAL", "severity": "error", "message": "ok", "resource_id": "MyBucket"}
}
"#,
            "B_ESL_LITERAL",
        );
        assert!(diags.is_empty(), "a literal string must yield no bounds, got {diags:?}");
    }

    #[test]
    fn regoengine_is_send_and_sync() {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<RegoEngine>();
    }

    #[test]
    fn evaluate_rules_isolates_model_and_region_across_threads() {
        // One engine, many threads, each with a distinct model and region. The
        // probes echo the region and the resource names the thread should see, so
        // any leakage between threads' evaluation contexts surfaces as a mismatch.
        let probe = r#"
package concurrency_probe
import rego.v1
violation contains v if {
    r := input_region()
    v := {"rule_id": "REGIONPROBE", "severity": "info", "message": r, "resource_id": ""}
}
violation contains v if {
    some name in resources_of_type("AWS::S3::Bucket")
    v := {"rule_id": "MODELPROBE", "severity": "info", "message": name, "resource_id": name}
}
violation contains v if {
    some name, _ in input.resources
    v := {"rule_id": "INPUTPROBE", "severity": "info", "message": name, "resource_id": name}
}
"#;
        let config = EngineConfig {
            custom_rules: vec![ExternalRuleSource { name: "concurrency_probe.rego".into(), content: probe.into() }],
            guard_rules: vec![],
            ..Default::default()
        };
        let engine = Arc::new(RegoEngine::new(config).unwrap());

        const THREAD_COUNT: usize = 8;
        const ITERATIONS_PER_THREAD: usize = 128;
        let barrier = Arc::new(Barrier::new(THREAD_COUNT));
        let handles: Vec<_> = (0..THREAD_COUNT)
            .map(|i| {
                let engine = Arc::clone(&engine);
                let barrier = Arc::clone(&barrier);
                thread::spawn(move || {
                    let region = format!("region-{i}");
                    let bucket = format!("Bucket{i}");
                    let model = make_model_from_yaml(&format!(
                        "AWSTemplateFormatVersion: \"2010-09-09\"\nResources:\n  {bucket}:\n    Type: AWS::S3::Bucket\n"
                    ));
                    let mut config = ValidateConfig::default();
                    config.pseudo_parameter_overrides.region = Some(region.clone());
                    // Release every thread at once so their evaluations overlap.
                    barrier.wait();
                    let mut diagnostics = Vec::new();
                    for _ in 0..ITERATIONS_PER_THREAD {
                        diagnostics = engine.evaluate_rules(&model, &config).unwrap();
                        let observed_regions: Vec<&str> = diagnostics
                            .iter()
                            .filter(|diagnostic| diagnostic.rule_id == "REGIONPROBE")
                            .map(|diagnostic| diagnostic.message.as_str())
                            .collect();
                        assert_eq!(observed_regions, vec![region.as_str()]);
                        let observed_resources: Vec<&str> = diagnostics
                            .iter()
                            .filter(|diagnostic| diagnostic.rule_id == "MODELPROBE")
                            .map(|diagnostic| diagnostic.message.as_str())
                            .collect();
                        assert_eq!(observed_resources, vec![bucket.as_str()]);
                        let observed_input_resources: Vec<&str> = diagnostics
                            .iter()
                            .filter(|diagnostic| diagnostic.rule_id == "INPUTPROBE")
                            .map(|diagnostic| diagnostic.message.as_str())
                            .collect();
                        assert_eq!(observed_input_resources, vec![bucket.as_str()]);
                    }
                    (region, bucket, diagnostics)
                })
            })
            .collect();

        for handle in handles {
            let (region, bucket, diags) = handle.join().unwrap();
            let regions: Vec<&str> =
                diags.iter().filter(|d| d.rule_id == "REGIONPROBE").map(|d| d.message.as_str()).collect();
            assert_eq!(regions, vec![region.as_str()], "a thread must observe only its own region");
            let resources: Vec<&str> =
                diags.iter().filter(|d| d.rule_id == "MODELPROBE").map(|d| d.message.as_str()).collect();
            assert_eq!(resources, vec![bucket.as_str()], "a thread must observe only its own model");
            let input_resources: Vec<&str> =
                diags.iter().filter(|d| d.rule_id == "INPUTPROBE").map(|d| d.message.as_str()).collect();
            assert_eq!(input_resources, vec![bucket.as_str()], "a thread must observe only its own Rego input");
        }
    }

    #[test]
    fn shared_engine_keeps_thousand_sequential_reports_template_local() {
        const CUSTOM_RULE: &str = r#"
package sequential_isolation
import rego.v1

violation contains v if {
    pattern := input.template.description
    some name, resource in input.resources
    regex.match(pattern, resource.properties.Value)
    region := input_region()
    message := sprintf("%s/%s", [name, region])
    v := {
        "rule_id": "SEQUENTIALPROBE",
        "severity": "info",
        "category": "isolation",
        "message": message,
        "resource_id": name,
    }
}
"#;
        let engine = RegoEngine::new(EngineConfig {
            custom_rules: vec![ExternalRuleSource {
                name: "sequential_isolation.rego".into(),
                content: CUSTOM_RULE.into(),
            }],
            ..Default::default()
        })
        .unwrap();
        let schema_validator = SchemaValidator::default();

        for iteration in 0..1_000 {
            let (resource_name, region, pattern, value) = if iteration % 2 == 0 {
                ("AlphaResource", "alpha-region", "^alpha+$", "alpha")
            } else {
                ("BetaResource", "beta-region", "^beta+$", "beta")
            };
            let template = format!(
                r#"{{
                    "AWSTemplateFormatVersion": "2010-09-09",
                    "Description": "{pattern}",
                    "Resources": {{
                        "{resource_name}": {{
                            "Type": "Custom::Probe",
                            "Properties": {{"Value": "{value}"}}
                        }}
                    }}
                }}"#
            );
            let mut config = ValidateConfig {
                detail_level: DetailLevel::Detailed,
                severity_level: Severity::Debug,
                disable_builtin_rules: true,
                ..Default::default()
            };
            config.pseudo_parameter_overrides.region = Some(region.to_string());

            let report = validate_bytes(&engine, &schema_validator, template.as_bytes(), config).unwrap();
            let probes: Vec<&Diagnostic> =
                report.diagnostics.iter().filter(|diagnostic| diagnostic.rule_id == "SEQUENTIALPROBE").collect();
            assert_eq!(probes.len(), 1, "iteration {iteration} must emit exactly one probe diagnostic");
            let expected_message = format!("{resource_name}/{region}");
            assert_eq!(probes[0].message, expected_message, "the current template supplies the diagnostic message");
            assert_eq!(
                probes[0].rule_description.as_deref(),
                Some(expected_message.as_str()),
                "the current template supplies its own detailed rule description"
            );
            assert_eq!(probes[0].category.as_deref(), Some("isolation"));
        }
    }

    #[test]
    fn failed_evaluation_does_not_contaminate_the_next_call() {
        const CUSTOM_RULE: &str = r#"
package error_cleanup
import rego.v1

violation contains v if {
    input.template.description == "error"
    ip_overlaps("not-a-cidr", "10.0.0.0/8")
    v := {"rule_id": "UNREACHABLE", "severity": "error", "message": "unreachable"}
}

violation contains v if {
    input.template.description == "ok"
    some name in resources_of_type("Custom::Probe")
    region := input_region()
    v := {
        "rule_id": "RECOVERYPROBE",
        "severity": "info",
        "message": sprintf("%s/%s", [name, region]),
        "resource_id": name,
    }
}
"#;
        let engine = RegoEngine::new(EngineConfig {
            custom_rules: vec![ExternalRuleSource { name: "error_cleanup.rego".into(), content: CUSTOM_RULE.into() }],
            ..Default::default()
        })
        .unwrap();
        let error_model =
            make_model_from_yaml("Description: error\nResources:\n  ErrorResource:\n    Type: Custom::Probe\n");
        let valid_model =
            make_model_from_yaml("Description: ok\nResources:\n  ValidResource:\n    Type: Custom::Probe\n");
        let mut error_config = ValidateConfig { disable_builtin_rules: true, ..Default::default() };
        error_config.pseudo_parameter_overrides.region = Some("error-region".to_string());
        assert!(engine.evaluate_rules(&error_model, &error_config).is_err(), "the first evaluation must fail");

        let mut valid_config = ValidateConfig { disable_builtin_rules: true, ..Default::default() };
        valid_config.pseudo_parameter_overrides.region = Some("valid-region".to_string());
        let diagnostics = engine.evaluate_rules(&valid_model, &valid_config).unwrap();
        let recovery: Vec<&Diagnostic> =
            diagnostics.iter().filter(|diagnostic| diagnostic.rule_id == "RECOVERYPROBE").collect();
        assert_eq!(recovery.len(), 1, "the next evaluation must recover after the prior error");
        assert_eq!(recovery[0].message, "ValidResource/valid-region");
    }

    const SUPPRESSION_TEMPLATE: &str = r#"
AWSTemplateFormatVersion: "2010-09-09"
Resources:
  MyInstance:
    Type: AWS::EC2::Instance
    Properties:
      ImageId: ami-12345678
      InstanceType: !Sub "t3.micro"
"#;

    fn exclude_ids(ids: &[&str]) -> ValidateConfig {
        ValidateConfig {
            filters: FilterConfig::new(
                RuleFilterConfig::default(),
                RuleFilterConfig { ids: ids.iter().map(|s| s.to_string()).collect(), ..Default::default() },
            ),
            ..Default::default()
        }
    }

    #[test]
    fn exclude_id_suppresses_matching_builtin_and_preserves_other_findings() {
        let engine = make_engine();
        let model = make_model_from_yaml(SUPPRESSION_TEMPLATE);

        let unfiltered = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();
        assert!(unfiltered.iter().any(|d| d.rule_id == "W9010"), "hardcoded AMI should trigger W9010");
        let mut other_findings: Vec<(String, String)> = unfiltered
            .iter()
            .filter(|d| d.rule_id != "W9010")
            .map(|d| (d.rule_id.clone(), d.message.clone()))
            .collect();
        assert!(!other_findings.is_empty(), "template must also produce a non-W9010 finding worth preserving");
        other_findings.sort();

        let filtered = engine.evaluate_rules(&model, &exclude_ids(&["W9010"])).unwrap();
        assert!(filtered.iter().all(|d| d.rule_id != "W9010"), "the excluded rule must not be emitted");
        let mut remaining: Vec<(String, String)> =
            filtered.iter().map(|d| (d.rule_id.clone(), d.message.clone())).collect();
        remaining.sort();
        assert_eq!(remaining, other_findings, "excluding one rule must not alter any other diagnostic");

        let unfiltered_again = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();
        assert!(
            unfiltered_again.iter().any(|diagnostic| diagnostic.rule_id == "W9010"),
            "a later unfiltered call must not inherit the previous call's suppressed-rule set"
        );
    }

    #[test]
    fn include_id_globally_suppresses_every_other_builtin() {
        let engine = make_engine();
        let model = make_model_from_yaml(SUPPRESSION_TEMPLATE);

        let config = ValidateConfig {
            filters: FilterConfig::new(
                RuleFilterConfig { ids: vec!["W9010".to_string()], ..Default::default() },
                RuleFilterConfig::default(),
            ),
            ..Default::default()
        };
        let filtered = engine.evaluate_rules(&model, &config).unwrap();
        assert!(filtered.iter().any(|d| d.rule_id == "W9010"), "the included rule must still fire");
        assert!(
            filtered.iter().all(|d| d.rule_id == "W9010"),
            "a non-empty include filter must globally suppress every rule it can never admit"
        );
    }

    #[test]
    fn severity_level_prunes_lower_severity_builtins() {
        let engine = make_engine();
        let model = make_model_from_yaml(SUPPRESSION_TEMPLATE);

        let default_run = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();
        assert!(default_run.iter().any(|d| d.rule_id == "W9010"), "W9010 fires at the default Info level");

        let config = ValidateConfig { severity_level: Severity::Error, ..Default::default() };
        let error_only = engine.evaluate_rules(&model, &config).unwrap();
        assert!(
            error_only.iter().all(|d| d.severity >= Severity::Error),
            "no sub-Error diagnostics should be produced when the level is Error"
        );
        assert!(
            error_only.iter().all(|d| d.rule_id != "W9010"),
            "the Warn rule W9010 must be pruned at the Error level"
        );
    }

    #[test]
    fn cfn_rule_active_guard_short_circuits_a_suppressed_rule_body() {
        // Every built-in clause is shaped like this probe: cfn_rule_active guards
        // the body, so a globally suppressed rule stops before its body runs. The
        // body here would evaluate regex.match on an invalid pattern if reached;
        // the probe diagnostic is the observable signal that the body executed.
        let probe = r#"
package guard_probe
import rego.v1
violation contains v if {
    cfn_rule_active("W9010")
    not regex.match("(unterminated", "anything")
    v := {"rule_id": "GUARDPROBE", "severity": "error", "message": "guarded body ran", "resource_id": ""}
}
"#;
        let config = EngineConfig {
            custom_rules: vec![ExternalRuleSource { name: "guard_probe.rego".into(), content: probe.into() }],
            guard_rules: vec![],
            ..Default::default()
        };
        let engine = RegoEngine::new(config).unwrap();
        let model = make_model_from_yaml(
            r#"
AWSTemplateFormatVersion: "2010-09-09"
Resources:
  MyBucket:
    Type: AWS::S3::Bucket
"#,
        );

        let active = engine.evaluate_rules(&model, &ValidateConfig::default()).unwrap();
        assert!(active.iter().any(|d| d.rule_id == "GUARDPROBE"), "an active rule's guarded body must run");

        let suppressed = engine.evaluate_rules(&model, &exclude_ids(&["W9010"])).unwrap();
        assert!(
            suppressed.iter().all(|d| d.rule_id != "GUARDPROBE"),
            "a suppressed rule's guarded body - including its invalid regex - must not run"
        );
    }
}