template_model/

model.rs

use crate::authored_template::render_authored_json;
use crate::budget::{BudgetKind, BudgetTracker};
use crate::conditions::{ConditionModel, Satisfiability};
use crate::consts::*;
use crate::defect::{DefectPhase, ParseDefect};
use crate::graph::{Edge, ReferenceGraph};
use crate::ir::*;
use crate::is_custom_resource_type;
use crate::json_value::JsonValue;
use crate::member_presence::absent_alternatives;
use crate::regions::*;
use crate::resolved_value::*;
use crate::resolver::*;
use crate::sam;
use crate::serialization::resolved_value_to_json;
use crate::span::SpanProvider;
use log::{debug, info, warn};
use serde::Serialize;
use std::cmp::Reverse;
use std::collections::{BTreeMap, BTreeSet, BinaryHeap, HashMap, HashSet};
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

/// A resource property path paired with a string value found at it, such as a substitution variable or literal.
#[derive(Debug, Clone, Serialize, Default)]
#[cfg_attr(feature = "wasm-bindings", derive(tsify::Tsify))]
#[cfg_attr(feature = "uniffi-bindings", derive(uniffi::Record))]
#[serde(rename_all = "camelCase")]
pub struct PathValuePair {
    /// Dot-separated property path within the resource (e.g. 'Properties.BucketName').
    pub path: String,
    pub value: String,
}

/// One deployment scenario for an authored member of a resolved list.
#[derive(Debug, Clone)]
pub struct ListMemberScenario {
    /// Identity shared by members from the same outer list alternative.
    pub container_scenario: usize,
    /// Conditions selecting the outer list alternative containing this member.
    pub container_conditions: HashMap<String, bool>,
    /// Authored zero-based position of the member in that list alternative.
    pub index: usize,
    /// Member value after expanding only conditions that wrap this member.
    pub value: ResolvedValue,
    /// Conditions selecting both the outer list alternative and this member value.
    pub conditions: HashMap<String, bool>,
}

/// A property that is dropped (resolves to AWS::NoValue) in one branch of an Fn::If condition.
#[derive(Debug, Clone, Serialize, Default)]
#[cfg_attr(feature = "wasm-bindings", derive(tsify::Tsify))]
#[cfg_attr(feature = "uniffi-bindings", derive(uniffi::Record))]
#[serde(rename_all = "camelCase")]
pub struct ConditionalNullEntry {
    /// Dot-separated property path that becomes absent in one branch.
    pub path: String,
    /// Name of the condition governing the Fn::If that drops this property.
    pub condition: String,
    /// True when the property is absent in the condition's true branch; false when absent in the false branch.
    pub null_in_true_branch: bool,
}

/// Per-resource observations collected while resolving intrinsics, used to drive lint checks.
#[derive(Debug, Clone, Serialize, Default)]
#[cfg_attr(feature = "wasm-bindings", derive(tsify::Tsify))]
#[cfg_attr(feature = "uniffi-bindings", derive(uniffi::Record))]
#[serde(rename_all = "camelCase")]
pub struct ResourceDiagnostics {
    /// Mapping names referenced by Fn::FindInMap within this resource.
    pub find_in_map_refs: Vec<String>,
    /// Fn::Sub uses whose template is a single variable that could be a plain Ref; each pairs the property path with the variable name.
    pub simple_subs: Vec<PathValuePair>,
    /// Property paths where Fn::Sub wraps a constant string with no variables to substitute.
    pub redundant_subs: Vec<String>,
    /// Property paths where Fn::Join uses an empty delimiter, concatenating its elements directly.
    pub empty_joins: Vec<String>,
    /// Names of conditions referenced by this resource's property values.
    pub condition_refs: Vec<String>,
    /// Property paths building ARNs with a hardcoded 'aws' partition instead of AWS::Partition.
    pub hardcoded_partition_arns: Vec<String>,
    pub conditionally_null_props: Vec<ConditionalNullEntry>,
    pub foreach_expansions: Vec<ForEachExpansion>,
    /// Occurrences of ${...} placeholders outside an Fn::Sub that will not be substituted; each pairs the property path with the placeholder text.
    pub unsubstituted_variables: Vec<PathValuePair>,
    /// Fn::Sub map keys not referenced in the template string; each pairs the property path with the unused key name.
    pub unused_sub_keys: Vec<PathValuePair>,
    /// Property values that are a raw pseudo-parameter string (e.g. "AWS::Region") instead of using Ref.
    pub raw_pseudo_params: Vec<PathValuePair>,
    /// Property paths containing a {{resolve:secretsmanager:...}} dynamic reference.
    pub secretsmanager_ref_paths: Vec<String>,
    /// References whose target is not a defined resource, parameter, or pseudo parameter; each pairs the property path with the missing target name.
    pub invalid_refs: Vec<PathValuePair>,
}

/// A template resource with its metadata and properties after intrinsic functions have been resolved.
#[derive(Debug, Clone, Serialize)]
#[cfg_attr(feature = "wasm-bindings", derive(tsify::Tsify))]
#[cfg_attr(feature = "uniffi-bindings", derive(uniffi::Record))]
#[serde(rename_all = "camelCase")]
pub struct ResolvedResource {
    pub logical_id: String,
    pub resource_type: String,
    #[cfg_attr(feature = "wasm-bindings", tsify(optional))]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub condition: Option<String>,
    pub depends_on: Vec<String>,
    #[cfg_attr(feature = "wasm-bindings", tsify(optional))]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub deletion_policy: Option<ResolvedValue>,
    #[cfg_attr(feature = "wasm-bindings", tsify(optional))]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub update_replace_policy: Option<ResolvedValue>,
    #[cfg_attr(feature = "wasm-bindings", tsify(optional, type = "JsonValue"))]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub update_policy: Option<JsonValue>,
    #[cfg_attr(feature = "wasm-bindings", tsify(optional, type = "JsonValue"))]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub creation_policy: Option<JsonValue>,
    #[cfg_attr(feature = "wasm-bindings", tsify(optional, type = "JsonValue"))]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub metadata: Option<JsonValue>,
    #[cfg_attr(feature = "wasm-bindings", tsify(type = "Record<string, ResolvedValue>"))]
    pub properties: HashMap<String, ResolvedValue>,
    /// True when a resource-level or property-set macro can determine the
    /// effective property set only at deploy time, or when a non-map
    /// `Properties` intrinsic has no statically inspectable property map.
    pub properties_dynamic: bool,
    pub diagnostics: ResourceDiagnostics,
}

/// Effective top-level state of a lifecycle attribute after reachable
/// `Fn::If` branches and `AWS::NoValue` removal are considered.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct LifecycleAttributeStatus {
    /// At least one reachable branch keeps the attribute on the resource.
    pub may_be_present: bool,
    /// The first reachable authored scalar or list, rendered for a type diagnostic.
    pub invalid_value: Option<String>,
}

struct PrimaryIdentifierScenario {
    identity_tuple: Vec<String>,
    display_tuple: Vec<String>,
    assumptions: Vec<(String, bool)>,
}

/// Where a candidate comparison sits in the original all-pairs traversal: left
/// resource, then right resource, then left scenario, then right scenario. Merging
/// every identity group by this key replays comparisons - and therefore the
/// satisfiability queries they issue - in the identical order and count the
/// all-pairs traversal produced, so the model's shared, cumulative satisfiability
/// budget is drawn down exactly as before.
type PrimaryIdentifierComparisonOrder = (usize, usize, usize, usize);

/// Yields the resource/scenario comparisons within one primary-identifier identity
/// group in [`PrimaryIdentifierComparisonOrder`]. Grouping by identity first skips
/// the cross-identity pairs the all-pairs traversal built only to discard, while
/// this cursor preserves the surviving comparisons' original order so results and
/// satisfiability accounting are unchanged.
struct IdentityGroupPairCursor {
    /// Distinct resources in the group in ascending `per_resource` index, each with
    /// its matching scenario indices in ascending order.
    resources: Vec<(usize, Vec<usize>)>,
    left_resource: usize,
    right_resource: usize,
    left_scenario: usize,
    right_scenario: usize,
}

impl IdentityGroupPairCursor {
    /// A cursor over one identity group's comparisons, or `None` when the group
    /// spans fewer than two resources and so has nothing to compare. `candidates`
    /// are `(per_resource index, scenario index)` pairs in ascending order, so
    /// consecutive entries that share a resource index belong to that resource.
    fn new(candidates: &[(usize, usize)]) -> Option<Self> {
        let mut resources: Vec<(usize, Vec<usize>)> = Vec::new();
        for &(resource_index, scenario_index) in candidates {
            match resources.last_mut() {
                Some((last_resource_index, scenario_indices)) if *last_resource_index == resource_index => {
                    scenario_indices.push(scenario_index);
                }
                _ => resources.push((resource_index, vec![scenario_index])),
            }
        }
        if resources.len() < 2 {
            return None;
        }
        Some(Self { resources, left_resource: 0, right_resource: 1, left_scenario: 0, right_scenario: 0 })
    }
}

impl Iterator for IdentityGroupPairCursor {
    type Item = PrimaryIdentifierComparisonOrder;

    fn next(&mut self) -> Option<Self::Item> {
        while self.left_resource + 1 < self.resources.len() {
            if self.right_resource >= self.resources.len() {
                self.left_resource += 1;
                self.right_resource = self.left_resource + 1;
                self.left_scenario = 0;
                self.right_scenario = 0;
                continue;
            }
            if self.left_scenario >= self.resources[self.left_resource].1.len() {
                self.right_resource += 1;
                self.left_scenario = 0;
                self.right_scenario = 0;
                continue;
            }
            if self.right_scenario >= self.resources[self.right_resource].1.len() {
                self.left_scenario += 1;
                self.right_scenario = 0;
                continue;
            }
            let comparison = (
                self.resources[self.left_resource].0,
                self.resources[self.right_resource].0,
                self.resources[self.left_resource].1[self.left_scenario],
                self.resources[self.right_resource].1[self.right_scenario],
            );
            self.right_scenario += 1;
            return Some(comparison);
        }
        None
    }
}

/// An Fn::ForEach loop within a resource that expands a property over a collection.
#[derive(Debug, Clone, Serialize)]
#[cfg_attr(feature = "wasm-bindings", derive(tsify::Tsify))]
#[cfg_attr(feature = "uniffi-bindings", derive(uniffi::Record))]
#[serde(rename_all = "camelCase")]
pub struct ForEachExpansion {
    /// Property path within the resource where the Fn::ForEach appears.
    pub property_path: String,
    /// Loop variable name bound to each element during expansion.
    pub identifier: String,
    /// Human-readable description of the collection being iterated over.
    pub collection_source: String,
}

/// A template output with its value and metadata after intrinsic functions have been resolved.
#[derive(Debug, Clone, Serialize)]
#[cfg_attr(feature = "wasm-bindings", derive(tsify::Tsify))]
#[cfg_attr(feature = "uniffi-bindings", derive(uniffi::Record))]
#[serde(rename_all = "camelCase")]
pub struct ResolvedOutput {
    pub value: ResolvedValue,
    #[cfg_attr(feature = "wasm-bindings", tsify(optional))]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub description: Option<String>,
    #[cfg_attr(feature = "wasm-bindings", tsify(optional))]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub condition: Option<String>,
    #[cfg_attr(feature = "wasm-bindings", tsify(optional))]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub export_name: Option<ResolvedValue>,
}

#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct TemplateRule {
    pub name: String,
    pub condition: Option<serde_json::Value>,
    #[serde(skip)]
    pub condition_node: crate::ir::NodeRef,
    pub assertions: Vec<RuleAssertion>,
}

#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct RuleAssertion {
    pub assert: serde_json::Value,
    #[serde(skip)]
    pub assert_node: crate::ir::NodeRef,
    pub description: Option<String>,
}

pub struct SemanticModel {
    pub arena: Arena,
    /// The template's top-level map in `arena`; the authored view is rendered from it.
    root: NodeRef,
    /// The template as CloudFormation JSON, rendered from `arena` on first request
    /// and shared by every consumer that evaluates the authored template.
    authored_template: OnceLock<serde_json::Value>,
    pub span_index: SourceSpanIndex,
    conditional_branch_first_descendant_spans: HashMap<String, SourceSpan>,
    pub format_version: Option<String>,
    pub description: Option<String>,
    pub transforms: Vec<String>,
    pub raw_top_level_keys: Vec<String>,
    pub template_metadata: Option<serde_json::Value>,
    pub rules: Option<serde_json::Value>,
    pub parsed_rules: Vec<TemplateRule>,
    pub parameters: HashMap<String, ParameterInfo>,
    pub mappings: MappingData,
    pub conditions: ConditionModel,
    pub resources: HashMap<String, ResolvedResource>,
    pub outputs: HashMap<String, ResolvedOutput>,
    pub graph: ReferenceGraph,
    pub resources_by_type: HashMap<String, Vec<String>>,
    pub diagnostics: Vec<ParseDefect>,
    pub output_empty_joins: Vec<String>,
    pub sam_globals: HashMap<String, HashMap<String, serde_json::Value>>,
    pub sam_implicit_resources: HashSet<String>,
    pub globals_param_refs: Vec<String>,
    pub is_cdk: bool,
    pub fn_if_conditions: Vec<String>,
    /// Mapping names referenced by an `Fn::FindInMap` with a literal map name,
    /// collected template-wide (resources, outputs, conditions, ForEach bodies).
    pub find_in_map_names: HashSet<String>,
    /// Parameter names referenced from within another parameter's definition
    /// (e.g. a Default of `!Ref OtherParam`); these still count as usage.
    pub params_referenced_in_definitions: HashSet<String>,
    /// True when any `Fn::FindInMap` uses a non-literal map name, which disables
    /// the unused-mapping check because usage can no longer be attributed to a
    /// specific mapping.
    pub has_dynamic_findinmap_name: bool,
    pub resolution_sources: HashMap<(String, String), String>,
    parameter_overrides: HashSet<String>,
    /// Per-resource set of property paths whose value is produced by an intrinsic
    /// function - either recorded with an `Intrinsic/*` resolution source or
    /// anchoring a reference edge. Built once at construction so `is_from_intrinsic`
    /// answers ancestor queries with borrowed lookups instead of rescanning the
    /// resolution sources and outgoing edges on every call.
    intrinsic_source_paths: HashMap<String, HashSet<String>>,
    pseudo_parameter_source_paths: HashMap<String, HashSet<String>>,
    /// (resource_id, property_path) → authored expression retained for opaque
    /// value identity or concrete intrinsic structure inspection.
    value_nodes: HashMap<(String, String), NodeRef>,
    /// Authored lifecycle attributes retained for condition-aware presence and
    /// top-level shape analysis without changing their public raw JSON view.
    lifecycle_attribute_nodes: HashMap<(String, String), NodeRef>,
    lifecycle_attribute_status_cache: Mutex<HashMap<(String, String), LifecycleAttributeStatus>>,
    /// Resource IDs whose authored `Condition` attribute is present but is not a
    /// condition-name string. Their deployment coexistence cannot be determined.
    invalid_resource_conditions: HashSet<String>,
    /// Synthetic condition names created for `Fn::If` first arguments that are
    /// expressions rather than condition-name strings. The expressions remain
    /// available for best-effort value resolution but cannot prove reachability.
    invalid_inline_conditions: HashSet<String>,
    resolve_memo: Mutex<HashMap<(String, String), Option<ResolvedValue>>>,
    raw_scenario_memo:
        Mutex<HashMap<(String, String, usize), (Arc<Vec<(ResolvedValue, HashMap<String, bool>)>>, bool)>>,
    properties_scenario_cache:
        Mutex<HashMap<(String, Option<BTreeSet<String>>), Arc<Vec<(ResolvedValue, HashMap<String, bool>)>>>>,
    list_member_scenario_cache: Mutex<HashMap<(String, String), Arc<Vec<ListMemberScenario>>>>,
    scenario_memo: Mutex<HashMap<(String, String), Arc<Vec<(serde_json::Value, HashMap<String, bool>)>>>>,
    lifecycle_policy_scenario_cache: Mutex<HashMap<(String, String), Vec<(serde_json::Value, HashMap<String, bool>)>>>,
    /// Cumulative count of scenarios materialized by `resolve_scenarios` across
    /// the whole validation, charged against `MAX_TOTAL_SCENARIO_COMBINATIONS`.
    /// Bounds total scenario-expansion work the way `ConditionModel`'s
    /// `sat_iterations_used` bounds total satisfiability work.
    scenario_combinations_used: AtomicU64,
    /// Deterministic, deduplicating tracker for all validation-budget
    /// exhaustions across model construction and downstream validation.
    budget_tracker: Arc<BudgetTracker>,
    /// Serializes scenario expansion and budget charging. Scenario queries may
    /// run concurrently in rule-engine workers; holding this lock from budget
    /// reservation through accounting prevents concurrent queries from jointly
    /// exceeding the model-wide limit.
    scenario_expansion_lock: Mutex<()>,
    /// Set once when scenario expansion omits at least one possible scenario,
    /// whether because a per-value product or the remaining model-wide budget
    /// was exhausted. The flag is monotonic and produces one advisory later.
    scenario_expansion_curtailed: AtomicBool,
}

/// Values used for AWS pseudo parameters (Ref AWS::Region, AWS::AccountId, ...) when
/// resolving the template; any field left unset falls back to a sensible default.
#[derive(Debug, Clone, Default, Serialize, serde::Deserialize)]
#[cfg_attr(feature = "wasm-bindings", derive(tsify::Tsify))]
#[cfg_attr(feature = "wasm-bindings", tsify(from_wasm_abi))]
#[cfg_attr(feature = "uniffi-bindings", derive(uniffi::Record))]
#[serde(rename_all = "camelCase")]
pub struct PseudoParameterOverrides {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub account_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub notification_arns: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub partition: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub region: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub stack_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub stack_name: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[cfg_attr(feature = "uniffi-bindings", uniffi(default))]
    pub url_suffix: Option<String>,
}

fn is_account_id_shaped(value: &str) -> bool {
    value.len() == 12 && value.bytes().all(|b| b.is_ascii_digit())
}

fn is_partition_shaped(value: &str) -> bool {
    value.starts_with("aws")
}

impl PseudoParameterOverrides {
    pub fn region(&self) -> &str {
        self.region.as_deref().unwrap_or(DEFAULT_REGION)
    }

    pub fn get(&self, name: &str) -> Option<String> {
        match name {
            PSEUDO_ACCOUNT_ID => Some(self.account_id.clone().unwrap_or_else(|| DEFAULT_ACCOUNT_ID.into())),
            PSEUDO_NOTIFICATION_ARNS => Some(self.notification_arns.clone().unwrap_or_else(|| {
                let r = self.region();
                let p = self.partition.as_deref().unwrap_or_else(|| partition_for_region(r));
                format!("arn:{}:sns:{}:{}:notification", p, r, DEFAULT_ACCOUNT_ID)
            })),
            PSEUDO_PARTITION => {
                Some(self.partition.clone().unwrap_or_else(|| partition_for_region(self.region()).into()))
            }
            PSEUDO_REGION => Some(self.region().to_string()),
            PSEUDO_STACK_ID => Some(self.stack_id.clone().unwrap_or_else(|| {
                let r = self.region();
                let p = self.partition.as_deref().unwrap_or_else(|| partition_for_region(r));
                format!(
                    "arn:{}:cloudformation:{}:{}:stack/{}/51af3dc0-da77-11e4-872e-1234567db123",
                    p, r, DEFAULT_ACCOUNT_ID, DEFAULT_STACK_NAME
                )
            })),
            PSEUDO_STACK_NAME => Some(self.stack_name.clone().unwrap_or_else(|| DEFAULT_STACK_NAME.into())),
            PSEUDO_URL_SUFFIX => {
                Some(self.url_suffix.clone().unwrap_or_else(|| url_suffix_for_region(self.region()).into()))
            }
            _ => None,
        }
    }

    #[must_use]
    pub fn invalid_overrides(&self) -> Vec<String> {
        let mut problems = Vec::new();
        if let Some(account_id) = self.account_id.as_deref()
            && !is_account_id_shaped(account_id)
        {
            problems.push(format!("{PSEUDO_ACCOUNT_ID} override '{account_id}' is not a 12-digit AWS account ID"));
        }
        if let Some(partition) = self.partition.as_deref()
            && !is_partition_shaped(partition)
        {
            problems.push(format!(
                "{PSEUDO_PARTITION} override '{partition}' is not a valid partition \
                 (expected an 'aws'-family value such as aws, aws-cn, or aws-us-gov)"
            ));
        }
        problems
    }

    /// Returns the user-supplied override for `name` only when the caller
    /// explicitly set the corresponding field. Auto-derived defaults - e.g. the
    /// commercial-vs-cn-vs-gov partition implied by `region` - are *not*
    /// returned here.
    ///
    /// The satisfiability solver uses this to decide whether a pseudo-parameter
    /// is a constant (user pinned its value) or a free variable ranging over
    /// the literals it is compared against plus a sentinel for "any other
    /// value". `get` always returns a default and would force the solver to
    /// treat every pseudo-parameter as a constant, producing false-positive
    /// "unreachable branch" diagnostics for templates that branch on
    /// `AWS::Partition`, `AWS::Region`, etc. - see `ConditionModel::eval_value_concrete`.
    pub fn fixed_value(&self, name: &str) -> Option<String> {
        match name {
            PSEUDO_ACCOUNT_ID => self.account_id.clone(),
            PSEUDO_NOTIFICATION_ARNS => self.notification_arns.clone(),
            PSEUDO_PARTITION => self.partition.clone(),
            PSEUDO_REGION => self.region.clone(),
            PSEUDO_STACK_ID => self.stack_id.clone(),
            PSEUDO_STACK_NAME => self.stack_name.clone(),
            PSEUDO_URL_SUFFIX => self.url_suffix.clone(),
            _ => None,
        }
    }
}

#[derive(Default)]
pub struct ParseConfig {
    pub parameters: HashMap<String, String>,
    pub pseudo_parameters: PseudoParameterOverrides,
}

#[must_use]
pub struct ParseResult {
    pub model: SemanticModel,
    /// Wall-clock time spent building the model, in milliseconds.
    pub model_build_ms: f64,
}

impl SemanticModel {
    pub fn from_bytes(bytes: &[u8]) -> Result<Self, ParseError> {
        Self::parse(bytes, Default::default()).map(|r| r.model)
    }

    pub fn parse(bytes: &[u8], config: ParseConfig) -> Result<ParseResult, ParseError> {
        if bytes.len() > MAX_TEMPLATE_SIZE_BYTES {
            return Err(ParseError {
                message: format!("Template exceeds maximum size of {}MB", MAX_TEMPLATE_SIZE_BYTES / (1024 * 1024)),
                line: None,
                column: None,
            });
        }

        let total_start = web_time::Instant::now();

        info!("Phase 1: Parsing IR ({} bytes)", bytes.len());
        let mut ir = crate::parser::parse(bytes)?;
        let foreach_diagnostics = crate::transform_expansion::expand_language_extensions(&mut ir);
        let condition_shape_diagnostics =
            crate::parser::condition_shape::validate_condition_bodies(&ir.arena, ir.conditions, &ir.span_index);
        let (parameters, parameter_diagnostics) = extract_parameters(&ir);
        // A parameter's definition can reference another parameter (e.g. a
        // Default given as `!Ref OtherParam`). Such a reference still counts as
        // usage, so collect the parameter names referenced from within the
        // Parameters section to feed the unused-parameter check.
        let params_referenced_in_definitions = collect_parameter_definition_refs(&ir, &parameters);
        let (mappings, mapping_diagnostics) = extract_mappings(&ir);

        // Create the shared budget tracker early so it is available before any
        // condition-model or resolver work can exhaust a budget.
        let budget_tracker = Arc::new(BudgetTracker::new());

        let mut conditions = ConditionModel::from_ir_with_tracker(
            &ir,
            &parameters,
            &config.pseudo_parameters,
            &mappings,
            &config.parameters,
            Arc::clone(&budget_tracker),
        );

        let resource_ids: Vec<String> = if ir.resources != NULL_REF {
            ir.arena
                .as_map(ir.resources)
                .map(|entries| entries.iter().map(|(k, _)| k.clone()).collect())
                .unwrap_or_default()
        } else {
            vec![]
        };
        let output_count =
            if ir.outputs != NULL_REF { ir.arena.as_map(ir.outputs).map(|e| e.len()).unwrap_or(0) } else { 0 };
        info!("Phase 2: Resolving {} resources, {} outputs", resource_ids.len(), output_count);

        let mut resolver = Resolver::new(
            &ir.arena,
            &parameters,
            &mappings,
            resource_ids.iter().cloned().collect(),
            &config.parameters,
            &config.pseudo_parameters,
        );
        let mut resources = HashMap::new();
        let mut lifecycle_attribute_nodes = HashMap::new();
        let mut invalid_resource_conditions = HashSet::new();
        if ir.resources != NULL_REF
            && let Some(entries) = ir.arena.as_map(ir.resources)
        {
            // Pre-scan: record each resource's DefinitionSubstitutions keys so
            // definition placeholders can be checked for membership per variable.
            for (rname, rnode) in entries {
                if let Some(props) = ir.arena.as_map(*rnode) {
                    for (key, val) in props {
                        if key == KEY_PROPERTIES
                            && let Some(prop_entries) = ir.arena.as_map(*val)
                            && let Some((_, subs_ref)) =
                                prop_entries.iter().find(|(k, _)| k == "DefinitionSubstitutions")
                            && let Some(subs) = ir.arena.as_map(*subs_ref)
                        {
                            resolver
                                .def_subs_resources
                                .entry(rname.clone())
                                .or_default()
                                .extend(subs.iter().map(|(k, _)| k.clone()));
                        }
                    }
                }
            }
            for (name, node_ref) in entries.iter().cloned() {
                if let Some(resource_entries) = ir.arena.as_map(node_ref) {
                    for attribute in
                        [KEY_CREATION_POLICY, KEY_UPDATE_POLICY, KEY_DELETION_POLICY, KEY_UPDATE_REPLACE_POLICY]
                    {
                        if let Some((_, value_ref)) = resource_entries.iter().find(|(key, _)| key == attribute) {
                            lifecycle_attribute_nodes.insert((name.clone(), attribute.to_string()), *value_ref);
                        }
                    }
                }
                if invalid_resource_condition_ref(&ir.arena, node_ref).is_some() {
                    invalid_resource_conditions.insert(name.clone());
                }
                // Validate resource body shape before resolution. `Fn::ForEach::`
                // synthetic IDs are expanded by the language-extensions transform
                // and do not have standard resource shapes.
                if name != FN_TRANSFORM && !name.starts_with(FN_FOR_EACH_KEY_PREFIX) {
                    let has_sam_transform = ir.transforms.iter().any(|t| t == TRANSFORM_SERVERLESS);
                    let shape_defects =
                        validate_resource_shape(&ir.arena, &name, node_ref, has_sam_transform, &ir.span_index);
                    resolver.diagnostics.extend(shape_defects);
                }
                resolver.set_current_resource(&name);
                let resolved = resolve_resource(&ir.arena, &name, node_ref, &mut resolver);
                resources.insert(name.clone(), resolved);
            }
        }
        for (name, res) in &resources {
            debug!(
                "Resolved '{}' ({}): {} properties, {} edges, condition={:?}",
                name,
                res.resource_type,
                res.properties.len(),
                resolver.edges.iter().filter(|e| e.source_resource == *name).count(),
                res.condition
            );
        }

        let mut outputs = HashMap::new();
        if ir.outputs != NULL_REF
            && let Some(entries) = ir.arena.as_map(ir.outputs)
        {
            for (name, node_ref) in entries.iter().cloned() {
                resolver.set_current_resource(&format!("{}{}", OUTPUT_PSEUDO_RESOURCE_PREFIX, name));
                resolver.set_current_path(&format!("Outputs/{}/Value", name));
                let resolved = resolve_output(&ir.arena, node_ref, &mut resolver);
                outputs.insert(name.clone(), resolved);
            }
        }

        // Walk the Rules section so that `Ref`/`Fn::Sub`/`Fn::ValueOf` etc.
        // appearing inside rule conditions and assertions emit reference
        // edges. Without this pass, parameters used only in Rules-section
        // assertions would appear unreferenced to downstream rule checks.
        if ir.rules != NULL_REF
            && let Some(rule_entries) = ir.arena.as_map(ir.rules)
        {
            for (rule_name, rule_node) in rule_entries.iter().cloned() {
                resolver.set_current_resource(&format!("{}{}", RULE_PSEUDO_RESOURCE_PREFIX, rule_name));
                let cond_ref = ir.arena.map_get(rule_node, KEY_RULE_CONDITION);
                if let Some(cond_ref) = cond_ref {
                    resolver.set_current_path(&format!("Rules/{}/{}", rule_name, KEY_RULE_CONDITION));
                    resolver.resolve_node(cond_ref);
                }
                let assertions_ref = ir.arena.map_get(rule_node, KEY_ASSERTIONS);
                if let Some(assertions_ref) = assertions_ref
                    && let Some(assertion_items) = ir.arena.as_list(assertions_ref)
                {
                    for (idx, item_ref) in assertion_items.to_vec().iter().enumerate() {
                        if let Some(assert_ref) = ir.arena.map_get(*item_ref, KEY_ASSERT) {
                            resolver.set_current_path(&format!(
                                "Rules/{}/{}/{}/{}",
                                rule_name, KEY_ASSERTIONS, idx, KEY_ASSERT
                            ));
                            resolver.resolve_node(assert_ref);
                        }
                    }
                }
            }
        }

        info!("Phase 3: Building reference graph from {} resolver edges", resolver.edges.len());

        // Register inline conditions (from IfExpr) into the condition model in
        // one batch, so the derived mutex/implication passes run once. Preserve
        // their invalid authored origin separately from their best-effort model.
        let invalid_inline_conditions = resolver.inline_conditions.iter().map(|(name, _)| name.clone()).collect();
        conditions.register_inline_batch(resolver.inline_conditions.drain(..));

        // Collect every mapping name referenced by an Fn::FindInMap anywhere in
        // the template (resources, outputs, conditions, ForEach bodies). A literal
        // first argument names a specific mapping; a non-literal one (e.g. a nested
        // Fn::FindInMap or Ref) means the referenced mapping can't be determined
        // statically, which disables the unused-mapping check.
        let mut find_in_map_names: std::collections::HashSet<String> = std::collections::HashSet::new();
        let mut has_dynamic_findinmap_name = false;
        for idx in 0..ir.arena.len() {
            if let Node::Intrinsic(IntrinsicFn::FindInMap(map_name_ref, _, _, _)) = ir.arena.node(idx as NodeRef) {
                match ir.arena.as_str(*map_name_ref) {
                    Some(name) => {
                        find_in_map_names.insert(name.to_string());
                    }
                    None => has_dynamic_findinmap_name = true,
                }
            }
        }

        let resolution_sources = resolver.resolution_sources();
        let value_nodes = resolver.value_nodes();
        let resolver_depth_exceeded = resolver.depth_exceeded;
        let resolver_enum_expansion_exceeded = resolver.enum_expansion_exceeded;
        let mut all_edges = resolver.edges;
        for (id, res) in &resources {
            for dep in &res.depends_on {
                all_edges.push(ResolverEdge {
                    source_resource: id.clone(),
                    source_path: KEY_DEPENDS_ON.to_string(),
                    target: dep.clone(),
                    kind: RefKind::DependsOn,
                    span: UNKNOWN_SPAN,
                    condition_context: None,
                });
            }
        }
        let graph = ReferenceGraph::build(all_edges, &resource_ids);

        let intrinsic_source_paths = build_intrinsic_source_paths(&resolution_sources, &graph);
        let pseudo_parameter_source_paths = build_pseudo_parameter_source_paths(&ir.arena, &value_nodes);
        let conditional_branch_first_descendant_spans = build_conditional_branch_first_descendant_spans(&ir.span_index);

        let mut resources_by_type: HashMap<String, Vec<String>> = HashMap::new();
        for (id, res) in &resources {
            resources_by_type.entry(res.resource_type.clone()).or_default().push(id.clone());
        }
        // Deterministic iteration across engines and runs: resource IDs within each
        // resource type are returned in template-declaration order (ascending by source
        // line/column). HashMap iteration is otherwise non-deterministic and was causing
        // per-pair diagnostic rules (e.g. subnet overlap) to attribute findings to
        // different resources on different runs.
        for ids in resources_by_type.values_mut() {
            ids.sort_by_key(|id| {
                let key = format!("Resources/{}", id);
                ir.span_index.get(key.as_str()).map(|s| (s.start_line, s.start_column)).unwrap_or((u32::MAX, u32::MAX))
            });
        }

        let mut diagnostics = ir.diagnostics;
        diagnostics.extend(foreach_diagnostics);
        diagnostics.extend(condition_shape_diagnostics);
        diagnostics.extend(mapping_diagnostics);
        diagnostics.extend(parameter_diagnostics);

        diagnostics.extend(crate::nesting::validate_intrinsic_nesting(&ir.arena));
        diagnostics.extend(crate::intrinsic_arg_shapes::validate_intrinsic_arg_shapes(&ir.arena, &ir.transforms));
        diagnostics.extend(crate::lang_ext_shapes::validate_lang_ext_parameter_shapes(&ir.arena, &ir.transforms));
        diagnostics.extend(crate::language_extensions::validate_language_extensions(&ir.arena, &ir.transforms));
        let lifecycle_resource_conditions: HashMap<String, String> = resources
            .iter()
            .filter_map(|(resource_id, resource)| {
                resource.condition.as_ref().map(|condition| (resource_id.clone(), condition.clone()))
            })
            .collect();
        diagnostics.extend(crate::language_extensions::validate_lifecycle_intrinsics(
            &ir.arena,
            &conditions,
            &lifecycle_attribute_nodes,
            &lifecycle_resource_conditions,
        ));
        diagnostics.extend(crate::dynamic_ref::validate_dynamic_references(&ir.arena, ir.resources));

        let mut fn_if_conditions: Vec<String> = Vec::new();
        for idx in 0..ir.arena.len() {
            match ir.arena.node(idx as NodeRef) {
                Node::Intrinsic(IntrinsicFn::If(cond_name, _, _)) => {
                    fn_if_conditions.push(cond_name.clone());
                    // Inside a Conditions-section body, Fn::If is not a valid
                    // condition function at all - that is the not-a-boolean
                    // finding's territory, so the name of a function that is
                    // itself rejected there is not checked.
                    let in_conditions_body =
                        ir.arena.get(idx as NodeRef).path.split('/').next() == Some(SECTION_CONDITIONS);
                    if !in_conditions_body && !conditions.conditions.contains_key(cond_name) {
                        // A single owner for the undefined-Fn::If-condition
                        // finding. Emitting it here (during the arena scan, which
                        // sees every Fn::If regardless of nesting) rather than in
                        // each engine keeps the two engines identical and covers
                        // the no-Conditions-section case, where a condition-name
                        // reference is still invalid.
                        let condition_path = format!("{}/{}/0", ir.arena.get(idx as NodeRef).path, FN_IF);
                        diagnostics.push(crate::make_parse_defect_at(
                            "E1028",
                            format!("Fn::If condition '{}' does not exist in Conditions section", cond_name),
                            ir.span_index
                                .get(&condition_path)
                                .copied()
                                .unwrap_or_else(|| ir.arena.span(idx as NodeRef)),
                            &condition_path,
                        ));
                    }
                }
                // A structurally malformed Fn::If (wrong arity, wrong type) is
                // rejected by the parser and left as a plain `Fn::If` map node
                // rather than an `IntrinsicFn::If`. Its condition is still
                // referenced, so collect the name here too - otherwise the
                // unused-condition check would wrongly flag a condition that the
                // template does reference.
                Node::Map(entries) if entries.len() == 1 && entries[0].0 == FN_IF => {
                    if let Some(first) = ir.arena.as_list(entries[0].1).and_then(|items| items.first())
                        && let Some(cond_name) = ir.arena.as_str(*first)
                    {
                        fn_if_conditions.push(cond_name.to_string());
                        // The structure error is reported separately; the
                        // undefined condition name is its own finding - a
                        // malformed Fn::If gets both. Conditions-section
                        // bodies are excluded for the same reason as the
                        // well-formed arm.
                        let in_conditions_body =
                            ir.arena.get(idx as NodeRef).path.split('/').next() == Some(SECTION_CONDITIONS);
                        if !in_conditions_body && !conditions.conditions.contains_key(cond_name) {
                            diagnostics.push(crate::make_parse_defect_at(
                                "E1028",
                                format!("Fn::If condition '{}' does not exist in Conditions section", cond_name),
                                ir.arena.span(*first),
                                &ir.arena.get(*first).path,
                            ));
                        }
                    }
                }
                _ => {}
            }
        }
        fn_if_conditions.sort();
        fn_if_conditions.dedup();
        let mut output_empty_joins: Vec<String> = Vec::new();
        for (key, joins) in &resolver.empty_joins {
            if key.starts_with(OUTPUT_PSEUDO_RESOURCE_PREFIX) || key == OUTPUTS_PSEUDO_RESOURCE {
                output_empty_joins.extend(joins.iter().cloned());
            }
        }

        // Raw pseudo-parameter strings in the Outputs section. Such findings are
        // collected against the `__output__`/`__outputs__` pseudo-resources,
        // which are filtered out of the serialized model the engines scan, so
        // the engine rule never sees them. Emit them here (both engines share
        // this parse-time output) so a pseudo-parameter used as a plain string
        // in an output Value is reported the same as one in a resource.
        {
            let mut output_raw_pseudo: Vec<(String, String)> = Vec::new();
            for (key, entries) in &resolver.raw_pseudo_params {
                if key.starts_with(OUTPUT_PSEUDO_RESOURCE_PREFIX) || key == OUTPUTS_PSEUDO_RESOURCE {
                    for (path, value) in entries {
                        output_raw_pseudo.push((path.clone(), value.clone()));
                    }
                }
            }
            output_raw_pseudo.sort();
            for (path, value) in output_raw_pseudo {
                diagnostics.push(crate::make_parse_defect(
                    "W1054",
                    format!(
                        "Found a string '{}' that appears to be a pseudo parameter reference; use 'Ref: {}' instead",
                        value, value
                    ),
                    ir.span_index.get(&path).copied().unwrap_or(UNKNOWN_SPAN),
                ));
            }
        }

        // Raw pseudo-parameter strings in parameter Default values: a Default
        // that is exactly a pseudo-parameter string (e.g. "AWS::Region") is
        // almost certainly a mistake. Parameters are not walked by the resolver,
        // so check the raw default strings directly.
        {
            let mut param_names: Vec<&String> = parameters.keys().collect();
            param_names.sort();
            for pname in param_names {
                if let Some(default) = &parameters[pname].default
                    && PSEUDO_PARAMETERS.contains(&default.as_str())
                {
                    diagnostics.push(crate::make_parse_defect(
                        "W1054",
                        format!("Found a string '{}' that appears to be a pseudo parameter reference; use 'Ref: {}' instead", default, default),
                        ir.span_index.get(&format!("Parameters/{}/Default", pname)).copied().unwrap_or(UNKNOWN_SPAN),
                    ));
                }
            }
        }

        // Raw pseudo-parameter strings in Mappings values and Conditions.
        // Neither section is walked by the resolver (it only visits resources
        // and outputs), so scan their string nodes directly. `Ref` targets are
        // not string nodes in the arena, so `Ref: AWS::Region` is naturally
        // exempt - only a pseudo-parameter used as plain string data fires.
        for idx in 0..ir.arena.len() {
            let spanned = ir.arena.get(idx as NodeRef);
            let Node::String(s) = &spanned.node else {
                continue;
            };
            let section = spanned.path.split('/').next().unwrap_or("");
            // A string that is a `Ref` target (path ends in the Ref key, e.g. a
            // raw `{"Ref": "AWS::Region"}` map the parser could not
            // canonicalize) is already a reference - only plain string *data*
            // warrants the use-Ref-instead advice.
            let is_ref_target = spanned.path.rsplit('/').next() == Some(FN_REF);
            if (section == SECTION_MAPPINGS || section == SECTION_CONDITIONS)
                && !is_ref_target
                && PSEUDO_PARAMETERS.contains(&s.as_str())
            {
                diagnostics.push(crate::make_parse_defect(
                    "W1054",
                    format!(
                        "Found a string '{}' that appears to be a pseudo parameter reference; use 'Ref: {}' instead",
                        s, s
                    ),
                    spanned.span,
                ));
            }
        }

        // Unsubstituted `${Variable}` strings in the Outputs section. Like the
        // raw-pseudo-parameter case above, these are collected against the
        // `__output__` pseudo-resources that never reach the engines, so the
        // Sub-needed finding must be emitted here.
        {
            let mut output_unsubstituted: Vec<(String, String)> = Vec::new();
            for (key, entries) in &resolver.unsubstituted_variables {
                if key.starts_with(OUTPUT_PSEUDO_RESOURCE_PREFIX) || key == OUTPUTS_PSEUDO_RESOURCE {
                    for (path, variable) in entries {
                        output_unsubstituted.push((path.clone(), variable.clone()));
                    }
                }
            }
            output_unsubstituted.sort();
            for (path, variable) in output_unsubstituted {
                diagnostics.push(crate::make_parse_defect(
                    "E1029",
                    format!("Found an embedded parameter '{}' outside of an 'Fn::Sub' at {}", variable, path),
                    ir.span_index.get(&path).copied().unwrap_or(UNKNOWN_SPAN),
                ));
            }
        }

        diagnostics.extend(resolver.diagnostics);
        // SAM handling keys on the exact transform id, matching how the engines
        // detect the transform. A substring match would misclassify a non-SAM
        // transform whose name merely contains "Serverless" (e.g. a typo'd date
        // or a custom macro) as SAM, running the transform-error validators and
        // suppressing the correct "serverless type without transform" finding.
        let is_sam = ir.transforms.iter().any(|t| t == TRANSFORM_SERVERLESS);
        for d in graph.cycle_diagnostics(&ir.span_index) {
            if is_sam && sam::cycle_involves_sam_diagnostic(&d, &resources) {
                continue;
            }
            diagnostics.push(d);
        }
        for (cond_name, always_val) in conditions.tautological_equals() {
            let result_str = if always_val { "True" } else { "False" };
            diagnostics.push(crate::make_parse_defect_at(
                "W8003",
                format!("Fn::Equals in condition '{}' will always return {}", cond_name, result_str),
                ir.span_index.get(&format!("Conditions/{}", cond_name)).copied().unwrap_or(UNKNOWN_SPAN),
                &format!("Conditions/{}", cond_name),
            ));
        }
        // A `Condition:` key that names a condition absent from the Conditions
        // section is reported here - a distinct rule for the resource case and
        // the output case, since they are separate concerns. Emitting both
        // during model build anchors each at its own source location and keeps
        // the two engines identical. Names are sorted for deterministic
        // ordering.
        {
            let mut resource_ids_sorted: Vec<&String> = resources.keys().collect();
            resource_ids_sorted.sort();
            for rid in resource_ids_sorted {
                if let Some(cond) = &resources[rid].condition
                    && !conditions.conditions.contains_key(cond)
                {
                    diagnostics.push(crate::make_parse_defect_for_resource(
                        "E8002",
                        format!("Condition '{}' referenced by resource '{}' is not defined", cond, rid),
                        ir.span_index.get(&format!("Resources/{}", rid)).copied().unwrap_or(UNKNOWN_SPAN),
                        rid,
                    ));
                }
            }
            let mut output_names_sorted: Vec<&String> = outputs.keys().collect();
            output_names_sorted.sort();
            for oname in output_names_sorted {
                if let Some(cond) = &outputs[oname].condition
                    && !conditions.conditions.contains_key(cond)
                {
                    diagnostics.push(crate::make_parse_defect_for_resource(
                        "E6005",
                        format!("Condition '{}' referenced by output '{}' is not defined", cond, oname),
                        ir.span_index.get(&format!("Outputs/{}", oname)).copied().unwrap_or(UNKNOWN_SPAN),
                        oname,
                    ));
                }
            }
        }
        for (owner, undefined_ref) in conditions.undefined_condition_refs() {
            // Synthetic conditions (`__`-prefixed, inserted for inline Fn::If and
            // Rules-section assertions) are internal; never surface their names.
            if owner.starts_with("__") || undefined_ref.starts_with("__") {
                continue;
            }
            diagnostics.push(crate::make_parse_defect(
                "E8007",
                format!("Condition '{}' references undefined condition '{}'", owner, undefined_ref),
                ir.span_index.get(&format!("Conditions/{}", owner)).copied().unwrap_or(UNKNOWN_SPAN),
            ));
        }
        for cycle in conditions.detect_cycles() {
            let cycle_desc = cycle.join(" -> ");
            let first = &cycle[0];
            diagnostics.push(crate::make_parse_defect(
                "E9106",
                format!("Circular dependency in conditions: {}", cycle_desc),
                ir.span_index.get(&format!("Conditions/{}", first)).copied().unwrap_or(UNKNOWN_SPAN),
            ));
        }
        for (first, other) in conditions.detect_equivalent_conditions() {
            diagnostics.push(crate::make_parse_defect(
                "W9053",
                format!("Condition '{}' is equivalent to condition '{}' - consider consolidating", other, first),
                ir.span_index.get(&format!("Conditions/{}", other)).copied().unwrap_or(UNKNOWN_SPAN),
            ));
        }
        // The satisfiability-budget-exhaustion advisory is deliberately NOT
        // emitted here: almost every satisfiability query runs later, during
        // engine rule evaluation, so the budget-exhausted set is still empty at
        // model-build time. The validation engine emits it after rule
        // evaluation, once those queries have run.
        let model_build_ms = total_start.elapsed().as_secs_f64() * 1000.0;

        if !diagnostics.is_empty() {
            warn!("{} parse-time diagnostics", diagnostics.len());
        }
        let total_props: usize = resources.values().map(|r| r.properties.len()).sum();
        info!(
            "Semantic model: {} resources ({} types, {} total properties), {} outputs, {} edges, {} conditions, {} diagnostics",
            resources.len(),
            resources_by_type.len(),
            total_props,
            outputs.len(),
            graph.edges.len(),
            conditions.conditions.len(),
            diagnostics.len()
        );

        let template_metadata =
            if ir.template_metadata != NULL_REF { Some(node_to_json(&ir.arena, ir.template_metadata)) } else { None };
        let rules = if ir.rules != NULL_REF { Some(node_to_json(&ir.arena, ir.rules)) } else { None };
        let parsed_rules = parse_rules(&rules, &ir.arena, ir.rules);
        let rule_implications: Vec<(String, crate::ir::NodeRef, Vec<crate::ir::NodeRef>)> = parsed_rules
            .iter()
            .map(|r| {
                let assertion_nodes: Vec<crate::ir::NodeRef> = r.assertions.iter().map(|a| a.assert_node).collect();
                (r.name.clone(), r.condition_node, assertion_nodes)
            })
            .collect();
        if !rule_implications.is_empty() {
            conditions.register_rule_implications(&ir.arena, &rule_implications);
        }
        let rule_diagnostics = crate::rules::validate_rules(&rules, &ir.arena, ir.rules);
        diagnostics.extend(rule_diagnostics);
        let sam_globals = sam::extract_sam_globals(&ir.arena, ir.globals);
        if !sam_globals.is_empty() {
            sam::apply_sam_globals(&mut resources, &sam_globals, &mut ir.span_index);
        }
        let sam_implicit_resources =
            if is_sam { sam::collect_sam_implicit_resources(&resources) } else { HashSet::new() };
        let globals_param_refs =
            if is_sam { sam::collect_globals_param_refs(&ir.arena, ir.globals) } else { Vec::new() };
        if is_sam {
            let parameter_names: HashSet<String> = parameters.keys().cloned().collect();
            diagnostics.extend(sam::collect_transform_errors(
                &ir.arena,
                ir.resources,
                ir.globals,
                &resources,
                &parameter_names,
                &ir.span_index,
            ));
        }
        let is_cdk = resources_by_type.contains_key(CDK_METADATA_TYPE);

        if resolver_depth_exceeded {
            budget_tracker.record(BudgetKind::ResolverDepth);
        }
        if resolver_enum_expansion_exceeded {
            budget_tracker.record(BudgetKind::EnumExpansion);
        }
        // Invalid Ref targets in Outputs. The resolver records these under the
        // `__output__<Name>` pseudo-resource key, but the engines only see
        // top-level edges (which are not created for unknown Ref targets). Emit
        // them here as parse-time diagnostics so both engines surface them.
        // Skip when MODULE resources or unexpanded ForEach are present, since
        // both can synthesize resource names unknown at parse time.
        {
            let has_module_or_foreach = resources.values().any(|r| r.resource_type.ends_with("::MODULE"))
                || resources.keys().any(|k| k.contains("Fn::ForEach"));
            if !has_module_or_foreach {
                let mut output_invalid_refs: Vec<(String, String, String)> = Vec::new();
                for (key, entries) in &resolver.invalid_refs {
                    if let Some(output_name) = key.strip_prefix(OUTPUT_PSEUDO_RESOURCE_PREFIX) {
                        for (path, target) in entries {
                            if sam_implicit_resources.contains(target) {
                                continue;
                            }
                            output_invalid_refs.push((output_name.to_string(), path.clone(), target.clone()));
                        }
                    }
                }
                output_invalid_refs.sort();
                for (output_name, path, target) in &output_invalid_refs {
                    let mut defect =
                        ParseDefect::new(
                            "F6101",
                            format!(
                                "Ref '{}' does not reference a valid resource, parameter, or pseudo-parameter",
                                target,
                            ),
                        )
                        .location(ir.span_index.get(path).copied().unwrap_or(UNKNOWN_SPAN))
                        .phase(DefectPhase::Parse);
                    defect = defect.property_path(format!("Outputs.{}.Value.Ref", output_name));
                    diagnostics.push(defect);
                }
            }
        }

        Ok(ParseResult {
            model: SemanticModel {
                arena: ir.arena,
                root: ir.root,
                authored_template: OnceLock::new(),
                span_index: ir.span_index,
                conditional_branch_first_descendant_spans,
                format_version: ir.format_version,
                description: ir.description,
                transforms: ir.transforms,
                raw_top_level_keys: ir.raw_top_level_keys,
                template_metadata,
                rules,
                parsed_rules,
                parameters,
                mappings,
                conditions,
                resources,
                outputs,
                graph,
                resources_by_type,
                diagnostics,
                output_empty_joins,
                sam_globals,
                sam_implicit_resources,
                globals_param_refs,
                is_cdk,
                fn_if_conditions,
                find_in_map_names,
                params_referenced_in_definitions,
                has_dynamic_findinmap_name,
                resolution_sources,
                parameter_overrides: config.parameters.keys().cloned().collect(),
                intrinsic_source_paths,
                pseudo_parameter_source_paths,
                value_nodes,
                lifecycle_attribute_nodes,
                lifecycle_attribute_status_cache: Mutex::new(HashMap::new()),
                invalid_resource_conditions,
                invalid_inline_conditions,
                resolve_memo: Mutex::new(HashMap::new()),
                raw_scenario_memo: Mutex::new(HashMap::new()),
                properties_scenario_cache: Mutex::new(HashMap::new()),
                list_member_scenario_cache: Mutex::new(HashMap::new()),
                scenario_memo: Mutex::new(HashMap::new()),
                lifecycle_policy_scenario_cache: Mutex::new(HashMap::new()),
                scenario_combinations_used: AtomicU64::new(0),
                budget_tracker,
                scenario_expansion_lock: Mutex::new(()),
                scenario_expansion_curtailed: AtomicBool::new(false),
            },
            model_build_ms,
        })
    }

    pub fn resource(&self, id: &str) -> Option<&ResolvedResource> {
        self.resources.get(id)
    }

    /// The template as the author wrote it, as CloudFormation JSON with every
    /// intrinsic function in long form. This is the view an external evaluator
    /// that understands CloudFormation syntax - rather than this crate's resolved
    /// model - operates on. Rendered once and cached for the model's lifetime.
    pub fn authored_template_json(&self) -> &serde_json::Value {
        self.authored_template.get_or_init(|| render_authored_json(&self.arena, self.root))
    }

    /// Reports whether a lifecycle attribute can legally survive condition
    /// evaluation, and whether any reachable authored value has a non-object
    /// shape. Only `UpdatePolicy` permits `Fn::If` to remove the whole attribute
    /// with `AWS::NoValue`; illegal values on the other attributes remain present
    /// for downstream diagnostics.
    #[must_use]
    pub fn lifecycle_attribute_status(&self, resource_id: &str, attribute: &str) -> LifecycleAttributeStatus {
        if !matches!(
            attribute,
            KEY_CREATION_POLICY | KEY_UPDATE_POLICY | KEY_DELETION_POLICY | KEY_UPDATE_REPLACE_POLICY
        ) {
            return LifecycleAttributeStatus::default();
        }
        let cache_key = (resource_id.to_string(), attribute.to_string());
        let mut cache = self.lifecycle_attribute_status_cache.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        if let Some(status) = cache.get(&cache_key) {
            return status.clone();
        }

        let mut status = LifecycleAttributeStatus::default();
        if let Some(node) = self.lifecycle_attribute_nodes.get(&cache_key) {
            let mut assumptions = self
                .resources
                .get(resource_id)
                .and_then(|resource| resource.condition.as_ref())
                .map(|condition| vec![(condition.clone(), true)])
                .unwrap_or_default();
            collect_lifecycle_attribute_status(
                &self.arena,
                *node,
                &self.conditions,
                attribute == KEY_UPDATE_POLICY,
                &mut assumptions,
                &mut status,
            );
        }
        cache.insert(cache_key, status.clone());
        status
    }

    /// Expands a lifecycle policy into reachable values while preserving the
    /// assumptions for each branch. Whole dynamic leaves are deferred, while a
    /// list or object remains invalid regardless of dynamic values nested inside it.
    #[must_use]
    pub fn lifecycle_policy_scenarios(
        &self,
        resource_id: &str,
        attribute: &str,
    ) -> Vec<(serde_json::Value, HashMap<String, bool>)> {
        let cache_key = (resource_id.to_string(), attribute.to_string());
        let mut cache = self.lifecycle_policy_scenario_cache.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        if let Some(scenarios) = cache.get(&cache_key) {
            return scenarios.clone();
        }

        let scenarios = if !self.resource_condition_is_valid(resource_id) {
            Vec::new()
        } else {
            let resolved = match attribute {
                KEY_DELETION_POLICY => self.resources.get(resource_id).and_then(|r| r.deletion_policy.as_ref()),
                KEY_UPDATE_REPLACE_POLICY => {
                    self.resources.get(resource_id).and_then(|r| r.update_replace_policy.as_ref())
                }
                _ => None,
            };
            match resolved {
                None => Vec::new(),
                Some(value) => {
                    let initial_assumptions: HashMap<String, bool> = self
                        .resources
                        .get(resource_id)
                        .and_then(|resource| resource.condition.as_ref())
                        .map(|condition| [(condition.clone(), true)].into_iter().collect())
                        .unwrap_or_default();
                    let assumptions: Vec<_> =
                        initial_assumptions.iter().map(|(name, value)| (name.clone(), *value)).collect();
                    if !assumptions.is_empty() && !self.conditions.is_satisfiable(&assumptions) {
                        Vec::new()
                    } else {
                        self.collect_scenarios_with_budget_from(
                            value,
                            &initial_assumptions,
                            MAX_SCENARIO_COMBINATIONS,
                            MAX_TOTAL_SCENARIO_COMBINATIONS,
                        )
                        .0
                        .into_iter()
                        .filter_map(|(value, conditions)| {
                            let assumptions: Vec<_> =
                                conditions.iter().map(|(name, value)| (name.clone(), *value)).collect();
                            if !assumptions.is_empty() && !self.conditions.is_satisfiable(&assumptions) {
                                return None;
                            }
                            let value = match value {
                                ResolvedValue::Reference { .. }
                                | ResolvedValue::Dynamic { .. }
                                | ResolvedValue::TypedDynamic { .. } => return None,
                                other => crate::serialization::resolved_value_to_json(&other),
                            };
                            Some((value, conditions))
                        })
                        .collect()
                    }
                }
            }
        };
        cache.insert(cache_key, scenarios.clone());
        scenarios
    }

    pub fn resources_of_type(&self, type_name: &str) -> &[String] {
        self.resources_by_type.get(type_name).map(|v| v.as_slice()).unwrap_or(&[])
    }

    #[must_use]
    pub fn resource_condition_is_valid(&self, resource_id: &str) -> bool {
        !self.invalid_resource_conditions.contains(resource_id)
    }

    /// Groups resources that can simultaneously resolve to the same primary-identifier tuple.
    /// Each displayed tuple is paired with the ordered set of resources participating in at
    /// least one satisfiable collision for the corresponding comparison identity.
    #[must_use]
    pub fn primary_identifier_conflicts(
        &self,
        resource_type: &str,
        identifier_properties: &[String],
    ) -> Vec<(Vec<String>, BTreeSet<String>)> {
        let mut per_resource = Vec::new();
        for resource_id in self.resources_of_type(resource_type) {
            let scenarios = self.primary_identifier_scenarios(resource_id, identifier_properties);
            if !scenarios.is_empty() {
                per_resource.push((resource_id, scenarios));
            }
        }

        // When every scenario has one comparison identity, grouping cannot skip
        // any work. Keep the original tight traversal for that dense case so the
        // optimization adds no cursor or heap overhead where it has nothing to
        // prune. The preliminary linear scan is negligible beside the pairwise
        // traversal and establishes that this is exactly one identity group.
        let first_identity = per_resource
            .first()
            .and_then(|(_, scenarios)| scenarios.first())
            .map(|scenario| scenario.identity_tuple.as_slice());
        let has_one_identity = first_identity.is_some_and(|identity| {
            per_resource
                .iter()
                .flat_map(|(_, scenarios)| scenarios)
                .all(|scenario| scenario.identity_tuple.as_slice() == identity)
        });
        if has_one_identity {
            let mut conflicts: BTreeMap<Vec<String>, (Vec<String>, BTreeSet<String>)> = BTreeMap::new();
            for left_index in 0..per_resource.len() {
                for right_index in (left_index + 1)..per_resource.len() {
                    let (left_resource, left_scenarios) = &per_resource[left_index];
                    let (right_resource, right_scenarios) = &per_resource[right_index];
                    for left in left_scenarios {
                        for right in right_scenarios {
                            let mut assumptions = left.assumptions.clone();
                            assumptions.extend(right.assumptions.iter().cloned());
                            if assumptions.is_empty() || self.conditions.is_satisfiable(&assumptions) {
                                let (_, resources) = conflicts
                                    .entry(left.identity_tuple.clone())
                                    .or_insert_with(|| (left.display_tuple.clone(), BTreeSet::new()));
                                resources.insert((*left_resource).clone());
                                resources.insert((*right_resource).clone());
                            }
                        }
                    }
                }
            }
            return conflicts.into_values().collect();
        }

        // Two resources can only collide when their identifier tuples are equal, so
        // group scenarios by comparison identity and confine every comparison to a
        // group. A resource with a unique tuple forms a singleton group and does no
        // cross-resource or satisfiability work - the common case the earlier
        // all-pairs traversal paid for on every template.
        let mut candidates_by_identity: HashMap<&[String], Vec<(usize, usize)>> = HashMap::new();
        for (resource_index, (_, scenarios)) in per_resource.iter().enumerate() {
            for (scenario_index, scenario) in scenarios.iter().enumerate() {
                candidates_by_identity
                    .entry(scenario.identity_tuple.as_slice())
                    .or_default()
                    .push((resource_index, scenario_index));
            }
        }
        let mut cursors: Vec<IdentityGroupPairCursor> = candidates_by_identity
            .into_values()
            .filter_map(|candidates| IdentityGroupPairCursor::new(&candidates))
            .collect();

        let mut conflicts: BTreeMap<Vec<String>, (Vec<String>, BTreeSet<String>)> = BTreeMap::new();
        {
            let mut record_conflict = |comparison: PrimaryIdentifierComparisonOrder| {
                let (left_resource_index, right_resource_index, left_scenario_index, right_scenario_index) = comparison;
                let (left_resource, left_scenarios) = &per_resource[left_resource_index];
                let (right_resource, right_scenarios) = &per_resource[right_resource_index];
                let left = &left_scenarios[left_scenario_index];
                let right = &right_scenarios[right_scenario_index];
                let mut assumptions = left.assumptions.clone();
                assumptions.extend(right.assumptions.iter().cloned());
                if assumptions.is_empty() || self.conditions.is_satisfiable(&assumptions) {
                    let (_, resources) = conflicts
                        .entry(left.identity_tuple.clone())
                        .or_insert_with(|| (left.display_tuple.clone(), BTreeSet::new()));
                    resources.insert((*left_resource).clone());
                    resources.insert((*right_resource).clone());
                }
            };

            // Visit the surviving comparisons in the exact order the all-pairs
            // traversal reached them by merging the groups on comparison order. Every
            // satisfiability query then runs in the identical sequence and count,
            // leaving the model's shared, cumulative budget drawn down exactly as
            // before. A single group is already ordered, so it needs no merge and
            // keeps the all-duplicate case as cheap as the traversal it replaces.
            if cursors.len() <= 1 {
                if let Some(cursor) = cursors.pop() {
                    for comparison in cursor {
                        record_conflict(comparison);
                    }
                }
            } else {
                let mut frontier = BinaryHeap::with_capacity(cursors.len());
                for (cursor_index, cursor) in cursors.iter_mut().enumerate() {
                    if let Some(comparison) = cursor.next() {
                        frontier.push(Reverse((comparison, cursor_index)));
                    }
                }
                while let Some(Reverse((comparison, cursor_index))) = frontier.pop() {
                    record_conflict(comparison);
                    if let Some(next_comparison) = cursors[cursor_index].next() {
                        frontier.push(Reverse((next_comparison, cursor_index)));
                    }
                }
            }
        }
        conflicts.into_values().collect()
    }

    fn primary_identifier_scenarios(
        &self,
        resource_id: &str,
        identifier_properties: &[String],
    ) -> Vec<PrimaryIdentifierScenario> {
        if !self.resource_condition_is_valid(resource_id) {
            return Vec::new();
        }
        let base_assumptions = self
            .resources
            .get(resource_id)
            .and_then(|resource| resource.condition.as_deref())
            .map(|condition| vec![(condition.to_string(), true)])
            .unwrap_or_default();
        let mut scenarios = vec![PrimaryIdentifierScenario {
            identity_tuple: Vec::with_capacity(identifier_properties.len()),
            display_tuple: Vec::with_capacity(identifier_properties.len()),
            assumptions: base_assumptions,
        }];

        for property in identifier_properties {
            let path = format!("Properties.{property}");
            let property_scenarios = self.resolve_scenarios_json(resource_id, &path);
            let expression_identity =
                self.value_identity(resource_id, &path).filter(|identity| identity.starts_with("expr:"));
            let effective_scenarios: Vec<(String, String, HashMap<String, bool>)> = if property_scenarios.is_empty()
                || (expression_identity.is_some()
                    && property_scenarios.iter().all(|(_, conditions)| conditions.is_empty()))
            {
                expression_identity
                    .map(|identity| {
                        let display = identity.strip_prefix("expr:").unwrap_or(&identity).to_string();
                        vec![(identity, display, HashMap::new())]
                    })
                    .unwrap_or_default()
            } else {
                property_scenarios
                    .into_iter()
                    .filter_map(|(value, conditions)| {
                        if value.is_null() {
                            return None;
                        }
                        let display = value.as_str().map(ToOwned::to_owned).unwrap_or_else(|| value.to_string());
                        let identity = format!("value:{}", crate::value_identity::concrete_value_fingerprint(&value));
                        Some((identity, display, conditions))
                    })
                    .collect()
            };
            let mut next = Vec::new();
            for existing in &scenarios {
                for (identity, display, conditions) in &effective_scenarios {
                    let mut assumptions = existing.assumptions.clone();
                    let mut consistent = true;
                    for (condition, truth) in conditions {
                        if let Some((_, prior)) = assumptions.iter().find(|(name, _)| name == condition) {
                            if prior != truth {
                                consistent = false;
                                break;
                            }
                        } else {
                            assumptions.push((condition.clone(), *truth));
                        }
                    }
                    if consistent {
                        let mut identity_tuple = existing.identity_tuple.clone();
                        identity_tuple.push(identity.clone());
                        let mut display_tuple = existing.display_tuple.clone();
                        display_tuple.push(display.clone());
                        next.push(PrimaryIdentifierScenario { identity_tuple, display_tuple, assumptions });
                    }
                }
            }
            scenarios = next;
            if scenarios.is_empty() {
                break;
            }
        }

        scenarios.retain(|scenario| {
            scenario.identity_tuple.len() == identifier_properties.len()
                && (scenario.assumptions.is_empty() || self.conditions.is_satisfiable(&scenario.assumptions))
        });
        scenarios
    }

    #[must_use]
    pub fn condition_is_valid_for_reachability(&self, condition: &str) -> bool {
        !self.invalid_inline_conditions.contains(condition)
    }

    #[must_use]
    pub fn resolve(&self, resource_id: &str, path: &str) -> Option<&ResolvedValue> {
        let resource = self.resources.get(resource_id)?;
        resource.properties.get(path.strip_prefix("Properties.").unwrap_or(path))
    }

    #[must_use]
    pub fn resolve_deep(&self, resource_id: &str, path: &str) -> Option<ResolvedValue> {
        let memo_key = (resource_id.to_string(), path.to_string());
        if let Some(memoized) = self.resolve_memo.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).get(&memo_key)
        {
            return memoized.clone();
        }
        let resource = self.resources.get(resource_id)?;
        let prop_path = path.strip_prefix("Properties.").unwrap_or(path);
        let mut segments = prop_path.splitn(2, '.');
        let top_key = segments.next()?;
        let resolved = resource.properties.get(top_key)?;
        let result = match segments.next() {
            Some(r) if !r.is_empty() => resolved_value_at_path(resolved, r),
            _ => Some(resolved.clone()),
        };
        self.resolve_memo.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).insert(memo_key, result.clone());
        result
    }

    /// Returns the resolved value tree at `path` without expanding `Fn::If` or
    /// parameter-enum alternatives. Unlike [`Self::resolve_deep`], this also
    /// traverses a whole-`Properties` `Fn::If`; a path absent from one branch is
    /// represented by the dynamic placeholder produced by
    /// [`resolved_value_at_path`]. This lookup does not consume scenario budget.
    #[must_use]
    pub fn resolve_unexpanded(&self, resource_id: &str, path: &str) -> Option<ResolvedValue> {
        self.resolve_deep(resource_id, path)
            .or_else(|| self.resolve(resource_id, path).cloned())
            .or_else(|| self.resolve_via_properties_if(resource_id, path))
    }

    /// Returns scenarios for each list member at `path`, expanding every member
    /// independently under conditions that wrap the whole list. Conditions on
    /// sibling members never multiply, and repeated callers share one cached
    /// allocation. This structural query does not consume scenario budget.
    #[must_use]
    pub fn resolve_list_member_scenarios_shared(&self, resource_id: &str, path: &str) -> Arc<Vec<ListMemberScenario>> {
        let cache_key = (resource_id.to_string(), path.to_string());
        let mut cache = self.list_member_scenario_cache.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        if let Some(scenarios) = cache.get(&cache_key) {
            return Arc::clone(scenarios);
        }

        let mut member_scenarios = Vec::new();
        if let Some(value) = self.resolve_unexpanded(resource_id, path) {
            let mut next_container_scenario = 0;
            self.collect_list_member_scenarios(
                &value,
                &HashMap::new(),
                &mut next_container_scenario,
                &mut member_scenarios,
            );
        }
        let shared = Arc::new(member_scenarios);
        cache.insert(cache_key, Arc::clone(&shared));
        shared
    }

    /// Whether two distinct members of one reachable list alternative have the
    /// same concrete string value for `field_name`.
    #[must_use]
    pub fn duplicate_object_field_value_scenario(
        &self,
        resource_id: &str,
        path: &str,
        field_name: &str,
    ) -> Option<HashMap<String, bool>> {
        let mut alternatives_by_value: BTreeMap<String, Vec<(usize, usize, HashMap<String, bool>)>> = BTreeMap::new();
        for (container, index, value, conditions) in self.list_member_value_scenarios(resource_id, path) {
            let field_value = match &value {
                ResolvedValue::Map { entries } => {
                    entries.iter().find(|entry| entry.key == field_name).map(|entry| entry.value.clone())
                }
                ResolvedValue::Concrete { value } => value
                    .as_object()
                    .and_then(|object| object.get(field_name))
                    .cloned()
                    .map(|value| ResolvedValue::Concrete { value: value.into() }),
                _ => None,
            };
            let Some(field_value) = field_value else {
                continue;
            };
            let field_path = format!("{path}.{index}.{field_name}");
            let uses_authored_identity = contains_dynamic_resolved(&field_value)
                || self.scenario_uses_unresolved_parameter(resource_id, &field_path, &conditions)
                || self.scenario_uses_pseudo_parameter(resource_id, &field_path, &conditions);
            let comparison_key = if uses_authored_identity {
                let source_path = self
                    .scenario_source_path(resource_id, &field_path, &conditions)
                    .unwrap_or_else(|| field_path.clone());
                let Some(identity) = self.resolved_value_identity(resource_id, &source_path, &field_value) else {
                    continue;
                };
                identity
            } else {
                let rendered = resolved_value_to_json(&field_value);
                let Some(field_value) = rendered.as_str() else {
                    continue;
                };
                format!("literal:{field_value}")
            };
            alternatives_by_value.entry(comparison_key).or_default().push((container, index, conditions));
        }
        for alternatives in alternatives_by_value.values() {
            for (left_position, (left_container, left_index, left_conditions)) in alternatives.iter().enumerate() {
                for (right_container, right_index, right_conditions) in alternatives.iter().skip(left_position + 1) {
                    if self.conditions.satisfiability_budget_exhausted() {
                        return None;
                    }
                    if left_container == right_container
                        && left_index != right_index
                        && let Some(conditions) =
                            self.merge_reachable_conditions(resource_id, left_conditions, right_conditions)
                    {
                        return Some(conditions);
                    }
                }
            }
        }
        None
    }

    /// Reachable list members where `required_field` is absent while
    /// `trigger_field` is present.
    #[must_use]
    pub fn list_object_missing_required_field_scenarios(
        &self,
        resource_id: &str,
        path: &str,
        required_field: &str,
        trigger_field: &str,
    ) -> Vec<(usize, HashMap<String, bool>)> {
        let mut missing = Vec::new();
        for (_, index, value, conditions) in self.list_member_value_scenarios(resource_id, path) {
            let value = resolved_value_to_json(&value);
            let Some(object) = value.as_object() else {
                continue;
            };
            if object.get(trigger_field).is_some_and(|value| !value.is_null())
                && object.get(required_field).is_none_or(serde_json::Value::is_null)
            {
                missing.push((index, conditions));
            }
        }
        missing.sort_by(|(left_index, left_conditions), (right_index, right_conditions)| {
            left_index
                .cmp(right_index)
                .then_with(|| canonical_conditions(left_conditions).cmp(&canonical_conditions(right_conditions)))
        });
        missing.dedup_by(|left, right| left == right);
        missing
    }

    fn list_member_value_scenarios(
        &self,
        resource_id: &str,
        path: &str,
    ) -> Vec<(usize, usize, ResolvedValue, HashMap<String, bool>)> {
        let members = self.resolve_list_member_scenarios_shared(resource_id, path);
        let mut alternatives = Vec::new();
        for member in members.iter() {
            let (scenarios, _) = self.collect_scenarios_with_budget_from(
                &member.value,
                &member.conditions,
                MAX_SCENARIO_COMBINATIONS,
                MAX_TOTAL_SCENARIO_COMBINATIONS,
            );
            alternatives.extend(
                scenarios
                    .into_iter()
                    .map(|(value, conditions)| (member.container_scenario, member.index, value, conditions)),
            );
        }
        alternatives
    }

    /// Returns each source location that can contain two distinct, non-null
    /// list members in a deployment compatible with `base_conditions`.
    /// Dynamic member values count as present. Members from different outer list
    /// alternatives or different alternatives of one member never form a pair.
    #[must_use]
    pub fn list_member_pair_source_paths(
        &self,
        resource_id: &str,
        path: &str,
        base_conditions: &HashMap<String, bool>,
    ) -> BTreeSet<String> {
        let member_scenarios = self.resolve_list_member_scenarios_shared(resource_id, path);
        let container_scenarios: BTreeSet<usize> =
            member_scenarios.iter().map(|scenario| scenario.container_scenario).collect();
        let mut source_paths = BTreeSet::new();
        for container_scenario in container_scenarios {
            if self.conditions.satisfiability_budget_exhausted() {
                break;
            }
            let Some(container) =
                member_scenarios.iter().find(|scenario| scenario.container_scenario == container_scenario)
            else {
                continue;
            };
            let Some(base_and_container) =
                self.merge_reachable_conditions(resource_id, base_conditions, &container.container_conditions)
            else {
                continue;
            };
            'container_members: for (left_position, left) in member_scenarios.iter().enumerate() {
                if left.container_scenario != container_scenario || is_resolved_null(&left.value) {
                    continue;
                }
                let Some(base_and_left) =
                    self.merge_reachable_conditions(resource_id, &base_and_container, &left.conditions)
                else {
                    continue;
                };
                for right in member_scenarios.iter().skip(left_position + 1) {
                    if self.conditions.satisfiability_budget_exhausted() {
                        return source_paths;
                    }
                    if right.container_scenario != container_scenario
                        || left.index == right.index
                        || is_resolved_null(&right.value)
                    {
                        continue;
                    }
                    if let Some(conditions) =
                        self.merge_reachable_conditions(resource_id, &base_and_left, &right.conditions)
                    {
                        let source_path = self
                            .scenario_source_path(resource_id, path, &conditions)
                            .unwrap_or_else(|| path.to_string());
                        let value_branch_prefix = format!("{path}.{FN_IF}.");
                        source_paths.insert(if source_path.starts_with(&value_branch_prefix) {
                            path.to_string()
                        } else {
                            source_path
                        });
                        break 'container_members;
                    }
                }
            }
        }
        source_paths
    }

    fn merge_reachable_conditions(
        &self,
        resource_id: &str,
        left: &HashMap<String, bool>,
        right: &HashMap<String, bool>,
    ) -> Option<HashMap<String, bool>> {
        let mut merged = left.clone();
        for (condition, value) in right {
            if merged.get(condition).is_some_and(|existing| existing != value) {
                return None;
            }
            merged.insert(condition.clone(), *value);
        }
        if let Some(resource_condition) =
            self.resources.get(resource_id).and_then(|resource| resource.condition.as_ref())
        {
            if merged.get(resource_condition) == Some(&false) {
                return None;
            }
            merged.insert(resource_condition.clone(), true);
        }
        let assumptions: Vec<(String, bool)> =
            merged.iter().map(|(condition, value)| (condition.clone(), *value)).collect();
        (self.conditions.satisfiability(&assumptions) == Satisfiability::Satisfiable).then_some(merged)
    }

    fn collect_list_member_scenarios(
        &self,
        value: &ResolvedValue,
        assumptions: &HashMap<String, bool>,
        next_container_scenario: &mut usize,
        member_scenarios: &mut Vec<ListMemberScenario>,
    ) {
        match value {
            ResolvedValue::Conditional { condition, if_true, if_false } => match assumptions.get(condition) {
                Some(true) => {
                    self.collect_list_member_scenarios(if_true, assumptions, next_container_scenario, member_scenarios)
                }
                Some(false) => {
                    self.collect_list_member_scenarios(if_false, assumptions, next_container_scenario, member_scenarios)
                }
                None => {
                    let mut true_assumptions = assumptions.clone();
                    true_assumptions.insert(condition.clone(), true);
                    self.collect_list_member_scenarios(
                        if_true,
                        &true_assumptions,
                        next_container_scenario,
                        member_scenarios,
                    );
                    let mut false_assumptions = assumptions.clone();
                    false_assumptions.insert(condition.clone(), false);
                    self.collect_list_member_scenarios(
                        if_false,
                        &false_assumptions,
                        next_container_scenario,
                        member_scenarios,
                    );
                }
            },
            ResolvedValue::Enum { variants } => {
                for variant in variants {
                    self.collect_list_member_scenarios(variant, assumptions, next_container_scenario, member_scenarios);
                }
            }
            ResolvedValue::List { items } => {
                let container_scenario = *next_container_scenario;
                *next_container_scenario = (*next_container_scenario).saturating_add(1);
                for (index, item) in items.iter().enumerate() {
                    Self::collect_member_scenarios(
                        item,
                        assumptions,
                        container_scenario,
                        index,
                        assumptions,
                        member_scenarios,
                    );
                }
            }
            ResolvedValue::Concrete { value } => {
                if let Some(items) = value.as_array() {
                    let container_scenario = *next_container_scenario;
                    *next_container_scenario = (*next_container_scenario).saturating_add(1);
                    member_scenarios.extend(items.iter().enumerate().map(|(index, item)| ListMemberScenario {
                        container_scenario,
                        container_conditions: assumptions.clone(),
                        index,
                        value: ResolvedValue::Concrete { value: item.clone().into() },
                        conditions: assumptions.clone(),
                    }));
                }
            }
            ResolvedValue::Map { .. }
            | ResolvedValue::Reference { .. }
            | ResolvedValue::Dynamic { .. }
            | ResolvedValue::TypedDynamic { .. } => {}
        }
    }

    fn collect_member_scenarios(
        value: &ResolvedValue,
        assumptions: &HashMap<String, bool>,
        container_scenario: usize,
        index: usize,
        container_conditions: &HashMap<String, bool>,
        member_scenarios: &mut Vec<ListMemberScenario>,
    ) {
        match value {
            ResolvedValue::Conditional { condition, if_true, if_false } => match assumptions.get(condition) {
                Some(true) => Self::collect_member_scenarios(
                    if_true,
                    assumptions,
                    container_scenario,
                    index,
                    container_conditions,
                    member_scenarios,
                ),
                Some(false) => Self::collect_member_scenarios(
                    if_false,
                    assumptions,
                    container_scenario,
                    index,
                    container_conditions,
                    member_scenarios,
                ),
                None => {
                    let mut true_assumptions = assumptions.clone();
                    true_assumptions.insert(condition.clone(), true);
                    Self::collect_member_scenarios(
                        if_true,
                        &true_assumptions,
                        container_scenario,
                        index,
                        container_conditions,
                        member_scenarios,
                    );
                    let mut false_assumptions = assumptions.clone();
                    false_assumptions.insert(condition.clone(), false);
                    Self::collect_member_scenarios(
                        if_false,
                        &false_assumptions,
                        container_scenario,
                        index,
                        container_conditions,
                        member_scenarios,
                    );
                }
            },
            ResolvedValue::Enum { variants } => {
                for variant in variants {
                    Self::collect_member_scenarios(
                        variant,
                        assumptions,
                        container_scenario,
                        index,
                        container_conditions,
                        member_scenarios,
                    );
                }
            }
            _ => member_scenarios.push(ListMemberScenario {
                container_scenario,
                container_conditions: container_conditions.clone(),
                index,
                value: value.clone(),
                conditions: assumptions.clone(),
            }),
        }
    }

    /// Whether the resource authors a value at `path`, whatever that value is.
    ///
    /// Presence is a different question from resolvability: a `Ref` to a
    /// parameter without a default, a `Fn::GetAtt`, or a `Fn::ImportValue` has no
    /// literal before deployment yet the property is set. A rule that asks "is
    /// this property provided?" must use this rather than a concrete lookup, or it
    /// reports a missing property on every template that supplies it indirectly.
    /// A `Properties` block wrapped in `Fn::If` keeps its values under the branch
    /// paths, so a property present in either branch counts as authored.
    #[must_use]
    pub fn has_property(&self, resource_id: &str, path: &str) -> bool {
        let Some(resource) = self.resources.get(resource_id) else {
            return false;
        };
        let property_path = path.strip_prefix(&format!("{KEY_PROPERTIES}.")).unwrap_or(path);
        resource.properties.contains_key(property_path)
            || self.resolve_deep(resource_id, path).is_some()
            || self.resolve_via_properties_if(resource_id, path).is_some()
    }

    /// Whether the value at `path` is unknown before deployment in at least one
    /// scenario, or has no scenario at all. A rule that reasons about every value
    /// a property can take must stop when one of them cannot be known.
    #[must_use]
    pub fn has_unresolved_scenario(&self, resource_id: &str, path: &str) -> bool {
        let scenarios = self.resolve_scenarios_shared(resource_id, path);
        scenarios.is_empty() || scenarios.iter().any(|(value, _)| contains_dynamic_resolved(value))
    }

    /// A key that two values share only when they provably name the same thing.
    ///
    /// A `Ref` or `Fn::GetAtt` to a resource in this template is keyed by that
    /// resource, so `!Ref Vpc` and `!GetAtt Vpc.VpcId` are one key. Every other
    /// value falls back to [`Self::value_identity`]. Callers that group by this key
    /// must treat `None` as "unknown" and neither group nor separate such values.
    #[must_use]
    pub fn referenced_resource_or_value_identity(&self, resource_id: &str, path: &str) -> Option<String> {
        if let Some(target) = self.follow_ref(resource_id, path) {
            return Some(format!("resource:{target}"));
        }
        self.value_identity(resource_id, path)
    }

    #[must_use]
    pub fn is_from_parameter(&self, resource_id: &str, path: &str) -> bool {
        self.parameter_name_at(resource_id, path).is_some()
    }

    pub(crate) fn authored_sub_template_at(&self, resource_id: &str, path: &str) -> Option<&str> {
        let resource_id = resource_id.to_string();
        let mut current = path;
        loop {
            if let Some(node) = self.value_nodes.get(&(resource_id.clone(), current.to_string()))
                && let Node::Intrinsic(IntrinsicFn::Sub(template, _)) = self.arena.node(*node)
            {
                return Some(template.as_str());
            }
            let (parent, _) = current.rsplit_once('.')?;
            current = parent;
        }
    }

    /// The parameter whose declaration stood in for the value at `path`, or `None`
    /// when the value came from somewhere else. A value that is only known at
    /// deployment is not necessarily a parameter - a cross-stack import and a
    /// dynamic reference are equally unknown - so callers that describe a value to
    /// a reader must ask rather than assume.
    #[must_use]
    pub fn parameter_name_at(&self, resource_id: &str, path: &str) -> Option<&str> {
        let source = self.resolution_sources.get(&(resource_id.to_string(), path.to_string()))?;
        parameter_name_from_source(source)
    }

    /// A key that two values share only when they are provably the same value, or
    /// `None` when nothing about the value settles the question.
    ///
    /// A value that resolves before deployment is keyed by its contents, so the
    /// authored form does not matter. A value that stays opaque is keyed by the
    /// expression that produces it, because that is the only thing that can show
    /// two such values are one value. Callers that compare keys must treat `None`
    /// and two differing keys alike: neither proves the values differ, so neither
    /// permits a claim that they are the same.
    #[must_use]
    pub fn value_identity(&self, resource_id: &str, path: &str) -> Option<String> {
        let resolved = self.resolve_deep(resource_id, path).or_else(|| self.resolve(resource_id, path).cloned())?;
        self.resolved_value_identity(resource_id, path, &resolved)
    }

    /// Builds an identity for an already-resolved value at its authored path.
    /// A concrete result that still depends on a reference uses the complete
    /// expression, because a parameter default may be overridden at deployment.
    pub(crate) fn resolved_value_identity(
        &self,
        resource_id: &str,
        path: &str,
        resolved: &ResolvedValue,
    ) -> Option<String> {
        let reference_prefix = format!("{path}.");
        let depends_on_reference = self
            .graph
            .outgoing_edges(resource_id)
            .any(|edge| edge.source_path == path || edge.source_path.starts_with(&reference_prefix));
        let as_json = crate::serialization::resolved_value_to_json(resolved);
        if !depends_on_reference && !json_contains_markers(&as_json) {
            let fingerprint = crate::value_identity::concrete_value_fingerprint(&as_json);
            return Some(format!("value:{fingerprint}"));
        }
        let node = *self.value_nodes.get(&(resource_id.to_string(), path.to_string()))?;
        let (result, depth_exhausted) = crate::value_identity::expression_fingerprint_signaled(&self.arena, node);
        if depth_exhausted {
            self.budget_tracker.record(BudgetKind::ExpressionFingerprintDepth);
        }
        result.map(|fingerprint| format!("expr:{fingerprint}"))
    }

    /// True when the value at `path` (or any ancestor up to the resource root) was
    /// produced by an intrinsic function. Used by rules that skip hardcoded-literal
    /// checks when the property value comes from `Fn::GetAZs`, `Ref`, etc.
    #[must_use]
    pub fn is_from_intrinsic(&self, resource_id: &str, path: &str) -> bool {
        if self.path_from_intrinsic(resource_id, path) {
            return true;
        }
        // Properties wrapped in Fn::If store per-branch intrinsic sources
        // under branch-qualified paths. When a caller queries the bare
        // `Properties.<prop>` path, also consult the branch-qualified
        // variants so intrinsic-sourced values inside either branch are
        // still recognised.
        let properties_prefix = format!("{}.", KEY_PROPERTIES);
        if let Some(rest) = path.strip_prefix(&properties_prefix) {
            for branch in ["1", "2"] {
                let branch_path = format!("{}.{}.{}.{}", KEY_PROPERTIES, FN_IF, branch, rest);
                if self.path_from_intrinsic(resource_id, &branch_path) {
                    return true;
                }
            }
        }
        false
    }

    #[must_use]
    pub fn substituted_paths_under(&self, resource_id: &str, base_path: &str) -> HashSet<String> {
        let prefix = format!("{}.", base_path);
        let relative_path = |path: &str| {
            if path == base_path { Some(String::new()) } else { path.strip_prefix(&prefix).map(String::from) }
        };
        let mut paths: HashSet<String> =
            self.graph.outgoing_edges(resource_id).filter_map(|edge| relative_path(&edge.source_path)).collect();
        paths.extend(self.resolution_sources.iter().filter_map(|((source_resource, source_path), _)| {
            (source_resource == resource_id).then(|| relative_path(source_path)).flatten()
        }));
        paths
    }

    fn path_from_intrinsic(&self, resource_id: &str, path: &str) -> bool {
        let Some(intrinsic_paths) = self.intrinsic_source_paths.get(resource_id) else {
            return false;
        };
        // Walk the property path and each of its ancestors: an intrinsic value at
        // any level makes this path intrinsic-sourced. The index already holds
        // every such path (from an `Intrinsic/*` resolution source or a reference
        // edge), so ancestors are tested as borrowed slices with no per-level
        // key or tuple allocation and no edge scan.
        let mut ancestor = path;
        loop {
            if intrinsic_paths.contains(ancestor) {
                return true;
            }
            match ancestor.rfind('.') {
                Some(dot) => ancestor = &ancestor[..dot],
                None => return false,
            }
        }
    }

    /// Whether this model has spent its cumulative scenario-expansion budget.
    /// Mirrors `ConditionModel::satisfiability_budget_exhausted` for the
    /// scenario path: callers that drive many `resolve_scenarios` queries stop
    /// producing new scenarios once further expansion would only be truncated.
    #[must_use]
    pub fn scenario_budget_exhausted(&self) -> bool {
        self.scenario_combinations_used.load(Ordering::Relaxed) >= MAX_TOTAL_SCENARIO_COMBINATIONS
    }

    /// Record that a budget kind was exhausted. Delegates to the shared tracker.
    pub fn record_budget_exhaustion(&self, kind: BudgetKind) {
        self.budget_tracker.record(kind);
    }

    /// Returns the set of exhausted budget kinds in deterministic order.
    pub fn exhausted_budget_kinds(&self) -> BTreeSet<BudgetKind> {
        self.budget_tracker.exhausted_kinds()
    }

    /// Whether any exhausted budget makes the analysis incomplete.
    pub fn budget_analysis_incomplete(&self) -> bool {
        self.budget_tracker.analysis_incomplete()
    }

    /// Cumulative scenarios materialized by this model so far.
    #[must_use]
    pub fn scenario_combinations_used(&self) -> u64 {
        self.scenario_combinations_used.load(Ordering::Relaxed)
    }

    /// Whether any scenario expansion for this model omitted possible results
    /// because a per-value or model-wide analysis limit was reached. The flag is
    /// monotonic so callers can distinguish complete expansion from conservative
    /// fallback behavior.
    #[must_use]
    pub fn scenario_expansion_curtailed(&self) -> bool {
        self.scenario_expansion_curtailed.load(Ordering::Relaxed)
    }

    fn collect_scenarios_with_budget_status(
        &self,
        value: &ResolvedValue,
        per_value_limit: usize,
        total_limit: u64,
    ) -> (Vec<(ResolvedValue, HashMap<String, bool>)>, bool) {
        self.collect_scenarios_with_budget_from(value, &HashMap::new(), per_value_limit, total_limit)
    }

    fn collect_scenarios_with_budget_from(
        &self,
        value: &ResolvedValue,
        assumptions: &HashMap<String, bool>,
        per_value_limit: usize,
        total_limit: u64,
    ) -> (Vec<(ResolvedValue, HashMap<String, bool>)>, bool) {
        let _expansion_guard = self.scenario_expansion_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        let used = self.scenario_combinations_used.load(Ordering::Relaxed);
        if used >= total_limit {
            self.scenario_expansion_curtailed.store(true, Ordering::Relaxed);
            if total_limit == MAX_TOTAL_SCENARIO_COMBINATIONS {
                self.budget_tracker.record(BudgetKind::ScenarioCombinationsTotal);
            }
            return (Vec::new(), true);
        }

        let remaining = total_limit - used;
        let remaining_limit = usize::try_from(remaining).unwrap_or(usize::MAX);
        let effective_limit = per_value_limit.min(remaining_limit);
        let mut scenarios = Vec::new();
        let is_reachable = |scenario_assumptions: &HashMap<String, bool>| {
            let assumptions: Vec<(String, bool)> =
                scenario_assumptions.iter().map(|(condition, value)| (condition.clone(), *value)).collect();
            self.conditions.is_satisfiable(&assumptions)
        };
        let mut used_covering = false;
        let was_curtailed = collect_scenarios_reachable_with_covering(
            value,
            assumptions,
            effective_limit,
            &mut scenarios,
            &is_reachable,
            &mut used_covering,
        );
        if used_covering {
            self.budget_tracker.record(BudgetKind::ScenarioProductCovering);
        }
        if was_curtailed {
            self.scenario_expansion_curtailed.store(true, Ordering::Relaxed);
            if per_value_limit == MAX_SCENARIO_COMBINATIONS && per_value_limit <= remaining_limit {
                self.budget_tracker.record(BudgetKind::ScenarioCombinationsPerValue);
            }
            if total_limit == MAX_TOTAL_SCENARIO_COMBINATIONS && remaining_limit <= per_value_limit {
                self.budget_tracker.record(BudgetKind::ScenarioCombinationsTotal);
            }
        }
        self.scenario_combinations_used.store(used + scenarios.len() as u64, Ordering::Relaxed);
        (scenarios, was_curtailed)
    }

    fn collect_scenarios_with_budget(
        &self,
        value: &ResolvedValue,
        per_value_limit: usize,
        total_limit: u64,
    ) -> Vec<(ResolvedValue, HashMap<String, bool>)> {
        self.collect_scenarios_with_budget_status(value, per_value_limit, total_limit).0
    }

    /// Test-only: advance the cumulative scenario counter directly, so the
    /// budget threshold and short-circuit behavior can be exercised without
    /// materializing `MAX_TOTAL_SCENARIO_COMBINATIONS` real scenarios.
    #[cfg(test)]
    pub(crate) fn add_scenario_combinations_for_test(&self, count: u64) {
        let _expansion_guard = self.scenario_expansion_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        self.scenario_combinations_used.fetch_add(count, Ordering::Relaxed);
    }

    /// Test-only: set the curtailment flag directly without needing to
    /// materialize a pathological number of real scenarios.
    #[cfg(test)]
    fn set_scenario_expansion_curtailed_for_test(&self) {
        self.scenario_expansion_curtailed.store(true, Ordering::Relaxed);
    }

    fn resolved_properties_value(&self, resource_id: &str) -> Option<ResolvedValue> {
        let resource = self.resources.get(resource_id)?;
        if resource.properties_dynamic {
            return None;
        }
        if resource.properties.len() == 1
            && let Some(conditional) = resource.properties.get(FN_IF)
        {
            return Some(conditional.clone());
        }
        let mut entries: Vec<MapEntry> = resource
            .properties
            .iter()
            .map(|(key, value)| MapEntry { key: key.clone(), value: value.clone() })
            .collect();
        entries.sort_by(|left, right| left.key.cmp(&right.key));
        Some(ResolvedValue::Map { entries })
    }

    /// Returns one immutable whole-properties scenario allocation shared across
    /// read-only callers, materializing and caching it on first access. The lock
    /// is held across expansion so a resource's scenarios are computed once and
    /// its budget charged once.
    pub fn resolve_properties_scenarios_shared(
        &self,
        resource_id: &str,
    ) -> Arc<Vec<(ResolvedValue, HashMap<String, bool>)>> {
        self.resolve_properties_scenarios_selected(resource_id, None)
    }

    /// Owned copy of [`Self::resolve_properties_scenarios_shared`]; read-only
    /// callers should use the shared accessor to avoid cloning the cached
    /// expansion.
    pub fn resolve_properties_scenarios(&self, resource_id: &str) -> Vec<(ResolvedValue, HashMap<String, bool>)> {
        self.resolve_properties_scenarios_shared(resource_id).as_ref().clone()
    }

    /// Like [`Self::resolve_properties_scenarios_shared`] but expands only the
    /// named top-level properties. Conditions that gate other properties never
    /// enter the expansion, so the scenario count grows with the conditions on
    /// the requested properties alone, and a scenario's assumptions carry only
    /// the conditions its values depend on.
    pub fn resolve_properties_scenarios_projected(
        &self,
        resource_id: &str,
        property_names: &[&str],
    ) -> Arc<Vec<(ResolvedValue, HashMap<String, bool>)>> {
        let selection: BTreeSet<String> = property_names.iter().map(|name| (*name).to_string()).collect();
        self.resolve_properties_scenarios_selected(resource_id, Some(selection))
    }

    /// The condition assignments under which the top-level property
    /// `property_name` is absent at deployment: undeclared, or resolving to
    /// `AWS::NoValue`. A macro-owned branch is unknown rather than absent, and
    /// an explicitly authored dynamic value remains present. Only the conditions
    /// that decide the property's own presence appear, never conditions nested
    /// inside a present value, so the result stays linear in the property's
    /// `Fn::If` depth. Empty when the property is always present or the resource's
    /// properties are otherwise dynamic. Scenarios incompatible with the resource's
    /// own condition or the template condition model are excluded.
    #[must_use]
    pub fn property_absence_scenarios(&self, resource_id: &str, property_name: &str) -> Vec<HashMap<String, bool>> {
        let Some(properties) = self.resolved_properties_value(resource_id) else {
            return Vec::new();
        };
        let mut absences = Vec::new();
        collect_property_absences(&properties, property_name, &mut Vec::new(), &mut absences);
        absences.retain(|conditions| {
            let mut assumptions: Vec<(String, bool)> =
                conditions.iter().map(|(condition, value)| (condition.clone(), *value)).collect();
            if let Some(resource_condition) =
                self.resources.get(resource_id).and_then(|resource| resource.condition.as_ref())
            {
                match conditions.get(resource_condition) {
                    Some(false) => return false,
                    Some(true) => {}
                    None => assumptions.push((resource_condition.clone(), true)),
                }
            }
            assumptions.is_empty() || self.conditions.is_satisfiable(&assumptions)
        });
        absences
    }

    fn resolve_properties_scenarios_selected(
        &self,
        resource_id: &str,
        selection: Option<BTreeSet<String>>,
    ) -> Arc<Vec<(ResolvedValue, HashMap<String, bool>)>> {
        let cache_key = (resource_id.to_string(), selection);
        let mut cache = self.properties_scenario_cache.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        if let Some(scenarios) = cache.get(&cache_key) {
            return Arc::clone(scenarios);
        }
        let properties = self.resolved_properties_value(resource_id).map(|properties| match &cache_key.1 {
            Some(selection) => project_properties_value(&properties, selection),
            None => properties,
        });
        let scenarios = match properties {
            Some(properties) => self.collect_scenarios_with_budget(
                &properties,
                MAX_SCENARIO_COMBINATIONS,
                MAX_TOTAL_SCENARIO_COMBINATIONS,
            ),
            None => Vec::new(),
        };
        let shared = Arc::new(scenarios);
        cache.insert(cache_key, Arc::clone(&shared));
        shared
    }

    /// Returns the CloudFormation name of the intrinsic authored exactly at `path`.
    #[doc(hidden)]
    #[must_use]
    pub fn authored_intrinsic_name_at(&self, resource_id: &str, path: &str) -> Option<&'static str> {
        let node_ref = self.value_nodes.get(&(resource_id.to_string(), path.to_string()))?;
        match self.arena.node(*node_ref) {
            Node::Intrinsic(intrinsic) => Some(cfn_function_name(intrinsic)),
            _ => None,
        }
    }

    /// Returns the selected Ref target when a literal-index `Fn::Select` at `path`
    /// chooses a Ref directly or through nested literal-index selects.
    #[doc(hidden)]
    #[must_use]
    pub fn authored_select_ref_target_at(&self, resource_id: &str, path: &str) -> Option<&str> {
        let node_ref = self.value_nodes.get(&(resource_id.to_string(), path.to_string()))?;
        selected_ref_target(&self.arena, *node_ref)
    }

    /// Whether the authored node at `path` is exactly a Ref to `target`.
    #[must_use]
    pub fn is_authored_ref_at(&self, resource_id: &str, path: &str, target: &str) -> bool {
        self.value_nodes
            .get(&(resource_id.to_string(), path.to_string()))
            .is_some_and(|node_ref| {
                matches!(self.arena.node(*node_ref), Node::Intrinsic(IntrinsicFn::Ref(actual)) if actual == target)
            })
    }

    /// Whether the caller supplied a concrete override for this parameter.
    #[must_use]
    pub fn parameter_has_override(&self, parameter_name: &str) -> bool {
        self.parameter_overrides.contains(parameter_name)
    }

    /// Returns the authored, branch-qualified source path for an effective path in
    /// one resolved scenario. The returned path uses the same dotted form as
    /// resolver source paths and can be passed to [`Self::resource_span`].
    pub fn scenario_source_path(
        &self,
        resource_id: &str,
        effective_path: &str,
        conditions: &HashMap<String, bool>,
    ) -> Option<String> {
        let properties = self.resolved_properties_value(resource_id)?;
        let relative_path = effective_path
            .strip_prefix("Properties.")
            .or_else(|| (effective_path == KEY_PROPERTIES).then_some(""))
            .unwrap_or(effective_path);
        scenario_source_path_at(&properties, relative_path, conditions, KEY_PROPERTIES)
    }

    /// Whether the selected value is authored from an AWS pseudo parameter.
    #[must_use]
    pub fn scenario_uses_pseudo_parameter(
        &self,
        resource_id: &str,
        effective_path: &str,
        conditions: &HashMap<String, bool>,
    ) -> bool {
        let source_path = self
            .scenario_source_path(resource_id, effective_path, conditions)
            .unwrap_or_else(|| effective_path.to_string());
        let descendant_prefix = format!("{source_path}.");
        let Some(source_paths) = self.pseudo_parameter_source_paths.get(resource_id) else {
            return false;
        };
        source_paths
            .iter()
            .any(|candidate_path| candidate_path == &source_path || candidate_path.starts_with(&descendant_prefix))
    }

    /// Whether a scenario depends on a parameter whose final deployment value is
    /// unknown. Ordinary caller overrides are fixed values; overrides for SSM
    /// value parameter types remain lookup names that CloudFormation resolves.
    #[must_use]
    pub fn scenario_uses_unresolved_parameter(
        &self,
        resource_id: &str,
        effective_path: &str,
        conditions: &HashMap<String, bool>,
    ) -> bool {
        self.graph.outgoing_edges(resource_id).any(|edge| {
            self.unresolved_parameter_edge_applies(edge, conditions)
                && Self::reference_path_overlaps(&edge.source_path, effective_path)
        })
    }

    /// Whether the value at `effective_path` is produced by an unresolved
    /// parameter reference at that path or one of its ancestors. References in
    /// descendant siblings do not make the value itself unknown.
    pub(crate) fn scenario_value_uses_unresolved_parameter(
        &self,
        resource_id: &str,
        effective_path: &str,
        conditions: &HashMap<String, bool>,
    ) -> bool {
        self.graph.outgoing_edges(resource_id).any(|edge| {
            self.unresolved_parameter_edge_applies(edge, conditions)
                && Self::reference_path_contains_value(&edge.source_path, effective_path)
        })
    }

    fn unresolved_parameter_edge_applies(&self, edge: &Edge, conditions: &HashMap<String, bool>) -> bool {
        self.parameters.get(&edge.target).is_some_and(|parameter| {
            (!self.parameter_overrides.contains(&edge.target)
                || parameter.param_type.starts_with(SSM_PARAMETER_VALUE_TYPE_PREFIX))
                && Self::edge_applies_to_scenario(edge.condition_context.as_deref(), conditions)
        })
    }

    fn reference_path_overlaps(source_path: &str, effective_path: &str) -> bool {
        let Some(reference_anchor) = Self::reference_anchor(source_path) else {
            return false;
        };
        Self::path_is_at_or_below(&reference_anchor, effective_path)
            || Self::path_is_at_or_below(effective_path, &reference_anchor)
    }

    fn reference_path_contains_value(source_path: &str, effective_path: &str) -> bool {
        Self::reference_anchor(source_path)
            .is_some_and(|reference_anchor| Self::path_is_at_or_below(effective_path, &reference_anchor))
    }

    fn reference_anchor(source_path: &str) -> Option<String> {
        let normalized_source = effective_path_from_scenario_source_path(source_path)?;
        Some(normalized_source.split_once(".Fn::").map_or(normalized_source.as_str(), |(prefix, _)| prefix).to_string())
    }

    fn path_is_at_or_below(candidate_path: &str, root_path: &str) -> bool {
        if root_path.is_empty() {
            return true;
        }
        let mut candidate_segments = candidate_path.split('.');
        root_path.split('.').all(|root_segment| {
            !root_segment.is_empty()
                && candidate_segments.next().is_some_and(|candidate_segment| {
                    !candidate_segment.is_empty() && (root_segment == "{}" || root_segment == candidate_segment)
                })
        })
    }

    fn edge_applies_to_scenario(condition_context: Option<&str>, conditions: &HashMap<String, bool>) -> bool {
        condition_context.is_none_or(|context| {
            context.split(',').all(|selector| {
                let (condition, expected) =
                    selector.strip_prefix('!').map_or((selector, true), |condition| (condition, false));
                !condition.is_empty() && conditions.get(condition).is_none_or(|actual| *actual == expected)
            })
        })
    }

    /// Returns one immutable allocation of the scenarios for the value at `path`
    /// under the default per-value limit, shared across read-only callers.
    pub fn resolve_scenarios_shared(
        &self,
        resource_id: &str,
        path: &str,
    ) -> Arc<Vec<(ResolvedValue, HashMap<String, bool>)>> {
        self.resolve_scenarios_with_limit_shared(resource_id, path, MAX_SCENARIO_COMBINATIONS).0
    }

    /// Owned copy of [`Self::resolve_scenarios_shared`]; read-only callers should
    /// use the shared accessor to avoid cloning the memoized expansion.
    pub fn resolve_scenarios(&self, resource_id: &str, path: &str) -> Vec<(ResolvedValue, HashMap<String, bool>)> {
        self.resolve_scenarios_shared(resource_id, path).as_ref().clone()
    }

    /// Expands the value at `path` under a per-value scenario limit, reporting
    /// whether the expansion was curtailed. Results are memoized per
    /// `(resource, path, limit)` so repeated queries, including the default-limit
    /// [`Self::resolve_scenarios_shared`], share one allocation and charge the
    /// cumulative budget once.
    pub fn resolve_scenarios_with_limit_shared(
        &self,
        resource_id: &str,
        path: &str,
        per_value_limit: usize,
    ) -> (Arc<Vec<(ResolvedValue, HashMap<String, bool>)>>, bool) {
        let key = (resource_id.to_string(), path.to_string(), per_value_limit);
        let mut memo = self.raw_scenario_memo.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        if let Some((scenarios, was_curtailed)) = memo.get(&key) {
            return (Arc::clone(scenarios), *was_curtailed);
        }
        let (scenarios, was_curtailed) = self.expand_scenarios_at_path(resource_id, path, per_value_limit);
        let scenarios = Arc::new(scenarios);
        memo.insert(key, (Arc::clone(&scenarios), was_curtailed));
        (scenarios, was_curtailed)
    }

    /// Owned copy of [`Self::resolve_scenarios_with_limit_shared`]; read-only
    /// callers should use the shared accessor to avoid cloning the memoized
    /// expansion.
    pub fn resolve_scenarios_with_limit(
        &self,
        resource_id: &str,
        path: &str,
        per_value_limit: usize,
    ) -> (Vec<(ResolvedValue, HashMap<String, bool>)>, bool) {
        let (scenarios, was_curtailed) = self.resolve_scenarios_with_limit_shared(resource_id, path, per_value_limit);
        (scenarios.as_ref().clone(), was_curtailed)
    }

    fn expand_scenarios_at_path(
        &self,
        resource_id: &str,
        path: &str,
        per_value_limit: usize,
    ) -> (Vec<(ResolvedValue, HashMap<String, bool>)>, bool) {
        let Some(value) = self.resolve_unexpanded(resource_id, path) else {
            return (Vec::new(), false);
        };
        self.collect_scenarios_with_budget_status(&value, per_value_limit, MAX_TOTAL_SCENARIO_COMBINATIONS)
    }

    /// Fallback lookup when `Properties` is wrapped in an `Fn::If`: walks
    /// the caller's path inside the synthetic `Fn::If` branch so scenario
    /// resolution still reaches properties that only exist conditionally.
    fn resolve_via_properties_if(&self, resource_id: &str, path: &str) -> Option<ResolvedValue> {
        let resource = self.resources.get(resource_id)?;
        let conditional = resource.properties.get(FN_IF)?;
        let properties_prefix = format!("{}.", KEY_PROPERTIES);
        let prop_path = path.strip_prefix(&properties_prefix).unwrap_or(path);
        if prop_path.is_empty() {
            return Some(conditional.clone());
        }
        resolved_value_at_path(conditional, prop_path)
    }

    fn resolve_scenarios_json_uncached(
        &self,
        resource_id: &str,
        path: &str,
    ) -> Vec<(serde_json::Value, HashMap<String, bool>)> {
        self.resolve_scenarios_shared(resource_id, path)
            .iter()
            .filter_map(|(val, conds)| {
                if contains_dynamic_resolved(val) {
                    return None;
                }
                if !conds.is_empty() {
                    let assumptions: Vec<(String, bool)> = conds.iter().map(|(k, v)| (k.clone(), *v)).collect();
                    if !self.conditions.is_satisfiable(&assumptions) {
                        return None;
                    }
                }
                let json = crate::serialization::resolved_value_to_json_clean(val)?;
                if json_contains_markers(&json) {
                    return None;
                }
                Some((json, conds.clone()))
            })
            .collect()
    }

    fn cache_json_scenarios(
        &self,
        memo_key: (String, String),
        scenarios: Vec<(serde_json::Value, HashMap<String, bool>)>,
    ) -> Arc<Vec<(serde_json::Value, HashMap<String, bool>)>> {
        let mut memo = self.scenario_memo.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        Arc::clone(memo.entry(memo_key).or_insert_with(|| Arc::new(scenarios)))
    }

    /// Returns one immutable scenario allocation shared across read-only callers.
    pub fn resolve_scenarios_json_shared(
        &self,
        resource_id: &str,
        path: &str,
    ) -> Arc<Vec<(serde_json::Value, HashMap<String, bool>)>> {
        let memo_key = (resource_id.to_string(), path.to_string());
        {
            let memo = self.scenario_memo.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
            if let Some(memoized) = memo.get(&memo_key) {
                return Arc::clone(memoized);
            }
        }
        let scenarios = self.resolve_scenarios_json_uncached(resource_id, path);
        self.cache_json_scenarios(memo_key, scenarios)
    }

    pub fn resolve_scenarios_json(
        &self,
        resource_id: &str,
        path: &str,
    ) -> Vec<(serde_json::Value, HashMap<String, bool>)> {
        let memo_key = (resource_id.to_string(), path.to_string());
        {
            let memo = self.scenario_memo.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
            if let Some(memoized) = memo.get(&memo_key) {
                return memoized.as_ref().clone();
            }
        }
        let scenarios = self.resolve_scenarios_json_uncached(resource_id, path);
        self.cache_json_scenarios(memo_key, scenarios).as_ref().clone()
    }

    pub fn follow_ref(&self, resource_id: &str, path: &str) -> Option<&str> {
        if let Some(ResolvedValue::Reference { target, kind: _ }) = self.resolve(resource_id, path) {
            return Some(target.as_str());
        }
        if let Some(ResolvedValue::Reference { target, kind: _ }) = self.resolve_deep(resource_id, path).as_ref() {
            for edge in self.graph.outgoing_edges(resource_id) {
                if edge.target == *target {
                    return Some(&edge.target);
                }
            }
        }
        None
    }

    pub fn source_location(&self, path: &str) -> Option<&SourceSpan> {
        self.span_index.get(path)
    }

    /// Returns the source span for a resource property path, falling back to
    /// the resource-level span if the specific path is not found.
    ///
    /// The span index is keyed with slash-separated paths, so a dotted,
    /// resource-relative path (`Properties.BucketName`) is normalized to slash
    /// form before lookup - the same normalization [`Self::diagnostic_span`]
    /// applies. Callers pass paths in either form, so accepting only slash form
    /// here would silently mislocate every dotted-path diagnostic onto the
    /// resource declaration line.
    pub fn resource_span(&self, resource_id: &str, prop_path: &str) -> SourceSpan {
        // An empty resource id means the path is already a section-absolute span-index
        // key (e.g. an output's `Outputs/X/Value.Fn::Join`); prefixing it with
        // `Resources/` would mislocate the finding onto the Resources block. Resolve it
        // as-is (no dot-to-slash conversion, since dots in segment names like `Fn::Join`
        // are literal, not path separators). Returns UNKNOWN when nothing resolves so
        // callers fall back to section-level or backfill-based location.
        if resource_id.is_empty() {
            return self.walk_up_span(prop_path).unwrap_or(UNKNOWN_SPAN);
        }
        // Resolver paths preserve literal dots and slashes inside map keys, so converting
        // them into the slash-keyed index can be ambiguous. A terminal intrinsic can be
        // resolved through its exact authored parent path without lossy conversion.
        if let Some((authored_path, function_name)) = prop_path.rsplit_once('.')
            && let Some(node_ref) = self.value_nodes.get(&(resource_id.to_string(), authored_path.to_string()))
            && let Node::Intrinsic(intrinsic) = self.arena.node(*node_ref)
            && cfn_function_name(intrinsic) == function_name
        {
            let intrinsic_path = format!("{}/{}", self.arena.get(*node_ref).path, function_name);
            if let Some(span) = self.walk_up_span(&intrinsic_path) {
                return span;
            }
        }
        let specific = if prop_path.is_empty() {
            format!("Resources/{}", resource_id)
        } else {
            format!("Resources/{}/{}", resource_id, prop_path.replace('.', "/"))
        };
        let exact_span = self.span_index.get(&specific).copied().filter(|span| *span != UNKNOWN_SPAN);
        if matches!(specific.rsplit_once('/'), Some((parent, "1" | "2")) if parent.ends_with("/Fn::If"))
            && let Some(descendant) = self.conditional_branch_first_descendant_spans.get(&specific).copied()
            && exact_span.is_none_or(|exact| {
                (descendant.start_line, descendant.start_column) < (exact.start_line, exact.start_column)
            })
        {
            return descendant;
        }
        if let Some(exact) = exact_span {
            return exact;
        }
        // Walk up from the exact path to the nearest indexed ancestor, so a leaf that
        // carries no span of its own - a synthetic intrinsic key (`…/Topic/Fn::Sub`),
        // an `Fn::If` branch index - anchors at its closest real parent rather than
        // collapsing straight to the resource declaration. The resource path itself is
        // the final ancestor, preserving the previous resource-level fallback.
        self.walk_up_span(&specific).unwrap_or(UNKNOWN_SPAN)
    }

    /// Walks up `key` (a `/`-separated span-index path), trimming one trailing
    /// segment at a time, and returns the first ancestor with a known span. This
    /// anchors a diagnostic as close to the offending node as the index allows -
    /// an unindexed leaf (e.g. an `Fn::If` branch index) falls back to its parent
    /// property, then the resource, then the section.
    fn walk_up_span(&self, key: &str) -> Option<SourceSpan> {
        let mut current = key;
        loop {
            // A key can be present in the index but mapped to UNKNOWN_SPAN (an
            // interior node whose byte span was never assigned). Treat that as
            // "not found" and keep walking up to the nearest ancestor that does
            // have a real span, rather than returning the unusable UNKNOWN.
            match self.span_index.get(current) {
                Some(&span) if span != UNKNOWN_SPAN => return Some(span),
                _ => match current.rfind('/') {
                    Some(cut) => current = &current[..cut],
                    None => return None,
                },
            }
        }
    }

    /// Best-effort span for a diagnostic identified by its optional resource and
    /// property path, walking up to the nearest indexed ancestor.
    ///
    /// The path form disambiguates how to root it, which matters because a segment
    /// like `Metadata` names both a top-level section and a resource property:
    /// * A **slash** form (`Outputs/X/Value`, `Conditions/C/Fn::And`) is already an
    ///   absolute, section-rooted span-index key and is resolved as written.
    /// * A **bare path with no resource** (`BogusSection`) is a top-level key and is
    ///   also resolved directly.
    /// * A **dotted** or bare form with a resource (`Properties.Foo`, `Metadata`) is
    ///   relative to that resource, so it is rooted at `Resources/<rid>` before
    ///   lookup - never matched against a same-named top-level section.
    ///
    /// Returns `None` when nothing along the chosen candidate is indexed, so callers
    /// can fall back to a section span.
    pub fn diagnostic_span(&self, resource_id: Option<&str>, property_path: &str) -> Option<SourceSpan> {
        let rid = resource_id.filter(|r| !r.is_empty());

        if property_path.contains('/') || (rid.is_none() && !property_path.is_empty()) {
            // Absolute, section-rooted path: resolve directly.
            if let Some(span) = self.walk_up_span(property_path) {
                return Some(span);
            }
        } else if let Some(rid) = rid {
            // Resource-relative path (dotted or bare): root at the resource so a
            // property named after a top-level section cannot mislocate onto it.
            let key = if property_path.is_empty() {
                format!("Resources/{}", rid)
            } else {
                format!("Resources/{}/{}", rid, property_path.replace('.', "/"))
            };
            if let Some(span) = self.walk_up_span(&key) {
                return Some(span);
            }
        }

        // Last resort: the bare resource span, when a resource id is known.
        rid.and_then(|rid| self.span_index.get(&format!("Resources/{}", rid)))
            .filter(|span| **span != UNKNOWN_SPAN)
            .copied()
    }

    /// Bounds on the length of a string value the template does not state
    /// literally - one built by an intrinsic, or chosen from a parameter's
    /// AllowedValues. `None` when the length cannot be pinned for every
    /// possibility, and also when the template does state the value literally,
    /// because schema validation checks a literal against the constraint directly.
    ///
    /// Both engines decide what to report from this one answer, so neither can
    /// reach a different conclusion about the same property.
    pub fn estimated_string_length_bounds(&self, resource_id: &str, path: &str) -> Option<(usize, usize)> {
        let val = self.resolve_deep(resource_id, path).or_else(|| self.resolve(resource_id, path).cloned())?;
        if matches!(&val, ResolvedValue::Concrete { value } if value.is_string()) {
            return None;
        }
        estimate_resolved_string_length_bounds(&val)
    }
}

impl SpanProvider for SemanticModel {
    fn source_location(&self, path: &str) -> Option<SourceSpan> {
        self.span_index.get(path).copied()
    }
}

fn collect_property_absences(
    properties: &ResolvedValue,
    property_name: &str,
    literals: &mut Vec<(String, bool)>,
    absences: &mut Vec<HashMap<String, bool>>,
) {
    let assignment = |literals: &[(String, bool)]| literals.iter().cloned().collect::<HashMap<String, bool>>();
    match properties {
        ResolvedValue::Conditional { condition, if_true, if_false } => {
            for (branch, holds) in [(if_true, true), (if_false, false)] {
                match literals.iter().find(|(name, _)| name == condition) {
                    Some((_, decided)) if *decided == holds => {
                        collect_property_absences(branch, property_name, literals, absences);
                    }
                    Some(_) => {}
                    None => {
                        literals.push((condition.clone(), holds));
                        collect_property_absences(branch, property_name, literals, absences);
                        literals.pop();
                    }
                }
            }
        }
        ResolvedValue::Map { entries } if entries.iter().any(|entry| entry.key == FN_TRANSFORM) => {}
        ResolvedValue::Map { entries } => match entries.iter().find(|entry| entry.key == property_name) {
            None => absences.push(assignment(literals)),
            Some(entry) => {
                for absence_literals in absent_alternatives(&entry.value, literals) {
                    absences.push(assignment(&absence_literals));
                }
            }
        },
        ResolvedValue::Concrete { value } if value.get(property_name).is_none_or(serde_json::Value::is_null) => {
            absences.push(assignment(literals));
        }
        _ => {}
    }
}

/// Keeps only the selected top-level properties of a resolved `Properties`
/// value, descending through an `Fn::If` that wraps the whole block.
fn project_properties_value(properties: &ResolvedValue, selection: &BTreeSet<String>) -> ResolvedValue {
    match properties {
        ResolvedValue::Conditional { condition, if_true, if_false } => ResolvedValue::Conditional {
            condition: condition.clone(),
            if_true: Box::new(project_properties_value(if_true, selection)),
            if_false: Box::new(project_properties_value(if_false, selection)),
        },
        ResolvedValue::Map { entries } => ResolvedValue::Map {
            entries: entries.iter().filter(|entry| selection.contains(&entry.key)).cloned().collect(),
        },
        ResolvedValue::Concrete { value } => match value.as_object() {
            Some(object) => ResolvedValue::Concrete {
                value: serde_json::Value::Object(
                    object
                        .iter()
                        .filter(|(key, _)| selection.contains(*key))
                        .map(|(k, v)| (k.clone(), v.clone()))
                        .collect(),
                )
                .into(),
            },
            None => properties.clone(),
        },
        other => other.clone(),
    }
}

fn parse_rules(rules_json: &Option<serde_json::Value>, arena: &Arena, rules_node: NodeRef) -> Vec<TemplateRule> {
    let Some(rules) = rules_json else {
        return Vec::new();
    };
    let Some(obj) = rules.as_object() else {
        return Vec::new();
    };

    // Build a lookup from rule name → NodeRef for the rule's IR subtree
    let rule_nodes: HashMap<&str, NodeRef> =
        arena.as_map(rules_node).unwrap_or(&[]).iter().map(|(k, v)| (k.as_str(), *v)).collect();

    obj.iter()
        .filter_map(|(name, rule)| {
            let rule_obj = rule.as_object()?;
            let condition = rule_obj.get(KEY_RULE_CONDITION).cloned();

            // Look up the IR node for this rule to find condition/assertion NodeRefs
            let rule_ir = rule_nodes.get(name.as_str()).copied().unwrap_or(NULL_REF);
            let condition_node = arena.map_get(rule_ir, KEY_RULE_CONDITION).unwrap_or(NULL_REF);

            let assertions_node = arena.map_get(rule_ir, KEY_ASSERTIONS).unwrap_or(NULL_REF);
            let assertion_items = arena.as_list(assertions_node).unwrap_or(&[]);

            let assertions = rule_obj
                .get(KEY_ASSERTIONS)
                .and_then(|a| a.as_array())
                .map(|arr| {
                    arr.iter()
                        .enumerate()
                        .filter_map(|(idx, a)| {
                            let a_obj = a.as_object()?;
                            let assert_node = assertion_items
                                .get(idx)
                                .and_then(|item_ref| arena.map_get(*item_ref, KEY_ASSERT))
                                .unwrap_or(NULL_REF);
                            Some(RuleAssertion {
                                assert: a_obj.get(KEY_ASSERT).cloned().unwrap_or(serde_json::Value::Null),
                                assert_node,
                                description: a_obj
                                    .get(KEY_ASSERT_DESCRIPTION)
                                    .and_then(|d| d.as_str())
                                    .map(String::from),
                            })
                        })
                        .collect()
                })
                .unwrap_or_default();
            Some(TemplateRule { name: name.clone(), condition, condition_node, assertions })
        })
        .collect()
}

/// Collect parameter names that are referenced (via `Ref`/`Fn::Sub`) from within
/// another parameter's definition. These references still count as usage for the
/// unused-parameter check even though they originate in the Parameters section.
fn collect_parameter_definition_refs(ir: &TemplateIR, parameters: &HashMap<String, ParameterInfo>) -> HashSet<String> {
    let mut referenced = HashSet::new();
    if ir.parameters == NULL_REF {
        return referenced;
    }
    let Some(param_entries) = ir.arena.as_map(ir.parameters) else {
        return referenced;
    };
    for (_, param_ref) in param_entries {
        collect_refs_in_subtree(&ir.arena, *param_ref, parameters, &mut referenced);
    }
    referenced
}

fn collect_refs_in_subtree(
    arena: &Arena,
    node_ref: NodeRef,
    parameters: &HashMap<String, ParameterInfo>,
    referenced: &mut HashSet<String>,
) {
    match arena.node(node_ref) {
        Node::Intrinsic(IntrinsicFn::Ref(target)) => {
            if parameters.contains_key(target) {
                referenced.insert(target.clone());
            }
        }
        Node::Intrinsic(IntrinsicFn::Sub(_, Some(bindings))) => {
            for (_, v) in bindings {
                collect_refs_in_subtree(arena, *v, parameters, referenced);
            }
        }
        Node::List(items) => {
            for item in items {
                collect_refs_in_subtree(arena, *item, parameters, referenced);
            }
        }
        Node::Map(entries) => {
            for (_, v) in entries {
                collect_refs_in_subtree(arena, *v, parameters, referenced);
            }
        }
        _ => {}
    }
}

fn selected_ref_target(arena: &Arena, node_ref: NodeRef) -> Option<&str> {
    match arena.node(node_ref) {
        Node::Intrinsic(IntrinsicFn::Ref(target)) => Some(target),
        Node::Intrinsic(IntrinsicFn::Select(index_ref, values_ref)) => {
            let index = match arena.node(*index_ref) {
                Node::Int(index) => usize::try_from(*index).ok()?,
                Node::String(index) => index.parse().ok()?,
                _ => return None,
            };
            let selected_ref = arena.as_list(*values_ref)?.get(index)?;
            selected_ref_target(arena, *selected_ref)
        }
        _ => None,
    }
}

fn canonical_conditions(conditions: &HashMap<String, bool>) -> Vec<(String, bool)> {
    let mut canonical: Vec<_> = conditions.iter().map(|(name, value)| (name.clone(), *value)).collect();
    canonical.sort_unstable();
    canonical
}

fn parameter_name_from_source(source: &str) -> Option<&str> {
    let rest = source.strip_prefix(SECTION_PARAMETERS)?.strip_prefix('/')?;
    let name = rest.split('/').next()?;
    if name.is_empty() { None } else { Some(name) }
}

/// Resolution-source marker for a value produced by an intrinsic function.
const INTRINSIC_SOURCE_PREFIX: &str = "Intrinsic/";

/// Builds the per-resource intrinsic-source path index consulted by
/// `is_from_intrinsic`. A path counts as intrinsic-sourced when its resolution
/// source is an `Intrinsic/*` entry or when it anchors a reference edge (Ref,
/// GetAtt, Sub) - the two conditions the ancestor walk previously re-derived by
/// probing the resolution sources and rescanning the outgoing edges on every
/// query. Materializing them once trades a small, bounded amount of memory for
/// allocation-free borrowed lookups on the hot path.
fn build_intrinsic_source_paths(
    resolution_sources: &HashMap<(String, String), String>,
    graph: &ReferenceGraph,
) -> HashMap<String, HashSet<String>> {
    let mut paths_by_resource: HashMap<String, HashSet<String>> = HashMap::new();
    for ((resource_id, path), source) in resolution_sources {
        if source.starts_with(INTRINSIC_SOURCE_PREFIX) {
            paths_by_resource.entry(resource_id.clone()).or_default().insert(path.clone());
        }
    }
    for edge in &graph.edges {
        paths_by_resource.entry(edge.source_resource.clone()).or_default().insert(edge.source_path.clone());
    }
    paths_by_resource
}

fn build_conditional_branch_first_descendant_spans(span_index: &SourceSpanIndex) -> HashMap<String, SourceSpan> {
    let mut first_by_branch = HashMap::new();
    for (path, span) in span_index {
        if *span == UNKNOWN_SPAN {
            continue;
        }
        let segments: Vec<&str> = path.split('/').collect();
        for index in 0..segments.len().saturating_sub(2) {
            if segments[index] != FN_IF || !matches!(segments[index + 1], "1" | "2") {
                continue;
            }
            let branch_path = segments[..=index + 1].join("/");
            first_by_branch
                .entry(branch_path)
                .and_modify(|known: &mut SourceSpan| {
                    if (span.start_line, span.start_column, span.end_line, span.end_column)
                        < (known.start_line, known.start_column, known.end_line, known.end_column)
                    {
                        *known = *span;
                    }
                })
                .or_insert(*span);
        }
    }
    first_by_branch
}

fn build_pseudo_parameter_source_paths(
    arena: &Arena,
    value_nodes: &HashMap<(String, String), NodeRef>,
) -> HashMap<String, HashSet<String>> {
    let mut paths_by_resource: HashMap<String, HashSet<String>> = HashMap::new();
    for ((resource_id, path), node_ref) in value_nodes {
        if node_uses_pseudo_parameter(arena, *node_ref) {
            paths_by_resource.entry(resource_id.clone()).or_default().insert(path.clone());
        }
    }
    paths_by_resource
}

fn node_uses_pseudo_parameter(arena: &Arena, node_ref: NodeRef) -> bool {
    match arena.node(node_ref) {
        Node::Intrinsic(IntrinsicFn::Ref(target)) => {
            target != PSEUDO_NO_VALUE && PSEUDO_PARAMETERS.contains(&target.as_str())
        }
        Node::Intrinsic(IntrinsicFn::Sub(template, bindings)) => {
            let binding_overrides = |name: &str| {
                bindings.as_ref().is_some_and(|entries| entries.iter().any(|(binding, _)| binding == name))
            };
            PSEUDO_PARAMETERS
                .iter()
                .filter(|name| **name != PSEUDO_NO_VALUE)
                .any(|name| !binding_overrides(name) && template.contains(&format!("${{{name}}}")))
                || bindings
                    .as_ref()
                    .is_some_and(|entries| entries.iter().any(|(_, value)| node_uses_pseudo_parameter(arena, *value)))
        }
        _ => false,
    }
}

/// Some intrinsic nodes stand in for a whole object - most notably
/// `Properties: {Fn::If: [...]}` which the parser folds into an
/// `IntrinsicFn::If` node. Return the CloudFormation function-name key
/// (`Fn::If`, `Fn::ForEach::*`, etc.) when the node is one of these
/// object-wrapping intrinsics, so downstream resolution can address the
/// conditional by a synthetic path.
fn intrinsic_synthetic_key(arena: &Arena, node_ref: NodeRef) -> Option<String> {
    let Node::Intrinsic(func) = arena.node(node_ref) else {
        return None;
    };
    match func {
        IntrinsicFn::If(_, _, _) | IntrinsicFn::IfExpr(_, _, _) => Some(FN_IF.to_string()),
        IntrinsicFn::ForEach(uid, _, _, _) => Some(format!("{}::{}", FN_FOR_EACH, uid)),
        _ => None,
    }
}

/// Valid resource-level attributes per the CloudFormation template anatomy.
/// These are the keys CloudFormation accepts directly under a resource's logical
/// ID in the `Resources` section.
const VALID_RESOURCE_ATTRIBUTES: &[&str] = &[
    KEY_TYPE,
    KEY_PROPERTIES,
    KEY_DEPENDS_ON,
    KEY_CONDITION,
    SECTION_METADATA,
    KEY_DELETION_POLICY,
    KEY_UPDATE_REPLACE_POLICY,
    KEY_UPDATE_POLICY,
    KEY_CREATION_POLICY,
    FN_TRANSFORM,
    "Version",
];

/// Additional resource-level attributes valid only when the SAM transform is
/// declared. These are SAM-specific extensions that the Serverless transform
/// processes before CloudFormation sees the resource.
const SAM_RESOURCE_ATTRIBUTES: &[&str] = &["Connectors", "IgnoreGlobals"];

fn invalid_resource_condition_ref(arena: &Arena, node_ref: NodeRef) -> Option<NodeRef> {
    let entries = arena.as_map(node_ref)?;
    let condition_ref = entries.iter().find(|(key, _)| key == KEY_CONDITION).map(|(_, value)| *value)?;
    (!matches!(arena.node(condition_ref), Node::String(_))).then_some(condition_ref)
}

/// Validates the structural shape of a resource body, producing diagnostics for
/// bodies that CloudFormation would reject: a non-object resource, a missing
/// `Type`, or unknown resource-level attribute keys.
fn validate_resource_shape(
    arena: &Arena,
    name: &str,
    node_ref: NodeRef,
    is_sam: bool,
    span_index: &SourceSpanIndex,
) -> Vec<ParseDefect> {
    let mut out = Vec::new();
    let resource_span = span_index.get(&format!("Resources/{}", name)).copied().unwrap_or(UNKNOWN_SPAN);

    // The resource body must be an object.
    let entries = match arena.as_map(node_ref) {
        Some(entries) => entries,
        None => {
            let shape = match arena.node(node_ref) {
                Node::Null => "null",
                Node::Bool(_) => "a boolean",
                Node::Int(_) | Node::Float(_) => "a number",
                Node::String(_) => "a string",
                Node::List(_) => "a list",
                Node::Intrinsic(_) => "an intrinsic function",
                Node::Map(_) => return out,
            };
            out.push(crate::make_parse_defect_for_resource(
                "E3001",
                format!("Resource '{}' body must be an object, got {}", name, shape),
                resource_span,
                name,
            ));
            return out;
        }
    };

    // `Type` is required and must be a string.
    match entries.iter().find(|(k, _)| k == KEY_TYPE) {
        None => {
            out.push(crate::make_parse_defect_for_resource(
                "E3001",
                format!("Resource '{}' is missing required property 'Type'", name),
                resource_span,
                name,
            ));
        }
        Some((_, type_ref)) if !matches!(arena.node(*type_ref), Node::String(_)) => {
            let type_span = span_index.get(&format!("Resources/{}/Type", name)).copied().unwrap_or(resource_span);
            out.push(crate::make_parse_defect_for_resource(
                "E3001",
                format!("Resource '{}' property 'Type' must be a string", name),
                type_span,
                name,
            ));
        }
        _ => {}
    }

    // `Condition` must be a string when present.
    if invalid_resource_condition_ref(arena, node_ref).is_some() {
        let cond_span = span_index.get(&format!("Resources/{}/Condition", name)).copied().unwrap_or(resource_span);
        out.push(crate::make_parse_defect_for_resource(
            "E3001",
            format!("Resource '{}' property 'Condition' must be a string", name),
            cond_span,
            name,
        ));
    }

    // `DependsOn` must be a string or a list of strings when present.
    if let Some((_, dep_ref)) = entries.iter().find(|(k, _)| k == KEY_DEPENDS_ON) {
        match arena.node(*dep_ref) {
            Node::String(_) => {}
            Node::List(items) => {
                for item_ref in items {
                    if !matches!(arena.node(*item_ref), Node::String(_)) {
                        let dep_span =
                            span_index.get(&format!("Resources/{}/DependsOn", name)).copied().unwrap_or(resource_span);
                        out.push(crate::make_parse_defect_for_resource(
                            "E3001",
                            format!("Resource '{}' property 'DependsOn' list elements must be strings", name),
                            dep_span,
                            name,
                        ));
                        break;
                    }
                }
            }
            _ => {
                let dep_span =
                    span_index.get(&format!("Resources/{}/DependsOn", name)).copied().unwrap_or(resource_span);
                out.push(crate::make_parse_defect_for_resource(
                    "E3001",
                    format!("Resource '{}' property 'DependsOn' must be a string or list of strings", name),
                    dep_span,
                    name,
                ));
            }
        }
    }

    let resource_type = entries.iter().find(|(key, _)| key == KEY_TYPE).and_then(|(_, value)| arena.as_str(*value));
    if let Some(resource_type) = resource_type {
        let is_custom_resource = is_custom_resource_type(resource_type);
        if let Some((_, version_ref)) = entries.iter().find(|(key, _)| key == "Version") {
            let version_span = span_index.get(&format!("Resources/{}/Version", name)).copied().unwrap_or(resource_span);
            if !is_custom_resource {
                out.push(crate::make_parse_defect_for_resource(
                    "E3001",
                    format!("Resource '{}' property 'Version' is only valid for custom resources", name),
                    version_span,
                    name,
                ));
            } else if !matches!(arena.node(*version_ref), Node::String(_) | Node::Int(_)) {
                out.push(crate::make_parse_defect_for_resource(
                    "E3001",
                    format!("Resource '{}' property 'Version' must be a string or integer", name),
                    version_span,
                    name,
                ));
            }
        }

        for policy_name in [KEY_CREATION_POLICY, KEY_UPDATE_POLICY] {
            let Some((_, policy_ref)) = entries.iter().find(|(key, _)| key == policy_name) else {
                continue;
            };
            let policy_span =
                span_index.get(&format!("Resources/{}/{}", name, policy_name)).copied().unwrap_or(resource_span);
            if is_custom_resource {
                out.push(crate::make_parse_defect_for_resource(
                    "E3001",
                    format!("Resource '{}' property '{}' is not valid for custom resources", name, policy_name),
                    policy_span,
                    name,
                ));
            } else if !matches!(arena.node(*policy_ref), Node::Map(_) | Node::Intrinsic(_)) {
                out.push(crate::make_parse_defect_for_resource(
                    "E3001",
                    format!("Resource '{}' property '{}' must be an object", name, policy_name),
                    policy_span,
                    name,
                ));
            }
        }
    }

    for (key, _) in entries {
        if VALID_RESOURCE_ATTRIBUTES.contains(&key.as_str()) {
            continue;
        }
        if is_sam && SAM_RESOURCE_ATTRIBUTES.contains(&key.as_str()) {
            continue;
        }
        let key_span = span_index.get(&format!("Resources/{}/{}", name, key)).copied().unwrap_or(resource_span);
        out.push(crate::make_parse_defect_for_resource(
            "E3001",
            format!(
                "Resource '{}' has invalid property '{}'. Valid resource attributes: {}",
                name,
                key,
                valid_attributes_display(is_sam),
            ),
            key_span,
            name,
        ));
    }

    out
}

/// Builds the human-readable list of valid resource attributes for error messages.
fn valid_attributes_display(is_sam: bool) -> String {
    let mut attrs: Vec<&str> = VALID_RESOURCE_ATTRIBUTES.to_vec();
    if is_sam {
        attrs.extend_from_slice(SAM_RESOURCE_ATTRIBUTES);
    }
    attrs.join(", ")
}

fn collect_lifecycle_attribute_status(
    arena: &Arena,
    node: NodeRef,
    conditions: &ConditionModel,
    allows_no_value_omission: bool,
    assumptions: &mut Vec<(String, bool)>,
    status: &mut LifecycleAttributeStatus,
) {
    if matches!(conditions.satisfiability(assumptions), Satisfiability::Unsatisfiable) {
        return;
    }

    if !allows_no_value_omission {
        status.may_be_present = true;
        if matches!(
            arena.node(node),
            Node::Null | Node::Bool(_) | Node::Int(_) | Node::Float(_) | Node::String(_) | Node::List(_)
        ) && status.invalid_value.is_none()
        {
            status.invalid_value = Some(crate::message::render_value(&node_to_json(arena, node)));
        }
        return;
    }

    match arena.node(node) {
        Node::Intrinsic(IntrinsicFn::Ref(target)) if target == PSEUDO_NO_VALUE => {}
        Node::Intrinsic(IntrinsicFn::If(condition, when_true, when_false)) => {
            assumptions.push((condition.clone(), true));
            collect_lifecycle_attribute_status(
                arena,
                *when_true,
                conditions,
                allows_no_value_omission,
                assumptions,
                status,
            );
            assumptions.pop();
            assumptions.push((condition.clone(), false));
            collect_lifecycle_attribute_status(
                arena,
                *when_false,
                conditions,
                allows_no_value_omission,
                assumptions,
                status,
            );
            assumptions.pop();
        }
        Node::Intrinsic(_) | Node::Map(_) => status.may_be_present = true,
        Node::Null | Node::Bool(_) | Node::Int(_) | Node::Float(_) | Node::String(_) | Node::List(_) => {
            status.may_be_present = true;
            if status.invalid_value.is_none() {
                status.invalid_value = Some(crate::message::render_value(&node_to_json(arena, node)));
            }
        }
    }
}

fn resolve_resource(arena: &Arena, name: &str, node_ref: NodeRef, resolver: &mut Resolver) -> ResolvedResource {
    let entries = arena.as_map(node_ref).unwrap_or(&[]);
    let resource_type =
        entries.iter().find(|(k, _)| k == KEY_TYPE).and_then(|(_, v)| arena.as_str(*v)).unwrap_or("").to_string();
    let condition =
        entries.iter().find(|(k, _)| k == KEY_CONDITION).and_then(|(_, v)| arena.as_str(*v)).map(|s| s.to_string());
    let depends_on = entries
        .iter()
        .find(|(k, _)| k == KEY_DEPENDS_ON)
        .map(|(_, v)| match arena.node(*v) {
            Node::String(s) => vec![s.clone()],
            Node::List(items) => items.iter().filter_map(|r| arena.as_str(*r).map(|s| s.to_string())).collect(),
            _ => vec![],
        })
        .unwrap_or_default();
    let deletion_policy = entries.iter().find(|(key, _)| key == KEY_DELETION_POLICY).map(|(_, value)| {
        resolver.set_current_path(KEY_DELETION_POLICY);
        resolver.resolve_node(*value)
    });
    let update_replace_policy = entries.iter().find(|(key, _)| key == KEY_UPDATE_REPLACE_POLICY).map(|(_, value)| {
        resolver.set_current_path(KEY_UPDATE_REPLACE_POLICY);
        resolver.resolve_node(*value)
    });
    let metadata_ref = entries.iter().find(|(k, _)| k == SECTION_METADATA).map(|(_, v)| *v);
    let metadata = metadata_ref.map(|v| node_to_json(arena, v));
    let update_policy = entries.iter().find(|(k, _)| k == KEY_UPDATE_POLICY).map(|(_, v)| node_to_json(arena, *v));
    let creation_policy = entries.iter().find(|(k, _)| k == KEY_CREATION_POLICY).map(|(_, v)| node_to_json(arena, *v));

    let resolved_metadata = if let Some(meta_ref) = metadata_ref {
        resolver.set_current_path(SECTION_METADATA);
        Some(resolver.resolve_node(meta_ref))
    } else {
        None
    };

    let mut properties = HashMap::new();
    let mut properties_dynamic = entries.iter().any(|(key, _)| key == FN_TRANSFORM);
    if let Some((_, props_ref)) = entries.iter().find(|(k, _)| k == KEY_PROPERTIES) {
        if let Some(prop_entries) = arena.as_map(*props_ref) {
            for (key, val_ref) in prop_entries {
                resolver.set_current_path(&format!("Properties.{}", key));
                properties.insert(key.clone(), resolver.resolve_node(*val_ref));
            }
        } else if let Some(synthetic_key) = intrinsic_synthetic_key(arena, *props_ref) {
            // `Properties: {Fn::If: [...]}` collapses in the IR into an intrinsic
            // node rather than a map. Preserve the intrinsic as a single entry so
            // downstream validators can enumerate each branch via scenario
            // resolution without losing the original conditional structure.
            // Use bare `Properties` as the base path so the Fn::If handler in
            // the resolver records branch sources at the single-prefix path
            // rather than doubling the `Fn::If` segment.
            resolver.set_current_path(KEY_PROPERTIES);
            properties.insert(synthetic_key, resolver.resolve_node(*props_ref));
        } else if matches!(
            arena.node(*props_ref),
            Node::Intrinsic(IntrinsicFn::Ref(target)) if target == PSEUDO_NO_VALUE
        ) && !properties_dynamic
        {
            // Retain the known non-object value so schema validation can reject
            // it without treating every schema property as missing.
            resolver.set_current_path(KEY_PROPERTIES);
            resolver.resolve_node(*props_ref);
            properties_dynamic = true;
        } else if matches!(arena.node(*props_ref), Node::Intrinsic(_)) {
            // The intrinsic may produce the effective property map only at
            // deployment, so static analysis cannot infer missing keys.
            properties_dynamic = true;
        }
    }

    let mut conditionally_null_props = Vec::new();
    for (key, val) in &properties {
        collect_conditional_nulls(val, key, &mut conditionally_null_props);
    }

    let mut condition_refs = Vec::new();
    for v in properties.values() {
        collect_condition_refs_from_resolved(v, &mut condition_refs);
    }
    if let Some(ref meta_resolved) = resolved_metadata {
        collect_condition_refs_from_resolved(meta_resolved, &mut condition_refs);
    }
    if let Some(ref dp) = deletion_policy {
        collect_condition_refs_from_resolved(dp, &mut condition_refs);
    }
    if let Some(ref urp) = update_replace_policy {
        collect_condition_refs_from_resolved(urp, &mut condition_refs);
    }
    if let Some(mut extra) = resolver.extra_condition_refs.remove(name) {
        condition_refs.append(&mut extra);
    }
    condition_refs.sort();
    condition_refs.dedup();

    ResolvedResource {
        logical_id: name.to_string(),
        resource_type,
        condition,
        depends_on,
        deletion_policy,
        update_replace_policy,
        update_policy: update_policy.map(JsonValue),
        creation_policy: creation_policy.map(JsonValue),
        metadata: metadata.map(JsonValue),
        properties,
        properties_dynamic,
        diagnostics: ResourceDiagnostics {
            find_in_map_refs: resolver.find_in_map_refs.remove(name).unwrap_or_default(),
            simple_subs: resolver
                .simple_subs
                .remove(name)
                .unwrap_or_default()
                .into_iter()
                .map(|(a, b)| PathValuePair { path: a, value: b })
                .collect(),
            redundant_subs: resolver.redundant_subs.remove(name).unwrap_or_default(),
            empty_joins: resolver.empty_joins.remove(name).unwrap_or_default(),
            hardcoded_partition_arns: collapse_list_sibling_arn_paths(
                resolver.hardcoded_partition_arns.remove(name).unwrap_or_default(),
            ),
            foreach_expansions: resolver
                .foreach_expansions
                .remove(name)
                .unwrap_or_default()
                .into_iter()
                .map(|(path, ident, coll)| ForEachExpansion {
                    property_path: path,
                    identifier: ident,
                    collection_source: coll,
                })
                .collect(),
            unsubstituted_variables: resolver
                .unsubstituted_variables
                .remove(name)
                .unwrap_or_default()
                .into_iter()
                .map(|(a, b)| PathValuePair { path: a, value: b })
                .collect(),
            unused_sub_keys: resolver
                .unused_sub_keys
                .remove(name)
                .unwrap_or_default()
                .into_iter()
                .map(|(a, b)| PathValuePair { path: a, value: b })
                .collect(),
            raw_pseudo_params: resolver
                .raw_pseudo_params
                .remove(name)
                .unwrap_or_default()
                .into_iter()
                .map(|(a, b)| PathValuePair { path: a, value: b })
                .collect(),
            secretsmanager_ref_paths: resolver.secretsmanager_ref_paths.remove(name).unwrap_or_default(),
            invalid_refs: resolver
                .invalid_refs
                .remove(name)
                .unwrap_or_default()
                .into_iter()
                .map(|(a, b)| PathValuePair { path: a, value: b })
                .collect(),
            conditionally_null_props: conditionally_null_props
                .into_iter()
                .map(|(a, b, c)| ConditionalNullEntry { path: a, condition: b, null_in_true_branch: c })
                .collect(),
            condition_refs,
        },
    }
}

/// Collapses hardcoded-partition ARN paths that are list siblings sharing one
/// source location into a single path at the lowest index.
///
/// When several `Fn::Sub` ARNs are list elements of the same property (e.g.
/// `Principal.AWS.0.Fn::Sub`, `Principal.AWS.1.Fn::Sub`), they map to the same
/// source span - the list's location - so they are one observable finding, not
/// several. Paths whose only difference is the list index immediately before the
/// trailing `.Fn::Sub` segment are therefore folded to the smallest index.
fn collapse_list_sibling_arn_paths(paths: Vec<String>) -> Vec<String> {
    // Group key: the path with the final list index (the segment just before
    // `.Fn::Sub`) blanked out. Tracks the minimum index seen per group so the
    // surviving path reports the first sibling.
    let mut min_index: HashMap<String, usize> = HashMap::new();
    let mut ungrouped: Vec<String> = Vec::new();
    for path in &paths {
        match split_trailing_list_index(path) {
            Some((prefix, idx, suffix)) => {
                let key = format!("{}\u{0}{}", prefix, suffix);
                min_index.entry(key).and_modify(|m| *m = (*m).min(idx)).or_insert(idx);
            }
            None => ungrouped.push(path.clone()),
        }
    }
    let mut out = ungrouped;
    for (key, idx) in min_index {
        let (prefix, suffix) = key.split_once('\u{0}').unwrap_or((key.as_str(), ""));
        out.push(format!("{}{}.{}", prefix, idx, suffix));
    }
    out.sort();
    out
}

/// Splits a path like `a.b.2.Fn::Sub` into (`"a.b."`, `2`, `"Fn::Sub"`) when a
/// numeric list index directly precedes the trailing `.Fn::Sub` segment.
fn split_trailing_list_index(path: &str) -> Option<(String, usize, String)> {
    let suffix = "Fn::Sub";
    let stem = path.strip_suffix(suffix)?.strip_suffix('.')?;
    let (head, last) = stem.rsplit_once('.')?;
    let idx: usize = last.parse().ok()?;
    Some((format!("{}.", head), idx, suffix.to_string()))
}

/// An output `Value` must resolve to a string. A literal non-empty list or map,
/// or a function whose result is a list (`Fn::GetAZs`, `Fn::Split`, `Fn::Cidr`),
/// can never be a string and is a guaranteed template error. `Fn::If` is
/// transparent: each branch is checked at its own location, so a conditional
/// that picks a string in one branch and a list in the other is flagged only on
/// the offending branch. Empty containers are accepted (they carry no members to
/// stringify) and every other function (`Ref`, `Fn::Sub`, `Fn::Join`,
/// `Fn::Select`, `Fn::FindInMap`, `Fn::ImportValue`, …) is treated as
/// string-producing here - a `Fn::GetAtt` returning a non-string is caught later
/// against the resource schema, which this parse-time shape check cannot see.
fn check_output_value_is_string(arena: &Arena, value_ref: NodeRef, output_name: &str, resolver: &mut Resolver) {
    let build_path = format!("Outputs/{}/Value", output_name);
    check_output_value_node(arena, value_ref, output_name, &build_path, resolver);
}

/// Recursive worker for [`check_output_value_is_string`]. `build_path` is the
/// slash path of the node under inspection; each `Fn::If` branch is visited with
/// its own branch path so that a conditional with a bad value in both branches
/// yields a separate, correctly-located diagnostic per branch rather than
/// collapsing to one.
fn check_output_value_node(
    arena: &Arena,
    value_ref: NodeRef,
    output_name: &str,
    build_path: &str,
    resolver: &mut Resolver,
) {
    let span = arena.span(value_ref);
    match arena.node(value_ref) {
        Node::List(items) if !items.is_empty() => {
            resolver.diagnostics.push(crate::make_parse_defect_at(
                "F6101",
                format!("Output '{}' value must be a string, not a list", output_name),
                span,
                build_path,
            ));
        }
        Node::Map(entries) if !entries.is_empty() => {
            resolver.diagnostics.push(crate::make_parse_defect_at(
                "F6101",
                format!("Output '{}' value must be a string, not an object", output_name),
                span,
                build_path,
            ));
        }
        Node::Intrinsic(IntrinsicFn::If(_, if_true, if_false) | IntrinsicFn::IfExpr(_, if_true, if_false)) => {
            let (if_true, if_false) = (*if_true, *if_false);
            check_output_value_node(arena, if_true, output_name, &format!("{}/{}/1", build_path, FN_IF), resolver);
            check_output_value_node(arena, if_false, output_name, &format!("{}/{}/2", build_path, FN_IF), resolver);
        }
        Node::Intrinsic(intrinsic) if returns_list(intrinsic) => {
            resolver.diagnostics.push(crate::make_parse_defect_at(
                "F6101",
                format!("Output '{}' value must be a string, not a list", output_name),
                span,
                build_path,
            ));
        }
        _ => {}
    }
}

/// Whether an intrinsic's result is a list value (never a string). These are the
/// only list-returning functions that can stand as a whole output `Value`;
/// `Fn::If` is handled separately (its branches are checked), and every other
/// function yields a string or an opaque deploy-time value.
fn returns_list(intrinsic: &IntrinsicFn) -> bool {
    matches!(intrinsic, IntrinsicFn::GetAZs(_) | IntrinsicFn::Split(_, _) | IntrinsicFn::Cidr(_, _, _))
}

fn resolve_output(arena: &Arena, node_ref: NodeRef, resolver: &mut Resolver) -> ResolvedOutput {
    let entries = arena.as_map(node_ref).unwrap_or(&[]);

    // Validate output property keys
    const VALID_OUTPUT_KEYS: &[&str] = &[KEY_VALUE, SECTION_DESCRIPTION, KEY_CONDITION, KEY_EXPORT];
    let output_name = resolver.current_resource.as_deref().unwrap_or("");
    let display_name = output_name.strip_prefix(OUTPUT_PSEUDO_RESOURCE_PREFIX).unwrap_or(output_name);
    for (key, _) in entries {
        if !VALID_OUTPUT_KEYS.contains(&key.as_str()) {
            resolver.diagnostics.push(crate::make_parse_defect(
                "E6001",
                format!(
                    "Output '{}' has invalid property '{}'. Valid properties: Value, Description, Condition, Export",
                    display_name, key
                ),
                crate::ir::UNKNOWN_SPAN,
            ));
        }
    }

    if let Some((_, value_ref)) = entries.iter().find(|(k, _)| k == KEY_VALUE) {
        let value_ref = *value_ref;
        let name = display_name.to_string();
        check_output_value_is_string(arena, value_ref, &name, resolver);
    }

    let value = entries
        .iter()
        .find(|(k, _)| k == KEY_VALUE)
        .map(|(_, v)| resolver.resolve_node(*v))
        .unwrap_or(ResolvedValue::Dynamic { reason: "missing output value".into() });
    let description = entries
        .iter()
        .find(|(k, _)| k == SECTION_DESCRIPTION)
        .and_then(|(_, v)| arena.as_str(*v))
        .map(|s| s.to_string());
    let condition =
        entries.iter().find(|(k, _)| k == KEY_CONDITION).and_then(|(_, v)| arena.as_str(*v)).map(|s| s.to_string());
    let export_name = entries
        .iter()
        .find(|(k, _)| k == KEY_EXPORT)
        .and_then(|(_, v)| arena.as_map(*v))
        .and_then(|m| m.iter().find(|(k, _)| k == KEY_NAME))
        .map(|(_, v)| resolver.resolve_node(*v));
    ResolvedOutput { value, description, condition, export_name }
}

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

    #[test]
    fn invalid_overrides_reports_nothing_when_unset_or_valid() {
        assert!(PseudoParameterOverrides::default().invalid_overrides().is_empty(), "unset overrides are valid");
        let valid = PseudoParameterOverrides {
            account_id: Some("123456789012".to_string()),
            partition: Some("aws-cn".to_string()),
            ..Default::default()
        };
        assert!(valid.invalid_overrides().is_empty(), "well-formed overrides produce no warnings");
    }

    #[test]
    fn invalid_overrides_reports_each_malformed_provided_value() {
        let bad = PseudoParameterOverrides {
            account_id: Some("unknown-account".to_string()),
            partition: Some("gcp".to_string()),
            ..Default::default()
        };
        let problems = bad.invalid_overrides();
        assert_eq!(problems.len(), 2, "one entry per invalid provided field");
        assert!(problems.iter().any(|p| p.contains("AccountId") && p.contains("unknown-account")));
        assert!(problems.iter().any(|p| p.contains("Partition") && p.contains("gcp")));
    }

    #[test]
    fn invalid_overrides_only_checks_account_id_and_partition() {
        let only_region = PseudoParameterOverrides { region: Some("not a region".to_string()), ..Default::default() };
        assert!(only_region.invalid_overrides().is_empty());
    }

    #[test]
    fn account_id_shape_requires_exactly_twelve_digits() {
        assert!(is_account_id_shaped("123456789012"));
        assert!(!is_account_id_shaped("12345678901"));
        assert!(!is_account_id_shaped("1234567890123"));
        assert!(!is_account_id_shaped("12345678901a"));
        assert!(!is_account_id_shaped(""));
    }

    #[test]
    fn partition_shape_accepts_aws_family_and_rejects_others() {
        for ok in ["aws", "aws-cn", "aws-us-gov", "aws-iso", "aws-iso-b"] {
            assert!(is_partition_shaped(ok), "{ok} should be accepted");
        }
        for bad in ["gcp", "AWS", "azure", ""] {
            assert!(!is_partition_shaped(bad), "{bad} should be rejected");
        }
    }

    #[test]
    fn model_from_simple_template() {
        let input = r#"
AWSTemplateFormatVersion: "2010-09-09"
Resources:
  MyBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: my-bucket
  MyBucket2:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: my-bucket-2
"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert_eq!(model.resources.len(), 2);
        assert_eq!(model.resources_of_type("AWS::S3::Bucket").len(), 2);
        assert_eq!(model.resources_of_type("AWS::Fake::Thing").len(), 0);
    }

    /// Resource `R` mixes literal and intrinsic-sourced properties; `IfR` wraps
    /// its whole `Properties` block in an `Fn::If` whose first branch sets
    /// `TopicName` via `Ref` and `DisplayName` to a literal. Exercises every
    /// `is_from_intrinsic` path: reference edges, concretely-resolved intrinsics,
    /// ancestor inheritance, `Fn::If` branch attribution, and plain literals.
    fn intrinsic_source_probe_model() -> SemanticModel {
        let input = r#"
Resources:
  Other:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: other-bucket
  R:
    Type: AWS::SNS::Topic
    Properties:
      LiteralProp: plain-literal
      RefProp: !Ref Other
      GetAttProp: !GetAtt Other.Arn
      JoinProp: !Join ["-", ["a", "b"]]
      NestedRef:
        Inner: !Ref Other
  IfR:
    Type: AWS::SNS::Topic
    Condition: MakeIt
    Properties:
      Fn::If:
        - MakeIt
        - TopicName: !Ref Other
          DisplayName: literal-in-true-branch
          BranchLiteral: literal-one
        - TopicName: plain-literal-two
          DisplayName: !Ref Other
          BranchLiteral: literal-two
Conditions:
  MakeIt: !Equals ["a", "a"]
"#;
        SemanticModel::from_bytes(input.as_bytes()).unwrap()
    }

    #[test]
    fn is_from_intrinsic_separates_intrinsic_values_from_literals() {
        let model = intrinsic_source_probe_model();
        assert!(model.is_from_intrinsic("R", "Properties.RefProp"), "a Ref value is intrinsic-sourced");
        assert!(model.is_from_intrinsic("R", "Properties.GetAttProp"), "a GetAtt value is intrinsic-sourced");
        assert!(
            model.is_from_intrinsic("R", "Properties.JoinProp"),
            "an Fn::Join value keeps its Intrinsic/* source even when it resolves concretely"
        );
        assert!(!model.is_from_intrinsic("R", "Properties.LiteralProp"), "a plain literal is not intrinsic-sourced");
    }

    #[test]
    fn is_from_intrinsic_inherits_from_intrinsic_ancestors() {
        let model = intrinsic_source_probe_model();
        // The Ref lives at Properties.NestedRef.Inner; that exact path and any
        // descendant of it are intrinsic-sourced through ancestor inheritance.
        assert!(model.is_from_intrinsic("R", "Properties.NestedRef.Inner"));
        assert!(model.is_from_intrinsic("R", "Properties.NestedRef.Inner.Deeper"));
        // The enclosing map itself holds no intrinsic; only its Inner value does.
        assert!(!model.is_from_intrinsic("R", "Properties.NestedRef"));
        // A resource that anchors no intrinsic paths is never intrinsic-sourced.
        assert!(!model.is_from_intrinsic("Missing", "Properties.Anything"));
    }

    #[test]
    fn is_from_intrinsic_attributes_fn_if_branch_sources_to_the_bare_path() {
        let model = intrinsic_source_probe_model();
        // The true branch sets TopicName via Ref, so the bare property path is
        // intrinsic-sourced through the branch-qualified index entry.
        assert!(model.is_from_intrinsic("IfR", "Properties.TopicName"));
        // The false branch independently sets DisplayName via Ref.
        assert!(model.is_from_intrinsic("IfR", "Properties.DisplayName"));
        // A property that is literal in both branches remains non-intrinsic.
        assert!(!model.is_from_intrinsic("IfR", "Properties.BranchLiteral"));
    }

    #[test]
    fn diagnostic_span_resolves_dotted_resource_property_path() {
        let input = r#"
Resources:
  MyBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: my-bucket
"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        // A dotted, resource-relative path is normalized to the slash-keyed index
        // entry and resolves to the specific property span (the BucketName value).
        let via_diag = model.diagnostic_span(Some("MyBucket"), "Properties.BucketName").expect("should resolve");
        let expected = model.source_location("Resources/MyBucket/Properties/BucketName").copied().expect("indexed");
        assert_eq!(via_diag, expected, "dotted path should resolve to the exact property span");
    }

    #[test]
    fn diagnostic_span_resolves_bare_top_level_path_without_resource() {
        let input = r#"
BogusSection:
  Value: invalid
Resources:
  MyBucket:
    Type: AWS::S3::Bucket
"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        let via_diag = model.diagnostic_span(None, "BogusSection").expect("should resolve");
        let expected = model.source_location("BogusSection").copied().expect("indexed");
        assert_eq!(via_diag, expected, "bare top-level path should resolve to its authored key");
    }

    #[test]
    fn diagnostic_span_resource_relative_path_never_matches_top_level_section() {
        // A resource property named after a top-level section (Metadata) must anchor
        // at the resource's own Metadata, not the unrelated top-level Metadata block.
        let input = r#"
Metadata:
  AWS::CloudFormation::Interface:
    ParameterGroups: []
Resources:
  MyBucket:
    Type: AWS::S3::Bucket
    Metadata:
      cfn_nag: skip
"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        let via_diag = model.diagnostic_span(Some("MyBucket"), "Metadata").expect("should resolve");
        let resource_meta = model.source_location("Resources/MyBucket/Metadata").copied().expect("indexed");
        let top_level_meta = model.source_location("Metadata").copied().expect("indexed");
        assert_eq!(via_diag, resource_meta, "resource-relative Metadata must anchor at the resource");
        assert_ne!(via_diag, top_level_meta, "must not mislocate onto the top-level Metadata section");
    }

    #[test]
    fn diagnostic_span_resolves_absolute_section_rooted_path() {
        let input = r#"
Resources:
  MyBucket:
    Type: AWS::S3::Bucket
Outputs:
  BucketRef:
    Value: !Ref MyBucket
"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        // An output-anchored path cannot be reached via resource_span (it only roots
        // at Resources/), but diagnostic_span resolves it directly from the index.
        let span = model.diagnostic_span(Some("BucketRef"), "Outputs/BucketRef/Value");
        assert!(span.is_some() && span != Some(UNKNOWN_SPAN), "output path should resolve to a real span");
    }

    #[test]
    fn diagnostic_span_walks_up_when_leaf_node_is_unindexed() {
        let input = r#"
Resources:
  MyBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: my-bucket
"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        // A deeper path that was never indexed falls back to the nearest indexed
        // ancestor (the property, then the resource) rather than UNKNOWN.
        let span = model.diagnostic_span(Some("MyBucket"), "Properties.BucketName.DoesNotExist.Deeper");
        assert!(span.is_some() && span != Some(UNKNOWN_SPAN), "unindexed leaf should walk up to a real span");
    }

    #[test]
    fn diagnostic_span_returns_none_when_nothing_resolvable() {
        let input = r#"
Resources:
  MyBucket:
    Type: AWS::S3::Bucket
"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        // No resource and no property path: there is no candidate key to resolve.
        assert_eq!(model.diagnostic_span(None, ""), None, "no resource and no path yields no span");
        // A property path whose every ancestor is unindexed (no matching section)
        // resolves to nothing.
        assert_eq!(
            model.diagnostic_span(None, "NoSuchSection/Deep/Path"),
            None,
            "an unindexed absolute path yields no span"
        );
    }

    #[test]
    fn model_resolve_property() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"Name":"hello"}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        match model.resolve("R", "Properties.Name").unwrap() {
            ResolvedValue::Concrete { value: v } => assert_eq!(v.as_str().unwrap(), "hello"),
            other => panic!("Expected Concrete, got {:?}", other),
        }
    }

    #[test]
    fn model_follow_ref() {
        let input = r#"{"Resources":{"Svc":{"Type":"T","Properties":{"TaskDef":{"Ref":"TD"}}},"TD":{"Type":"T2"}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert_eq!(model.follow_ref("Svc", "Properties.TaskDef"), Some("TD"));
    }

    #[test]
    fn model_to_diagnostic_json() {
        let input = r#"{"Resources":{"R":{"Type":"AWS::S3::Bucket","Properties":{"Name":"test"}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        let json = serde_json::to_value(model.to_diagnostic_json()).unwrap();
        assert_ne!(json.get("resources"), None, "expected 'resources' key in diagnostic JSON");
        assert_ne!(json["resources"].get("R"), None, "expected resource 'R' in diagnostic JSON");
    }

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

    #[test]
    fn model_with_conditions() {
        let input = r#"
Parameters:
  Env:
    Type: String
    AllowedValues: [Prod, Dev]
Conditions:
  IsProd:
    Fn::Equals: [!Ref Env, Prod]
Resources:
  R:
    Type: T
    Condition: IsProd
    Properties:
      Size:
        Fn::If: [IsProd, 100, 20]
"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert_eq!(model.conditions.conditions.len(), 1);
        let r = model.resource("R").unwrap();
        assert_eq!(r.condition.as_deref(), Some("IsProd"));
        assert!(matches!(
            r.properties.get("Size"),
            Some(ResolvedValue::Conditional { condition: _, if_true: _, if_false: _ })
        ));
    }

    #[test]
    fn model_with_outputs() {
        let input =
            "Resources:\n  R:\n    Type: T\nOutputs:\n  Out1:\n    Value: !Ref R\n    Description: test output\n";
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert_eq!(model.outputs.len(), 1);
        assert!(model.outputs.contains_key("Out1"));
    }

    #[test]
    fn resolve_deep_nested_object() {
        let input =
            r#"{"Resources":{"R":{"Type":"T","Properties":{"Config":{"SubKey":"value","Nested":{"Deep":"found"}}}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        match model.resolve_deep("R", "Properties.Config.SubKey") {
            Some(ResolvedValue::Concrete { value: v }) => assert_eq!(v.as_str().unwrap(), "value"),
            other => panic!("Expected Concrete, got {:?}", other),
        }
        match model.resolve_deep("R", "Properties.Config.Nested.Deep") {
            Some(ResolvedValue::Concrete { value: v }) => assert_eq!(v.as_str().unwrap(), "found"),
            other => panic!("Expected Concrete, got {:?}", other),
        }
    }

    #[test]
    fn resolve_deep_array_index() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"Tags":[{"Key":"Env","Value":"prod"},{"Key":"App","Value":"web"}]}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        match model.resolve_deep("R", "Properties.Tags.0.Key") {
            Some(ResolvedValue::Concrete { value: v }) => assert_eq!(v.as_str().unwrap(), "Env"),
            other => panic!("Expected Concrete, got {:?}", other),
        }
    }

    #[test]
    fn resolve_deep_out_of_bounds() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"Tags":[{"Key":"A"}]}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert!(model.resolve_deep("R", "Properties.Tags.5.Key").is_none(), "out-of-bounds index should return None");
    }

    #[test]
    fn resolve_deep_missing_intermediate() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"Name":"hello"}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert!(model.resolve_deep("R", "Properties.NonExistent.Sub").is_none());
    }

    #[test]
    fn resolve_deep_top_level_still_works() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"Name":"hello"}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        match model.resolve_deep("R", "Properties.Name") {
            Some(ResolvedValue::Concrete { value: v }) => assert_eq!(v.as_str().unwrap(), "hello"),
            other => panic!("Expected Concrete, got {:?}", other),
        }
    }

    #[test]
    fn resolve_unexpanded_preserves_whole_properties_and_nested_conditions_without_spending_budget() {
        let input = br#"{
            "Parameters": {"Mode": {"Type": "String"}},
            "Conditions": {
                "ChooseProperties": {"Fn::Equals": [{"Ref": "Mode"}, "properties"]},
                "ChooseValue": {"Fn::Equals": [{"Ref": "Mode"}, "value"]}
            },
            "Resources": {
                "Direct": {"Type": "T", "Properties": {"Values": ["direct"]}},
                "Conditional": {"Type": "T", "Properties": {
                    "Fn::If": [
                        "ChooseProperties",
                        {"Values": [{"Fn::If": ["ChooseValue", "first", "second"]}]},
                        {"Values": ["fallback"]}
                    ]
                }}
            }
        }"#;
        let model = SemanticModel::from_bytes(input).unwrap();
        let direct = model.resolve_unexpanded("Direct", "Properties.Values").expect("direct value");
        let direct_deep = model.resolve_deep("Direct", "Properties.Values").expect("direct deep value");
        assert_eq!(resolved_value_to_json(&direct), serde_json::json!(["direct"]));
        assert_eq!(resolved_value_to_json(&direct_deep), serde_json::json!(["direct"]));

        let conditional = model
            .resolve_unexpanded("Conditional", "Properties.Values")
            .expect("value inside whole-Properties condition");
        let ResolvedValue::Conditional { condition, if_true, if_false } = conditional else {
            panic!("expected the whole-Properties condition");
        };
        assert_eq!(condition, "ChooseProperties");
        let ResolvedValue::List { items: true_items } = *if_true else {
            panic!("expected true-branch list");
        };
        assert!(
            matches!(true_items.as_slice(), [ResolvedValue::Conditional { condition, .. }] if condition == "ChooseValue")
        );
        assert_eq!(resolved_value_to_json(if_false.as_ref()), serde_json::json!(["fallback"]));
        assert_eq!(model.scenario_combinations_used(), 0, "an unexpanded lookup must not materialize scenarios");
    }

    #[test]
    fn list_member_scenarios_expand_members_independently_and_reuse_cache() {
        const MEMBER_COUNT: usize = 13;
        let mut parameters = serde_json::Map::new();
        let mut conditions = serde_json::Map::new();
        let mut members = Vec::new();
        for index in 0..MEMBER_COUNT {
            let parameter = format!("Toggle{index:02}");
            let condition = format!("Enabled{index:02}");
            parameters.insert(parameter.clone(), serde_json::json!({"Type": "String"}));
            conditions.insert(condition.clone(), serde_json::json!({"Fn::Equals": [{"Ref": parameter}, "true"]}));
            members.push(serde_json::json!({
                "Fn::If": [condition, format!("value-{index:02}"), {"Ref": "AWS::NoValue"}]
            }));
        }
        let template = serde_json::json!({
            "Parameters": parameters,
            "Conditions": conditions,
            "Resources": {"R": {"Type": "T", "Properties": {"Values": members}}}
        })
        .to_string();
        let model = SemanticModel::from_bytes(template.as_bytes()).unwrap();

        let first = model.resolve_list_member_scenarios_shared("R", "Properties.Values");
        let combinations_after_first = model.scenario_combinations_used();
        let second = model.resolve_list_member_scenarios_shared("R", "Properties.Values");

        assert_eq!(first.len(), MEMBER_COUNT * 2, "each member contributes only its own two branches");
        for index in 0..MEMBER_COUNT {
            let member_scenarios: Vec<&ListMemberScenario> =
                first.iter().filter(|scenario| scenario.index == index).collect();
            assert_eq!(member_scenarios.len(), 2, "member {index} must retain both branches");
            assert!(member_scenarios.iter().all(|scenario| scenario.container_scenario == 0));
            assert!(member_scenarios.iter().all(|scenario| scenario.container_conditions.is_empty()));
            assert!(member_scenarios.iter().all(|scenario| scenario.conditions.len() == 1));
        }
        assert_eq!(combinations_after_first, 0, "member extraction must not materialize joint scenarios");
        assert!(Arc::ptr_eq(&first, &second), "repeated member queries must share their cached allocation");
        assert_eq!(model.scenario_combinations_used(), combinations_after_first);
    }

    #[test]
    fn list_member_scenarios_keep_outer_enum_alternatives_separate() {
        let model = SemanticModel::from_bytes(br#"{"Resources":{"R":{"Type":"T"}}}"#).unwrap();
        let value = ResolvedValue::Enum {
            variants: vec![
                ResolvedValue::Concrete { value: serde_json::json!(["left", null]).into() },
                ResolvedValue::Concrete { value: serde_json::json!([null, "right"]).into() },
            ],
        };
        let mut next_container_scenario = 0;
        let mut member_scenarios = Vec::new();

        model.collect_list_member_scenarios(
            &value,
            &HashMap::new(),
            &mut next_container_scenario,
            &mut member_scenarios,
        );

        assert_eq!(next_container_scenario, 2);
        assert_eq!(member_scenarios.len(), 4);
        for container_scenario in 0..2 {
            let members: Vec<&ListMemberScenario> =
                member_scenarios.iter().filter(|scenario| scenario.container_scenario == container_scenario).collect();
            assert_eq!(members.len(), 2);
            assert_eq!(members.iter().filter(|scenario| !is_resolved_null(&scenario.value)).count(), 1);
        }
    }

    #[test]
    fn list_member_pair_source_paths_collapse_value_branches() {
        let input = br#"{
            "Parameters": {"Mode": {"Type": "String"}},
            "Conditions": {"ChooseFirst": {"Fn::Equals": [{"Ref": "Mode"}, "first"]}},
            "Resources": {
                "Invalid": {"Type": "T", "Properties": {
                    "Values": {"Fn::If": ["ChooseFirst", ["a", "b"], ["c", "d"]]}
                }},
                "Valid": {"Type": "T", "Properties": {
                    "Values": {"Fn::If": ["ChooseFirst", ["a"], ["b"]]}
                }}
            }
        }"#;
        let model = SemanticModel::from_bytes(input).unwrap();

        assert_eq!(
            model.list_member_pair_source_paths("Invalid", "Properties.Values", &HashMap::new()),
            BTreeSet::from(["Properties.Values".to_string()])
        );
        assert!(
            model.list_member_pair_source_paths("Valid", "Properties.Values", &HashMap::new()).is_empty(),
            "members from mutually exclusive one-item branches must not form a pair"
        );
    }

    #[test]
    fn model_findinmap_refs_tracked() {
        let input = br#"{"Resources":{"R":{"Type":"T","Properties":{"V":{"Fn::FindInMap":["MyMap","k1","k2"]}}}}}"#;
        let model = SemanticModel::from_bytes(input).unwrap();
        assert!(model.resource("R").unwrap().diagnostics.find_in_map_refs.contains(&"MyMap".to_string()));
    }

    #[test]
    fn model_findinmap_refs_tracked_yaml() {
        let input = b"Resources:\n  R:\n    Type: T\n    Properties:\n      V: !FindInMap [MyMap, k1, k2]\n";
        let model = SemanticModel::from_bytes(input).unwrap();
        assert!(model.resource("R").unwrap().diagnostics.find_in_map_refs.contains(&"MyMap".to_string()));
    }

    #[test]
    fn model_to_diagnostic_json_has_outputs_and_mappings() {
        let input = b"AWSTemplateFormatVersion: '2010-09-09'\nMappings:\n  M:\n    k1:\n      k2: val\nResources:\n  R:\n    Type: AWS::S3::Bucket\nOutputs:\n  Out:\n    Value: !Ref R\n";
        let model = SemanticModel::from_bytes(input).unwrap();
        let json = serde_json::to_value(model.to_diagnostic_json()).unwrap();
        assert_ne!(json.get("outputs"), None, "expected 'outputs' key in diagnostic JSON");
        assert_ne!(json.get("mappings"), None, "expected 'mappings' key in diagnostic JSON");
        assert_eq!(json["mappings"]["M"]["k1"]["k2"], "val");
    }

    #[test]
    fn resolve_deep_into_reference() {
        let input = r#"{"Resources":{"Svc":{"Type":"T","Properties":{"TaskDef":{"Ref":"TD"}}},"TD":{"Type":"T2"}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert!(
            model.resolve_deep("Svc", "Properties.TaskDef.Something").is_none(),
            "path through Reference should return None"
        );
        match model.resolve_deep("Svc", "Properties.TaskDef") {
            Some(ResolvedValue::Reference { target, kind: _ }) => assert_eq!(target, "TD"),
            other => panic!("Expected Reference, got {:?}", other),
        }
    }

    #[test]
    fn scenario_source_path_maps_effective_indexes_to_authored_branches() {
        let input = br#"{
            "Parameters": {"Mode": {"Type": "String"}},
            "Conditions": {"ChooseFirst": {"Fn::Equals": [{"Ref": "Mode"}, "first"]}},
            "Resources": {"R": {"Type": "T", "Properties": {
                "Values": {"Fn::If": ["ChooseFirst", ["bad"], ["good"]]}
            }}}
        }"#;
        let model = SemanticModel::from_bytes(input).unwrap();
        let true_conditions = HashMap::from([("ChooseFirst".to_string(), true)]);
        let false_conditions = HashMap::from([("ChooseFirst".to_string(), false)]);
        assert_eq!(
            model.scenario_source_path("R", "Properties.Values.0", &true_conditions),
            Some("Properties.Values.Fn::If.1.0".to_string())
        );
        assert_eq!(
            model.scenario_source_path("R", "Properties.Values.0", &false_conditions),
            Some("Properties.Values.Fn::If.2.0".to_string())
        );
    }

    #[test]
    fn resolve_scenarios_json_filters_dynamic() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"V":{"Fn::ImportValue":"Stack"}}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        let scenarios = model.resolve_scenarios_json("R", "Properties.V");
        assert!(scenarios.is_empty(), "ImportValue should be filtered as dynamic");
    }

    #[test]
    fn resolve_scenarios_json_returns_concrete() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"V":"hello"}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        let scenarios = model.resolve_scenarios_json("R", "Properties.V");
        assert_eq!(scenarios.len(), 1);
        assert_eq!(scenarios[0].0, serde_json::json!("hello"));
    }

    #[test]
    fn resolve_scenarios_json_memoized() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"V":"hello"}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        let s1 = model.resolve_scenarios_json("R", "Properties.V");
        let s2 = model.resolve_scenarios_json("R", "Properties.V");
        assert_eq!(s1, s2);
    }

    const CONDITIONAL_PROPERTY_TEMPLATE: &[u8] = br#"{
        "Parameters": {"Mode": {"Type": "String"}},
        "Conditions": {"ChooseFirst": {"Fn::Equals": [{"Ref": "Mode"}, "first"]}},
        "Resources": {"R": {"Type": "T", "Properties": {
            "V": {"Fn::If": ["ChooseFirst", "a", "b"]}
        }}}
    }"#;

    #[test]
    fn resolve_scenarios_json_shared_reuses_allocation_and_matches_owned() {
        let model = SemanticModel::from_bytes(CONDITIONAL_PROPERTY_TEMPLATE).unwrap();

        let first = model.resolve_scenarios_json_shared("R", "Properties.V");
        let combinations_after_first = model.scenario_combinations_used();
        let second = model.resolve_scenarios_json_shared("R", "Properties.V");
        let owned = model.resolve_scenarios_json("R", "Properties.V");

        assert!(Arc::ptr_eq(&first, &second));
        assert_eq!(first.as_ref(), owned.as_slice());
        assert!(combinations_after_first > 0);
        assert_eq!(model.scenario_combinations_used(), combinations_after_first);
    }

    #[test]
    fn resolve_properties_scenarios_shared_reuses_allocation_and_matches_owned() {
        let model = SemanticModel::from_bytes(CONDITIONAL_PROPERTY_TEMPLATE).unwrap();

        let first = model.resolve_properties_scenarios_shared("R");
        let combinations_after_first = model.scenario_combinations_used();
        let second = model.resolve_properties_scenarios_shared("R");
        let owned = model.resolve_properties_scenarios("R");

        assert!(Arc::ptr_eq(&first, &second), "repeated shared access must reuse one allocation");
        assert!(first.len() > 1, "the Fn::If value must expand the whole-properties object into multiple scenarios");
        assert_eq!(
            serde_json::to_value(first.as_ref()).expect("shared scenarios serialize"),
            serde_json::to_value(&owned).expect("owned scenarios serialize"),
            "the owned API must return the same scenarios as the shared accessor"
        );
        assert_eq!(
            model.scenario_combinations_used(),
            combinations_after_first,
            "a cached shared access must not consume additional scenario budget"
        );
    }

    #[test]
    fn resolve_scenarios_shared_reuses_allocation_and_matches_owned() {
        let model = SemanticModel::from_bytes(CONDITIONAL_PROPERTY_TEMPLATE).unwrap();

        let first = model.resolve_scenarios_shared("R", "Properties.V");
        let combinations_after_first = model.scenario_combinations_used();
        let second = model.resolve_scenarios_shared("R", "Properties.V");
        let owned = model.resolve_scenarios("R", "Properties.V");

        assert!(Arc::ptr_eq(&first, &second), "repeated shared access must reuse one allocation");
        assert_eq!(first.len(), 2, "the Fn::If value must expand into one scenario per branch");
        assert_eq!(
            serde_json::to_value(first.as_ref()).expect("shared scenarios serialize"),
            serde_json::to_value(&owned).expect("owned scenarios serialize"),
            "the owned API must return the same scenarios as the shared accessor"
        );
        assert_eq!(
            model.scenario_combinations_used(),
            combinations_after_first,
            "cached shared and owned access must not consume additional scenario budget"
        );
    }

    #[test]
    fn resolve_scenarios_with_limit_owned_matches_shared_including_curtailment() {
        let model = SemanticModel::from_bytes(CONDITIONAL_PROPERTY_TEMPLATE).unwrap();

        let (shared, shared_was_curtailed) = model.resolve_scenarios_with_limit_shared("R", "Properties.V", 1);
        let (repeated, repeated_was_curtailed) = model.resolve_scenarios_with_limit_shared("R", "Properties.V", 1);
        let (owned, owned_was_curtailed) = model.resolve_scenarios_with_limit("R", "Properties.V", 1);

        assert!(Arc::ptr_eq(&shared, &repeated), "a repeated limited query must reuse its memoized allocation");
        assert_eq!(shared.len(), 1, "a per-value limit of one must keep only the first branch scenario");
        assert!(shared_was_curtailed, "omitting the second branch scenario must be reported as curtailment");
        assert!(repeated_was_curtailed, "a memoized result must keep its curtailment status");
        assert!(owned_was_curtailed, "the owned API must report the memoized curtailment status");
        assert_eq!(
            serde_json::to_value(shared.as_ref()).expect("shared scenarios serialize"),
            serde_json::to_value(&owned).expect("owned scenarios serialize"),
            "the owned API must return the same scenarios as the shared accessor"
        );
    }

    #[test]
    fn resolve_scenarios_with_limit_shared_memoizes_each_limit_separately() {
        let model = SemanticModel::from_bytes(CONDITIONAL_PROPERTY_TEMPLATE).unwrap();

        let (limited, _) = model.resolve_scenarios_with_limit_shared("R", "Properties.V", 1);
        let default_limit = model.resolve_scenarios_shared("R", "Properties.V");
        let (explicit_default_limit, was_curtailed) =
            model.resolve_scenarios_with_limit_shared("R", "Properties.V", MAX_SCENARIO_COMBINATIONS);

        assert_eq!(limited.len(), 1, "a per-value limit of one must keep only the first branch scenario");
        assert_eq!(default_limit.len(), 2, "a curtailed limited expansion must not be served to a default-limit query");
        assert!(
            Arc::ptr_eq(&default_limit, &explicit_default_limit),
            "the default-limit accessor and an explicit default-limit query must share one memoized allocation"
        );
        assert!(!was_curtailed, "the complete default-limit expansion must not report curtailment");
    }

    #[test]
    fn default_scenario_queries_are_memoized_and_global_budget_halts_uncached_expansion() {
        let input = br#"{
            "Parameters": {"Env": {"Type": "String"}},
            "Conditions": {"IsProd": {"Fn::Equals": [{"Ref": "Env"}, "prod"]}},
            "Resources": {"R": {"Type": "T", "Properties": {
                "V": {"Fn::If": ["IsProd", "a", "b"]},
                "W": {"Fn::If": ["IsProd", "c", "d"]},
                "X": {"Fn::If": ["IsProd", "e", "f"]}
            }}}
        }"#;
        let model = SemanticModel::from_bytes(input).unwrap();

        assert_eq!(model.scenario_combinations_used(), 0, "a freshly built model has materialized no scenarios");
        assert!(!model.scenario_budget_exhausted());

        let first = model.resolve_scenarios("R", "Properties.V");
        assert!(first.len() > 1, "an Fn::If value must expand into multiple scenarios; got {}", first.len());
        let after_first = model.scenario_combinations_used();
        assert_eq!(after_first, first.len() as u64, "the first query charges exactly the scenarios it produced");

        let repeated = model.resolve_scenarios("R", "Properties.V");
        assert_eq!(repeated.len(), first.len());
        assert_eq!(
            model.scenario_combinations_used(),
            after_first,
            "a repeated default-limit query must reuse its raw scenarios without consuming budget"
        );

        let distinct = model.resolve_scenarios("R", "Properties.W");
        assert!(!distinct.is_empty());
        assert!(
            model.scenario_combinations_used() > after_first,
            "a distinct scenario path must still charge the model-wide budget"
        );

        let to_threshold = MAX_TOTAL_SCENARIO_COMBINATIONS - model.scenario_combinations_used();
        model.add_scenario_combinations_for_test(to_threshold);
        assert!(model.scenario_budget_exhausted());

        let cached = model.resolve_scenarios("R", "Properties.V");
        assert_eq!(
            cached.len(),
            first.len(),
            "cached scenarios remain usable after unrelated queries consume the budget"
        );
        assert!(!model.scenario_expansion_curtailed(), "reading a cached result does not curtail expansion");

        let before_short_circuit = model.scenario_combinations_used();
        let curtailed = model.resolve_scenarios("R", "Properties.X");
        assert!(curtailed.is_empty(), "an uncached query must stop when the model-wide budget is exhausted");
        assert_eq!(model.scenario_combinations_used(), before_short_circuit);
        assert!(model.scenario_expansion_curtailed());
    }

    fn two_scenario_value() -> ResolvedValue {
        ResolvedValue::Conditional {
            condition: "C".to_string(),
            if_true: Box::new(ResolvedValue::Concrete { value: serde_json::json!("yes").into() }),
            if_false: Box::new(ResolvedValue::Concrete { value: serde_json::json!("no").into() }),
        }
    }

    #[test]
    fn exact_remaining_scenario_budget_does_not_mark_curtailment() {
        let model = SemanticModel::from_bytes(br#"{"Resources":{"R":{"Type":"T"}}}"#).unwrap();
        let scenarios = model.collect_scenarios_with_budget(&two_scenario_value(), 8, 2);
        assert_eq!(scenarios.len(), 2);
        assert_eq!(model.scenario_combinations_used(), 2);
        assert!(
            !model.scenario_expansion_curtailed(),
            "materializing exactly every possible scenario at the budget boundary is not curtailment"
        );
    }

    #[test]
    fn remaining_scenario_budget_truncates_and_marks_curtailment() {
        let model = SemanticModel::from_bytes(br#"{"Resources":{"R":{"Type":"T"}}}"#).unwrap();
        let first = model.collect_scenarios_with_budget(&two_scenario_value(), 8, 3);
        assert_eq!(first.len(), 2);
        assert!(!model.scenario_expansion_curtailed());

        let second = model.collect_scenarios_with_budget(&two_scenario_value(), 8, 3);
        assert_eq!(second.len(), 1, "only the one remaining global slot may be used");
        assert_eq!(model.scenario_combinations_used(), 3);
        assert!(model.scenario_expansion_curtailed());

        let exhausted = model.collect_scenarios_with_budget(&two_scenario_value(), 8, 3);
        assert!(exhausted.is_empty());
        assert_eq!(model.scenario_combinations_used(), 3, "the global limit must never be exceeded");
    }

    #[test]
    fn concurrent_scenario_queries_cannot_exceed_global_budget() {
        let model = std::sync::Arc::new(SemanticModel::from_bytes(br#"{"Resources":{"R":{"Type":"T"}}}"#).unwrap());
        let handles: Vec<_> = (0..8)
            .map(|_| {
                let model = std::sync::Arc::clone(&model);
                std::thread::spawn(move || {
                    let value = ResolvedValue::Concrete { value: serde_json::json!("value").into() };
                    model.collect_scenarios_with_budget(&value, 8, 4).len()
                })
            })
            .collect();
        let produced: usize =
            handles.into_iter().map(|handle| handle.join().expect("scenario worker must not panic")).sum();

        assert_eq!(produced, 4, "exactly the global budget may be materialized across all workers");
        assert_eq!(model.scenario_combinations_used(), 4, "concurrent accounting must not overshoot the limit");
        assert!(model.scenario_expansion_curtailed(), "workers denied by exhaustion must mark curtailment");
    }

    #[test]
    fn scenario_expansion_curtailment_flag_starts_false() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"V":"literal"}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert!(
            !model.scenario_expansion_curtailed(),
            "a freshly built model must not have scenario expansion curtailed"
        );
    }

    #[test]
    fn scenario_expansion_curtailment_flag_set_by_test_helper() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"V":"literal"}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        model.set_scenario_expansion_curtailed_for_test();
        assert!(model.scenario_expansion_curtailed(), "the test helper must set the curtailment flag");
    }

    #[test]
    fn scenario_expansion_curtailment_flag_is_monotonic() {
        // Once set, the flag never resets — regardless of subsequent queries that
        // stay within budget.
        let input = r#"{
            "Resources": {"R": {"Type": "T", "Properties": {"V": {"Fn::If": ["C", "a", "b"]}}}}
        }"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        model.set_scenario_expansion_curtailed_for_test();
        // A normal query should not reset the flag.
        let _ = model.resolve_scenarios("R", "Properties.V");
        assert!(model.scenario_expansion_curtailed(), "the curtailment flag must be monotonic — once set it stays set");
    }

    #[test]
    fn estimated_bounds_are_absent_for_a_literal_string() {
        // Schema validation checks a literal against its constraint directly, so
        // the estimating rule is deliberately given nothing for one.
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"V":"hello"}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert_eq!(model.estimated_string_length_bounds("R", "Properties.V"), None);
    }

    #[test]
    fn estimated_bounds_span_every_allowed_value() {
        let input = r#"{"Parameters":{"Size":{"Type":"String","AllowedValues":["xs","large"]}},
            "Resources":{"R":{"Type":"T","Properties":{"V":{"Ref":"Size"}}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert_eq!(model.estimated_string_length_bounds("R", "Properties.V"), Some((2, 5)));
    }

    #[test]
    fn estimated_bounds_are_absent_when_a_possibility_is_unknown() {
        let input = r#"{"Parameters":{"Free":{"Type":"String"}},
            "Resources":{"R":{"Type":"T","Properties":{"V":{"Ref":"Free"}}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert_eq!(model.estimated_string_length_bounds("R", "Properties.V"), None);
    }

    #[test]
    fn estimate_string_length_missing_property() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"V":"hello"}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert!(model.estimated_string_length_bounds("R", "Properties.Missing").is_none());
    }

    #[test]
    fn resource_span_specific_path() {
        let input = "Resources:\n  R:\n    Type: T\n    Properties:\n      Name: hello\n";
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        let span = model.resource_span("R", "Properties/Name");
        // Should find a span (not UNKNOWN_SPAN) for the specific path
        assert!(span.start_line > 0 || span.end_line > 0, "expected non-zero span for Properties/Name");
    }

    #[test]
    fn resource_span_specific_path_dotted_form() {
        let input = "Resources:\n  R:\n    Type: T\n    Properties:\n      Name: hello\n";
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        // A dotted, resource-relative path must resolve to the SAME specific span
        // as its slash-form equivalent - not fall back to the resource span.
        let dotted = model.resource_span("R", "Properties.Name");
        let slashed = model.resource_span("R", "Properties/Name");
        let resource = model.resource_span("R", "");
        assert_eq!(dotted, slashed, "dotted and slash forms must resolve to the same span");
        assert_ne!(dotted, resource, "specific property span must differ from the resource span");
    }

    #[test]
    fn resource_span_falls_back_to_resource() {
        let input = "Resources:\n  R:\n    Type: T\n    Properties:\n      Name: hello\n";
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        let span = model.resource_span("R", "Properties/NonExistent/Deep");
        // Should fall back to the resource-level span
        assert!(span.start_line > 0 || span.end_line > 0, "expected non-zero span, got {:?}", span);
    }

    #[test]
    fn resource_span_nested_array_property_anchors_at_element() {
        // A property inside an array element must anchor at that element, not at the
        // array or the resource. The span index is keyed with the array index.
        let input =
            "Resources:\n  R:\n    Type: T\n    Properties:\n      Ingress:\n      - Port: 80\n      - Port: 443\n";
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        let first = model.resource_span("R", "Properties.Ingress.0.Port");
        let second = model.resource_span("R", "Properties.Ingress.1.Port");
        assert_eq!(first.start_line, 6, "Ingress[0].Port is on line 6, got {:?}", first);
        assert_eq!(second.start_line, 7, "Ingress[1].Port is on line 7, got {:?}", second);
    }

    #[test]
    fn resource_span_empty_id_resolves_section_absolute_path_precisely() {
        // A finding with no resource id (e.g. an Outputs-level diagnostic) carries a
        // section-absolute span-index path. It must resolve against that path directly,
        // NOT be prefixed with `Resources/` (which would mislocate onto the Resources
        // block). A path whose exact node is indexed resolves to that node.
        let input = concat!(
            "Resources:\n  R:\n    Type: T\n",                                  // lines 1-3
            "Outputs:\n  Combined:\n    Value: !Join [\"\", [\"a\", \"b\"]]\n", // lines 4-6
        );
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        // The `Value` node is indexed (line 6) - resolution lands on it exactly.
        let value = model.resource_span("", "Outputs/Combined/Value");
        assert_eq!(value.start_line, 6, "Outputs/Combined/Value is on line 6, got {:?}", value);
        // The output key itself resolves to its own line (line 5).
        let combined = model.resource_span("", "Outputs/Combined");
        assert_eq!(combined.start_line, 5, "Outputs/Combined is on line 5, got {:?}", combined);
    }

    #[test]
    fn resource_span_empty_id_fused_intrinsic_suffix_anchors_at_nearest_slash_ancestor() {
        // A synthetic intrinsic suffix (`.Fn::Join`) is joined to its parent by a DOT,
        // which is deliberately not treated as a path separator here: real span-index
        // keys contain literal dots inside a single segment (e.g. API Gateway's
        // `method.request.path.proxy`), so splitting on dots would shred those paths and
        // mis-anchor. The walk-up therefore trims the whole `Value.Fn::Join` segment on
        // the nearest `/`, landing on the enclosing output - still within Outputs, never
        // on the Resources block. Both engines resolve this identically, which is what
        // keeps them at parity.
        let input = concat!(
            "Resources:\n  R:\n    Type: T\n",                                  // lines 1-3
            "Outputs:\n  Combined:\n    Value: !Join [\"\", [\"a\", \"b\"]]\n", // lines 4-6
        );
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        let span = model.resource_span("", "Outputs/Combined/Value.Fn::Join");
        assert_eq!(
            span.start_line, 5,
            "fused-suffix path must anchor at the enclosing output (line 5), got {:?}",
            span
        );
    }

    #[test]
    fn resource_span_empty_id_does_not_leak_onto_resources_block() {
        // Regression guard: an unresolvable empty-id path must return UNKNOWN rather
        // than walking up a spuriously `Resources/`-prefixed key onto the Resources
        // section. This is the divergence that made the two engines disagree.
        let input = "Resources:\n  R:\n    Type: T\n";
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        let span = model.resource_span("", "Outputs/DoesNotExist/Value");
        assert_eq!(span, UNKNOWN_SPAN, "unresolvable empty-id path must be UNKNOWN, got {:?}", span);
    }

    #[test]
    fn resource_span_empty_path_returns_resource_span() {
        let input = "Resources:\n  R:\n    Type: T\n";
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        let span = model.resource_span("R", "");
        assert!(span.start_line > 0 || span.end_line > 0, "expected non-zero span, got {:?}", span);
    }

    #[test]
    fn resolve_nonexistent_resource_returns_none() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"V":"x"}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert!(model.resolve("NoSuchResource", "Properties.V").is_none(), "nonexistent resource should return None");
        assert!(model.resolve_deep("NoSuchResource", "Properties.V").is_none());
    }

    #[test]
    fn parse_config_with_parameter_overrides() {
        let input = r#"{"Parameters":{"Env":{"Type":"String","AllowedValues":["dev","prod"]}},"Resources":{"R":{"Type":"T","Properties":{"V":{"Ref":"Env"}}}}}"#;
        let config = ParseConfig {
            parameters: [("Env".to_string(), "staging".to_string())].into_iter().collect(),
            pseudo_parameters: PseudoParameterOverrides::default(),
        };
        let result = SemanticModel::parse(input.as_bytes(), config).unwrap();
        match result.model.resolve("R", "Properties.V") {
            Some(ResolvedValue::Concrete { value: v }) => {
                assert_eq!(v.as_str().unwrap(), "staging")
            }
            other => panic!("Expected Concrete(\"staging\"), got {:?}", other),
        }
    }

    #[test]
    fn parse_config_with_pseudo_parameter_overrides() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"V":{"Ref":"AWS::Region"}}}}}"#;
        let config = ParseConfig {
            parameters: HashMap::new(),
            pseudo_parameters: PseudoParameterOverrides { region: Some("eu-west-1".to_string()), ..Default::default() },
        };
        let result = SemanticModel::parse(input.as_bytes(), config).unwrap();
        match result.model.resolve("R", "Properties.V") {
            Some(ResolvedValue::Concrete { value: v }) => {
                assert_eq!(v.as_str().unwrap(), "eu-west-1")
            }
            other => panic!("Expected Concrete(\"eu-west-1\"), got {:?}", other),
        }
    }

    #[test]
    fn pseudo_parameter_overrides_defaults() {
        let overrides = PseudoParameterOverrides::default();
        assert_eq!(overrides.region(), DEFAULT_REGION);
        assert_eq!(overrides.get("AWS::AccountId").unwrap(), DEFAULT_ACCOUNT_ID);
        assert_eq!(overrides.get("AWS::Partition").unwrap(), DEFAULT_PARTITION);
        assert_eq!(overrides.get("AWS::StackName").unwrap(), DEFAULT_STACK_NAME);
        assert_eq!(overrides.get("AWS::URLSuffix").unwrap(), DEFAULT_URL_SUFFIX);
        assert!(overrides.get("AWS::StackId").unwrap().contains("arn:"));
        assert!(overrides.get("AWS::NotificationARNs").unwrap().contains("sns"));
        assert_eq!(overrides.get("Unknown"), None, "unknown pseudo-param should return None");
    }

    #[test]
    fn pseudo_parameter_overrides_china_region() {
        let overrides = PseudoParameterOverrides { region: Some("cn-north-1".to_string()), ..Default::default() };
        assert_eq!(overrides.get("AWS::Partition").unwrap(), "aws-cn");
        assert_eq!(overrides.get("AWS::URLSuffix").unwrap(), "amazonaws.com.cn");
    }

    /// `fixed_value` underpins the SAT-solver decision "treat this pseudo-param
    /// as a constant or a free variable". It must return `Some` only when the
    /// caller pinned the corresponding field - never for region-derived
    /// defaults - otherwise `Fn::Equals[Ref AWS::Partition, "aws"]` would be
    /// falsely deterministic and `find_unreachable_branches` would emit
    /// false-positive unreachable-branch diagnostics.
    #[test]
    fn fixed_value_returns_none_for_unset_pseudo_parameters() {
        let overrides = PseudoParameterOverrides::default();

        for name in [
            "AWS::AccountId",
            "AWS::NotificationARNs",
            "AWS::Partition",
            "AWS::Region",
            "AWS::StackId",
            "AWS::StackName",
            "AWS::URLSuffix",
        ] {
            assert_eq!(
                overrides.fixed_value(name),
                None,
                "{name} must be a free variable when the user has not explicitly pinned it"
            );
        }
    }

    #[test]
    fn parameter_name_at_names_the_parameter_behind_a_value() {
        let input = r#"{"Parameters":{"BucketName":{"Type":"String"}},
            "Resources":{"R":{"Type":"AWS::S3::Bucket","Properties":{"BucketName":{"Ref":"BucketName"}}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert_eq!(model.parameter_name_at("R", "Properties.BucketName"), Some("BucketName"));
        assert!(model.is_from_parameter("R", "Properties.BucketName"));
    }

    #[test]
    fn parameter_name_at_is_absent_for_a_value_no_parameter_produced() {
        // A cross-stack import is unknown until deployment without being a
        // parameter, so nothing may name one.
        let input = r#"{"Resources":{"R":{"Type":"AWS::S3::Bucket",
            "Properties":{"BucketName":{"Fn::ImportValue":"SharedName"}}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert_eq!(model.parameter_name_at("R", "Properties.BucketName"), None);
        assert!(!model.is_from_parameter("R", "Properties.BucketName"));
    }

    #[test]
    fn fixed_value_returns_user_supplied_overrides() {
        let overrides = PseudoParameterOverrides {
            account_id: Some("999999999999".to_string()),
            partition: Some("aws-cn".to_string()),
            region: Some("cn-north-1".to_string()),
            stack_name: Some("MyStack".to_string()),
            ..Default::default()
        };

        assert_eq!(overrides.fixed_value("AWS::AccountId"), Some("999999999999".to_string()));
        assert_eq!(overrides.fixed_value("AWS::Partition"), Some("aws-cn".to_string()));
        assert_eq!(overrides.fixed_value("AWS::Region"), Some("cn-north-1".to_string()));
        assert_eq!(overrides.fixed_value("AWS::StackName"), Some("MyStack".to_string()));
        assert_eq!(overrides.fixed_value("AWS::URLSuffix"), None, "URLSuffix not set; must remain a free variable");
        assert_eq!(overrides.fixed_value("Unknown"), None);
    }

    #[test]
    fn parameter_name_from_source_reads_only_a_parameter_entry() {
        assert_eq!(parameter_name_from_source("Parameters/InstanceType/Default"), Some("InstanceType"));
        assert_eq!(parameter_name_from_source("Parameters/InstanceType/AllowedValues"), Some("InstanceType"));
        assert_eq!(parameter_name_from_source("Parameters/InstanceType"), Some("InstanceType"));
        assert_eq!(parameter_name_from_source("Intrinsic/Ref"), None);
        assert_eq!(parameter_name_from_source("Parameters/"), None);
        assert_eq!(parameter_name_from_source("ParametersLookAlike/Name"), None);
    }

    /// Setting `region` alone must NOT cause `fixed_value("AWS::Partition")`
    /// to return a value. The convenience defaulting that `get` performs (so
    /// resource-property substitution sees a sensible partition string) is
    /// deliberately separate from the SAT-solver pinning, because constraining
    /// the region constrains where the stack deploys, not the partition the
    /// template was written against.
    #[test]
    fn fixed_value_does_not_propagate_region_to_partition() {
        let overrides = PseudoParameterOverrides { region: Some("cn-north-1".to_string()), ..Default::default() };

        assert_eq!(overrides.fixed_value("AWS::Region"), Some("cn-north-1".to_string()));
        assert_eq!(
            overrides.fixed_value("AWS::Partition"),
            None,
            "region override must not propagate to partition - partition stays free unless the caller pins it"
        );
    }

    #[test]
    fn model_rejects_oversized_template() {
        let huge = vec![b' '; 11 * 1024 * 1024];
        let result = SemanticModel::from_bytes(&huge);
        match result {
            Err(e) => assert!(e.message.contains("maximum size")),
            Ok(_) => panic!("Expected error for oversized template"),
        }
    }

    #[test]
    fn follow_ref_returns_none_for_non_reference() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"V":"hello"}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        assert_eq!(model.follow_ref("R", "Properties.V"), None, "non-reference value should return None");
    }

    #[test]
    fn resolve_deep_memoization_consistency() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"Config":{"A":"1","B":"2"}}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        let r1 = model.resolve_deep("R", "Properties.Config.A");
        let r2 = model.resolve_deep("R", "Properties.Config.A");
        match (&r1, &r2) {
            (Some(ResolvedValue::Concrete { value: a }), Some(ResolvedValue::Concrete { value: b })) => {
                assert_eq!(a, b);
            }
            _ => panic!("Expected matching Concrete values"),
        }
    }

    #[test]
    fn model_with_mappings_and_findinmap() {
        let input = r#"
Mappings:
  RegionMap:
    us-east-1:
      AMI: ami-12345
    us-west-2:
      AMI: ami-67890
Resources:
  R:
    Type: T
    Properties:
      ImageId: !FindInMap [RegionMap, us-east-1, AMI]
"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        match model.resolve("R", "Properties.ImageId") {
            Some(ResolvedValue::Concrete { value: v }) => {
                assert_eq!(v.as_str().unwrap(), "ami-12345")
            }
            other => panic!("Expected Concrete, got {:?}", other),
        }
    }

    #[test]
    fn model_invalid_ref_tracked() {
        let input = r#"{"Resources":{"R":{"Type":"T","Properties":{"V":{"Ref":"NonExistent"}}}}}"#;
        let model = SemanticModel::from_bytes(input.as_bytes()).unwrap();
        let r = model.resource("R").unwrap();
        assert!(!r.diagnostics.invalid_refs.is_empty());
        assert!(r.diagnostics.invalid_refs.iter().any(|s| s.value == "NonExistent"));
    }

    #[test]
    fn collapse_arn_paths_folds_list_siblings_to_lowest_index() {
        let input = vec![
            "Properties.KeyPolicy.Statement.2.Principal.AWS.1.Fn::Sub".to_string(),
            "Properties.KeyPolicy.Statement.2.Principal.AWS.0.Fn::Sub".to_string(),
        ];
        let out = collapse_list_sibling_arn_paths(input);
        assert_eq!(out, vec!["Properties.KeyPolicy.Statement.2.Principal.AWS.0.Fn::Sub".to_string()]);
    }

    #[test]
    fn collapse_arn_paths_keeps_distinct_parent_lists() {
        // Differing index is the Statement index, not the one before Fn::Sub, so
        // these are separate source locations and must both survive.
        let input = vec![
            "Properties.PolicyDocument.Statement.0.Resource.Fn::Sub".to_string(),
            "Properties.PolicyDocument.Statement.1.Resource.Fn::Sub".to_string(),
        ];
        let mut out = collapse_list_sibling_arn_paths(input.clone());
        out.sort();
        assert_eq!(out, input);
    }

    #[test]
    fn collapse_arn_paths_leaves_scalar_paths_untouched() {
        let input = vec![
            "Properties.KeyPolicy.Statement.0.Principal.AWS.Fn::Sub".to_string(),
            "Properties.KeyPolicy.Statement.1.Principal.AWS.Fn::Sub".to_string(),
        ];
        let mut out = collapse_list_sibling_arn_paths(input.clone());
        out.sort();
        assert_eq!(out, input);
    }

    /// Number of output value-type diagnostics (F6101) the parse phase produced.
    fn output_string_type_diagnostics(template: &str) -> usize {
        let model = SemanticModel::from_bytes(template.as_bytes()).unwrap();
        model.diagnostics.iter().filter(|d| d.rule_id == "F6101").count()
    }

    #[test]
    fn output_literal_list_value_flagged() {
        let template = "Resources:\n  R:\n    Type: T\nOutputs:\n  O:\n    Value:\n      - a\n      - b\n";
        assert_eq!(output_string_type_diagnostics(template), 1);
    }

    #[test]
    fn output_literal_object_value_flagged() {
        let template = "Resources:\n  R:\n    Type: T\nOutputs:\n  O:\n    Value:\n      Key: v\n";
        assert_eq!(output_string_type_diagnostics(template), 1);
    }

    #[test]
    fn output_list_returning_functions_flagged() {
        for value in ["!GetAZs \"\"", "!Split [\",\", \"a,b\"]", "!Cidr [\"10.0.0.0/16\", 1, 8]"] {
            let template = format!("Resources:\n  R:\n    Type: T\nOutputs:\n  O:\n    Value: {}\n", value);
            assert_eq!(output_string_type_diagnostics(&template), 1, "expected F6101 for output value {}", value);
        }
    }

    #[test]
    fn output_empty_container_value_not_flagged() {
        // An empty list/object has no members to stringify; CloudFormation does
        // not reject it, so neither does the parse-time check.
        for value in ["[]", "{}"] {
            let template = format!("Resources:\n  R:\n    Type: T\nOutputs:\n  O:\n    Value: {}\n", value);
            assert_eq!(output_string_type_diagnostics(&template), 0, "empty container {} must not be flagged", value);
        }
    }

    #[test]
    fn output_string_producing_values_not_flagged() {
        // Scalars coerce to strings; Ref/Fn::Sub/Fn::Join/Fn::Select/Fn::ImportValue
        // produce (or are treated as) strings. None are string-type violations.
        for value in ["hello", "42", "true", "!Ref R", "!Sub \"${R}\"", "!Join [\",\", [\"a\"]]", "!ImportValue X"] {
            let template = format!("Resources:\n  R:\n    Type: T\nOutputs:\n  O:\n    Value: {}\n", value);
            assert_eq!(output_string_type_diagnostics(&template), 0, "string-valued output {} must not fire", value);
        }
    }

    #[test]
    fn output_find_in_map_list_value_not_flagged() {
        // Fn::FindInMap can resolve to a list, but it is not flagged here (the
        // mapping's shape is validated elsewhere): only literal lists and
        // list-returning functions are string-type violations. This is exactly
        // the case a resolved-value check would get wrong, since the resolved
        // value is indistinguishable from a literal list.
        let template = "Mappings:\n  M:\n    k:\n      l:\n        - a\n        - b\nResources:\n  R:\n    Type: T\nOutputs:\n  O:\n    Value: !FindInMap [M, k, l]\n";
        assert_eq!(output_string_type_diagnostics(template), 0);
    }

    #[test]
    fn output_fn_if_flags_each_list_branch() {
        // Fn::If is transparent: each branch is checked. Both branches are lists,
        // so both are flagged.
        let template = "Conditions:\n  C:\n    Fn::Equals: [\"a\", \"a\"]\nResources:\n  R:\n    Type: T\nOutputs:\n  O:\n    Value:\n      Fn::If: [C, [\"a\"], [\"b\"]]\n";
        assert_eq!(output_string_type_diagnostics(template), 2);
    }

    #[test]
    fn output_fn_if_string_branches_not_flagged() {
        let template = "Conditions:\n  C:\n    Fn::Equals: [\"a\", \"a\"]\nResources:\n  R:\n    Type: T\nOutputs:\n  O:\n    Value:\n      Fn::If: [C, \"yes\", \"no\"]\n";
        assert_eq!(output_string_type_diagnostics(template), 0);
    }

    #[test]
    fn join_with_enum_list_exact_4096_not_curtailed() {
        use crate::consts::MAX_ENUM_EXPANSION;
        use crate::resolver::join_with_enum_list;
        // Build items that produce exactly MAX_ENUM_EXPANSION combinations:
        // 64 * 64 = 4096 (no truncation needed)
        let variants_a: Vec<ResolvedValue> = (0..64)
            .map(|i| ResolvedValue::Concrete { value: serde_json::Value::String(format!("a{i}")).into() })
            .collect();
        let variants_b: Vec<ResolvedValue> = (0..64)
            .map(|i| ResolvedValue::Concrete { value: serde_json::Value::String(format!("b{i}")).into() })
            .collect();
        let items = vec![ResolvedValue::Enum { variants: variants_a }, ResolvedValue::Enum { variants: variants_b }];
        let (result, curtailed) = join_with_enum_list(",", &items);
        assert!(!curtailed, "exactly {MAX_ENUM_EXPANSION} combinations must not signal curtailment");
        if let ResolvedValue::Enum { variants } = &result {
            assert_eq!(variants.len(), MAX_ENUM_EXPANSION);
        } else {
            panic!("expected Enum result");
        }
    }

    #[test]
    fn join_with_enum_list_one_over_curtailed() {
        use crate::consts::MAX_ENUM_EXPANSION;
        use crate::resolver::join_with_enum_list;
        // 65 * 64 = 4160 > 4096 — must curtail
        let variants_a: Vec<ResolvedValue> = (0..65)
            .map(|i| ResolvedValue::Concrete { value: serde_json::Value::String(format!("a{i}")).into() })
            .collect();
        let variants_b: Vec<ResolvedValue> = (0..64)
            .map(|i| ResolvedValue::Concrete { value: serde_json::Value::String(format!("b{i}")).into() })
            .collect();
        let items = vec![ResolvedValue::Enum { variants: variants_a }, ResolvedValue::Enum { variants: variants_b }];
        let (result, curtailed) = join_with_enum_list(",", &items);
        assert!(curtailed, "exceeding {MAX_ENUM_EXPANSION} combinations must signal curtailment");
        if let ResolvedValue::Enum { variants } = &result {
            assert_eq!(variants.len(), MAX_ENUM_EXPANSION);
        } else {
            panic!("expected Enum result");
        }
    }

    #[test]
    fn primary_id_same_parameter_across_ref_sub_join_collides() {
        let template = r#"
Parameters:
  Name: {Type: String}
Resources:
  A:
    Type: AWS::S3::Bucket
    Properties: {BucketName: !Ref Name}
  B:
    Type: AWS::S3::Bucket
    Properties: {BucketName: !Sub '${Name}'}
  C:
    Type: AWS::S3::Bucket
    Properties: {BucketName: !Sub ['${V}', {V: !Ref Name}]}
  D:
    Type: AWS::S3::Bucket
    Properties: {BucketName: !Join ['', [!Ref Name]]}
"#;
        let model = SemanticModel::from_bytes(template.as_bytes()).unwrap();
        let conflicts = model.primary_identifier_conflicts("AWS::S3::Bucket", &["BucketName".to_string()]);
        assert_eq!(conflicts.len(), 1, "all four forms must collide into one group");
        let (_, resources) = conflicts.into_iter().next().unwrap();
        assert_eq!(resources.len(), 4, "all four resources must be in the group");
    }

    #[test]
    fn primary_id_different_parameters_with_equal_defaults_stay_distinct() {
        let template = r#"
Parameters:
  A: {Type: String, Default: shared}
  B: {Type: String, Default: shared}
Resources:
  One:
    Type: AWS::S3::Bucket
    Properties: {BucketName: !Ref A}
  Two:
    Type: AWS::S3::Bucket
    Properties: {BucketName: !Ref B}
"#;
        let model = SemanticModel::from_bytes(template.as_bytes()).unwrap();
        let conflicts = model.primary_identifier_conflicts("AWS::S3::Bucket", &["BucketName".to_string()]);
        assert!(conflicts.is_empty(), "different overridable parameters must not collide: {:?}", conflicts);
    }

    #[test]
    fn primary_id_comparison_identity_is_separate_from_display_value() {
        let template = r#"
Parameters:
  Name: {Type: String}
Resources:
  DynamicOne:
    Type: AWS::S3::Bucket
    Properties: {BucketName: !Ref Name}
  DynamicTwo:
    Type: AWS::S3::Bucket
    Properties: {BucketName: !Ref Name}
  LiteralOne:
    Type: AWS::S3::Bucket
    Properties: {BucketName: 'Ref("Name")'}
  LiteralTwo:
    Type: AWS::S3::Bucket
    Properties: {BucketName: 'Ref("Name")'}
"#;
        let model = SemanticModel::from_bytes(template.as_bytes()).unwrap();
        let conflicts = model.primary_identifier_conflicts("AWS::S3::Bucket", &["BucketName".to_string()]);
        assert_eq!(conflicts.len(), 2, "equal display text must not merge distinct identities");
        assert!(
            conflicts
                .iter()
                .all(|(tuple, resources)| { tuple == &["Ref(\"Name\")".to_string()] && resources.len() == 2 })
        );
    }

    #[test]
    fn primary_id_mutually_exclusive_conditions_do_not_collide() {
        let template = r#"
Parameters:
  Env: {Type: String}
  Name: {Type: String}
Conditions:
  IsProd: !Equals [!Ref Env, prod]
  IsNotProd: !Not [!Condition IsProd]
Resources:
  ProdBucket:
    Type: AWS::S3::Bucket
    Condition: IsProd
    Properties: {BucketName: !Ref Name}
  DevBucket:
    Type: AWS::S3::Bucket
    Condition: IsNotProd
    Properties: {BucketName: !Ref Name}
"#;
        let model = SemanticModel::from_bytes(template.as_bytes()).unwrap();
        let conflicts = model.primary_identifier_conflicts("AWS::S3::Bucket", &["BucketName".to_string()]);
        assert!(conflicts.is_empty(), "mutually exclusive conditions must not collide");
    }

    #[test]
    fn primary_id_no_internal_marker_leakage_in_tuple_values() {
        let template = r#"
Parameters:
  Name: {Type: String}
Resources:
  A:
    Type: AWS::S3::Bucket
    Properties: {BucketName: !Ref Name}
  B:
    Type: AWS::S3::Bucket
    Properties: {BucketName: !Ref Name}
"#;
        let model = SemanticModel::from_bytes(template.as_bytes()).unwrap();
        let conflicts = model.primary_identifier_conflicts("AWS::S3::Bucket", &["BucketName".to_string()]);
        assert_eq!(conflicts.len(), 1);
        let (tuple, _) = conflicts.into_iter().next().unwrap();
        let value = &tuple[0];
        assert!(!value.contains("$dyn:"), "tuple value must not contain $dyn: marker");
        assert!(!value.contains("expr:"), "tuple value must not contain expr: prefix");
    }

    /// The pre-grouping all-pairs traversal, kept verbatim so tests can prove the
    /// grouped implementation is byte-for-byte equivalent - including how the shared
    /// satisfiability budget is spent.
    fn primary_identifier_conflicts_all_pairs_reference(
        model: &SemanticModel,
        resource_type: &str,
        identifier_properties: &[String],
    ) -> Vec<(Vec<String>, BTreeSet<String>)> {
        let mut per_resource = Vec::new();
        for resource_id in model.resources_of_type(resource_type) {
            let scenarios = model.primary_identifier_scenarios(resource_id, identifier_properties);
            if !scenarios.is_empty() {
                per_resource.push((resource_id, scenarios));
            }
        }
        let mut conflicts: BTreeMap<Vec<String>, (Vec<String>, BTreeSet<String>)> = BTreeMap::new();
        for left_index in 0..per_resource.len() {
            for right_index in (left_index + 1)..per_resource.len() {
                let (left_resource, left_scenarios) = &per_resource[left_index];
                let (right_resource, right_scenarios) = &per_resource[right_index];
                for left in left_scenarios {
                    for right in right_scenarios {
                        if left.identity_tuple != right.identity_tuple {
                            continue;
                        }
                        let mut assumptions = left.assumptions.clone();
                        assumptions.extend(right.assumptions.iter().cloned());
                        if assumptions.is_empty() || model.conditions.is_satisfiable(&assumptions) {
                            let (_, resources) = conflicts
                                .entry(left.identity_tuple.clone())
                                .or_insert_with(|| (left.display_tuple.clone(), BTreeSet::new()));
                            resources.insert((*left_resource).clone());
                            resources.insert((*right_resource).clone());
                        }
                    }
                }
            }
        }
        conflicts.into_values().collect()
    }

    #[test]
    fn primary_id_unconditional_resource_collides_with_every_conditional_duplicate() {
        let template = r#"
Parameters:
  Env: {Type: String}
Conditions:
  IsAlpha: !Equals [!Ref Env, alpha]
  IsBeta: !Equals [!Ref Env, beta]
Resources:
  Always:
    Type: AWS::S3::Bucket
    Properties: {BucketName: shared}
  WhenAlpha:
    Type: AWS::S3::Bucket
    Condition: IsAlpha
    Properties: {BucketName: shared}
  WhenBeta:
    Type: AWS::S3::Bucket
    Condition: IsBeta
    Properties: {BucketName: shared}
"#;
        let model = SemanticModel::from_bytes(template.as_bytes()).unwrap();
        let conflicts = model.primary_identifier_conflicts("AWS::S3::Bucket", &["BucketName".to_string()]);
        assert_eq!(conflicts.len(), 1, "the single shared name is one collision group: {conflicts:?}");
        let (display, resources) = conflicts.into_iter().next().unwrap();
        assert_eq!(display, vec!["shared".to_string()]);
        assert_eq!(
            resources,
            BTreeSet::from(["Always".to_string(), "WhenAlpha".to_string(), "WhenBeta".to_string()]),
            "the unconditional resource collides with each conditional duplicate, joining all three even though the two conditions are themselves mutually exclusive"
        );
    }

    #[test]
    fn primary_id_mutually_exclusive_conditions_across_identities_never_collide() {
        let template = r#"
Parameters:
  Env: {Type: String}
  Left: {Type: String}
  Right: {Type: String}
Conditions:
  IsProd: !Equals [!Ref Env, prod]
  IsNotProd: !Not [!Condition IsProd]
Resources:
  LeftProd:
    Type: AWS::S3::Bucket
    Condition: IsProd
    Properties: {BucketName: !Ref Left}
  LeftDev:
    Type: AWS::S3::Bucket
    Condition: IsNotProd
    Properties: {BucketName: !Ref Left}
  RightProd:
    Type: AWS::S3::Bucket
    Condition: IsProd
    Properties: {BucketName: !Ref Right}
  RightDev:
    Type: AWS::S3::Bucket
    Condition: IsNotProd
    Properties: {BucketName: !Ref Right}
"#;
        let model = SemanticModel::from_bytes(template.as_bytes()).unwrap();
        let conflicts = model.primary_identifier_conflicts("AWS::S3::Bucket", &["BucketName".to_string()]);
        assert!(
            conflicts.is_empty(),
            "each identity's two resources are guarded by mutually exclusive conditions, so neither identity group yields a collision: {conflicts:?}"
        );
    }

    #[test]
    fn primary_id_duplicate_scenarios_of_one_resource_do_not_self_collide() {
        let template = r#"
Parameters:
  Env: {Type: String}
Conditions:
  IsProd: !Equals [!Ref Env, prod]
Resources:
  Lonely:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !If [IsProd, shared, shared]
"#;
        let model = SemanticModel::from_bytes(template.as_bytes()).unwrap();
        let scenarios = model.primary_identifier_scenarios("Lonely", &["BucketName".to_string()]);
        assert!(
            scenarios.len() >= 2,
            "the Fn::If must expand into multiple scenarios so self-comparison is actually exercised: {}",
            scenarios.len()
        );
        let identities: BTreeSet<Vec<String>> =
            scenarios.iter().map(|scenario| scenario.identity_tuple.clone()).collect();
        assert_eq!(identities.len(), 1, "the branches must share one comparison identity so they land in one group");

        let conflicts = model.primary_identifier_conflicts("AWS::S3::Bucket", &["BucketName".to_string()]);
        assert!(
            conflicts.is_empty(),
            "a single resource whose branches produce the same identifier must not be reported as colliding with itself: {conflicts:?}"
        );
    }

    #[test]
    fn primary_id_overlapping_conditions_collide() {
        let template = r#"
Parameters:
  Env: {Type: String}
  Name: {Type: String}
Conditions:
  NotAlpha: !Not [!Equals [!Ref Env, alpha]]
  NotBeta: !Not [!Equals [!Ref Env, beta]]
Resources:
  One:
    Type: AWS::S3::Bucket
    Condition: NotAlpha
    Properties: {BucketName: !Ref Name}
  Two:
    Type: AWS::S3::Bucket
    Condition: NotBeta
    Properties: {BucketName: !Ref Name}
"#;
        let model = SemanticModel::from_bytes(template.as_bytes()).unwrap();
        let conflicts = model.primary_identifier_conflicts("AWS::S3::Bucket", &["BucketName".to_string()]);
        assert_eq!(
            conflicts.len(),
            1,
            "the two resources share one identity and their conditions can both hold: {conflicts:?}"
        );
        let (_, resources) = conflicts.into_iter().next().unwrap();
        assert_eq!(resources, BTreeSet::from(["One".to_string(), "Two".to_string()]));
    }

    #[test]
    fn primary_id_display_tuple_comes_from_first_compatible_pair_not_first_candidate() {
        let template = r#"
Parameters:
  Env: {Type: String}
Conditions:
  IsProd: !Equals [!Ref Env, prod]
  IsDev: !Not [!Condition IsProd]
Resources:
  ProdKeyed:
    Type: AWS::S3::Bucket
    Condition: IsProd
    Properties:
      BucketName: {alpha: one, beta: two}
  DevKeyedFirst:
    Type: AWS::S3::Bucket
    Condition: IsDev
    Properties:
      BucketName: {beta: two, alpha: one}
  DevKeyedSecond:
    Type: AWS::S3::Bucket
    Condition: IsDev
    Properties:
      BucketName: {beta: two, alpha: one}
"#;
        let model = SemanticModel::from_bytes(template.as_bytes()).unwrap();
        let conflicts = model.primary_identifier_conflicts("AWS::S3::Bucket", &["BucketName".to_string()]);
        assert_eq!(conflicts.len(), 1, "all three author the same key-sorted identity: {conflicts:?}");
        let (display, resources) = conflicts.into_iter().next().unwrap();
        assert_eq!(
            resources,
            BTreeSet::from(["DevKeyedFirst".to_string(), "DevKeyedSecond".to_string()]),
            "the earlier-declared prod resource is mutually exclusive with both dev resources, so it never forms a compatible pair and is absent from the group"
        );
        assert!(
            display[0].starts_with("{\"beta\""),
            "display must come from the left scenario of the first compatible pair (a dev resource, authored beta-first), not the earlier incompatible prod candidate authored alpha-first: {display:?}"
        );
    }

    #[test]
    fn primary_id_grouped_traversal_matches_all_pairs_reference_and_sat_budget() {
        let template = r#"
Parameters:
  Env: {Type: String}
  Shared: {Type: String}
Conditions:
  IsProd: !Equals [!Ref Env, prod]
  IsStaging: !Equals [!Ref Env, staging]
  IsProdAlias: !Equals [!Ref Env, prod]
Resources:
  AlphaAlways:
    Type: AWS::S3::Bucket
    Properties: {BucketName: alpha}
  AlphaProd:
    Type: AWS::S3::Bucket
    Condition: IsProd
    Properties: {BucketName: alpha}
  AlphaStaging:
    Type: AWS::S3::Bucket
    Condition: IsStaging
    Properties: {BucketName: alpha}
  BetaProd:
    Type: AWS::S3::Bucket
    Condition: IsProd
    Properties: {BucketName: beta}
  BetaProdAlias:
    Type: AWS::S3::Bucket
    Condition: IsProdAlias
    Properties: {BucketName: beta}
  SharedOne:
    Type: AWS::S3::Bucket
    Properties: {BucketName: !Ref Shared}
  SharedTwo:
    Type: AWS::S3::Bucket
    Properties: {BucketName: !Ref Shared}
  Unique:
    Type: AWS::S3::Bucket
    Properties: {BucketName: unique}
"#;
        let identifier_properties = ["BucketName".to_string()];

        let reference_model = SemanticModel::from_bytes(template.as_bytes()).unwrap();
        let reference = primary_identifier_conflicts_all_pairs_reference(
            &reference_model,
            "AWS::S3::Bucket",
            &identifier_properties,
        );
        let reference_budget = reference_model.conditions.sat_iterations_used();

        let model = SemanticModel::from_bytes(template.as_bytes()).unwrap();
        let grouped = model.primary_identifier_conflicts("AWS::S3::Bucket", &identifier_properties);
        let grouped_budget = model.conditions.sat_iterations_used();

        assert_eq!(
            grouped, reference,
            "grouped traversal must produce the identical conflicts as the all-pairs reference"
        );
        assert!(
            reference_budget > 0,
            "the fixture must drive the satisfiability solver so the budget comparison is meaningful"
        );
        assert_eq!(
            grouped_budget, reference_budget,
            "grouped traversal must issue satisfiability queries in the same order and count, spending the shared budget identically across multiple identities"
        );
    }

    #[test]
    fn output_invalid_ref_has_canonical_path() {
        let template = "Resources:\n  R:\n    Type: T\nOutputs:\n  Bad:\n    Value: !Ref Missing\n";
        let result = SemanticModel::parse(template.as_bytes(), ParseConfig::default()).unwrap();
        let ref_defect = result
            .model
            .diagnostics
            .iter()
            .find(|d| d.rule_id == "F6101" && d.message.contains("Missing"))
            .expect("F6101 for missing ref");
        assert_eq!(
            ref_defect.property_path.as_deref(),
            Some("Outputs.Bad.Value.Ref"),
            "parse-time output ref diagnostic must have dot-separated path"
        );
        assert_ne!(ref_defect.span, crate::UNKNOWN_SPAN, "span must be resolved");
    }
}