cloudformation-validate-cel-engine 1.12.0

use super::EvalContext;
use super::patterns::AMI_ID_RE;
use data_source::rule_data::{PathSegment, ResourcePropertyPath};
use diagnostics::Diagnostic;
use diagnostics::RelatedResource;
use diagnostics::ResourceRef;
use std::collections::{BTreeSet, HashMap, HashSet};
use std::net::{Ipv4Addr, Ipv6Addr};
use std::sync::{Arc, LazyLock};
use template_model::SemanticModel;
use template_model::coercion::{coerce_port_to_string, coerce_to_integer, coerce_to_string, scalar_eq};
use template_model::consts::{
    EDGE_KIND_GET_ATT, EDGE_KIND_REF, EDGE_KIND_SELECT, FIELD_ATTR, FIELD_KIND, FIELD_MAPPINGS, FIELD_OUTGOING_REFS,
    FIELD_PROPERTIES, FIELD_RESOURCE_TYPE, FIELD_RESOURCES, FIELD_SOURCE_PATH, FIELD_TARGET, FN_IF, FN_REF,
    KEY_CREATION_POLICY, KEY_PROPERTIES, KEY_UPDATE_POLICY, PARAM_TYPE_STRING, TRANSFORM_SERVERLESS,
};
use template_model::dynamodb::analyze_dynamodb_table_scenarios;
use template_model::fargate::{CPU_UNIT_LABELS, VCPU_SIZES, cpu_is_offered};
use template_model::iam_policy::{inline_identity_policy_document_paths, validate_identity_policy_scenarios};
use template_model::message::{primary_identifier_conflict_message, render_str_list, render_value};
use template_model::resolver::{RefKind, ResolvedValue};
use template_model::route_table::duplicate_subnet_associations;
use template_model::vpc_cidr::subnets_outside_vpc;
use template_model::{
    CAA_RECORD_PATTERN, IAM_ROLE_ARN_RULE_PATTERN, MARKER_DYNAMIC, MARKER_PARAM_TYPE, MX_RECORD_PATTERN, SourceSpan,
};
use template_model::{hardcoded_az, region_enums, schedule_expression_errors};
use validation_engine::{make_resource_diagnostic, make_resource_diagnostic_at_source};

static DOMAIN_RE: LazyLock<regex::Regex> = LazyLock::new(|| {
    regex::Regex::new(r"^(?:[a-z0-9\*](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9][a-z0-9-]{0,61}[a-z0-9]$")
        .expect("Invalid DOMAIN_RE pattern")
});

static ARN_RE: LazyLock<regex::Regex> =
    LazyLock::new(|| regex::Regex::new(IAM_ROLE_ARN_RULE_PATTERN).expect("Invalid ARN_RE pattern"));

static W1030_VPC_RE: LazyLock<regex::Regex> = LazyLock::new(|| {
    regex::Regex::new(r"^vpc-(([0-9A-Fa-f]{8})|([0-9A-Fa-f]{17}))$").expect("Invalid W1030_VPC_RE pattern")
});

static TXT_RECORD_RE: LazyLock<regex::Regex> =
    LazyLock::new(|| regex::Regex::new(r#"^("[^"]{1,255}" *)*"[^"]{1,255}"$"#).expect("Invalid TXT_RECORD_RE pattern"));

static CAA_RECORD_RE: LazyLock<regex::Regex> =
    LazyLock::new(|| regex::Regex::new(CAA_RECORD_PATTERN).expect("Invalid CAA_RECORD_RE pattern"));

static MX_RECORD_RE: LazyLock<regex::Regex> =
    LazyLock::new(|| regex::Regex::new(MX_RECORD_PATTERN).expect("Invalid MX_RECORD_RE pattern"));

/// Matches an ARN that already contains a 12-digit account-id segment
/// (`:123456789012:`). A SourceArn with such a segment pins the calling account,
/// so no separate SourceAccount is needed (W3663).
static ARN_HAS_ACCOUNT_RE: LazyLock<regex::Regex> =
    LazyLock::new(|| regex::Regex::new(r":\d{12}:").expect("Invalid ARN_HAS_ACCOUNT_RE pattern"));

/// Converts a `ResourcePropertyPath`'s segments (after the `Properties` prefix)
/// into a dot-separated property path string suitable for `resolve_concrete` and
/// diagnostic anchoring. Wildcard segments are excluded from the prefix because
/// the caller iterates over them explicitly.
///
/// Example: segments `[Properties, BlockDeviceMappings, *, Ebs]` yields
/// `"Properties.BlockDeviceMappings"` — the caller iterates the array and
/// appends `.{idx}.Ebs` per element.
fn property_path_prefix(path: &ResourcePropertyPath) -> String {
    path.segments
        .iter()
        .take_while(|s| !matches!(s, PathSegment::Wildcard))
        .filter_map(|s| match s {
            PathSegment::Literal(l) => Some(l.as_str()),
            _ => None,
        })
        .collect::<Vec<_>>()
        .join(".")
}

/// Converts a full `ResourcePropertyPath` to a dot-separated property path
/// string, using `*` for wildcard segments. Suitable for diagnostic path
/// rendering when the complete path is needed.
fn full_property_path(path: &ResourcePropertyPath) -> String {
    path.segments
        .iter()
        .map(|s| match s {
            PathSegment::Literal(l) => l.as_str(),
            PathSegment::Wildcard => "*",
        })
        .collect::<Vec<_>>()
        .join(".")
}

/// Counts the number of wildcard segments in a path.
fn wildcard_count(path: &ResourcePropertyPath) -> usize {
    path.segments.iter().filter(|s| matches!(s, PathSegment::Wildcard)).count()
}

/// Navigates a dot-separated property path through a JSON value, returning the
/// leaf value if every segment resolves. Returns `None` if any segment is missing.
fn navigate_json_path<'a>(value: &'a serde_json::Value, dot_path: &str) -> Option<&'a serde_json::Value> {
    let mut current = value;
    for seg in dot_path.split('.') {
        if seg.is_empty() {
            continue;
        }
        current = current.get(seg)?;
    }
    Some(current)
}

/// Renders a resolved value to JSON while preserving `Fn::If` structure as
/// `{"Fn::If": [condition, then, else]}`, so callers can enumerate both branches
/// of a conditional. `resolved_to_json_best_effort` collapses a conditional to
/// its true branch, which hides the false branch from branch-sensitive checks
/// (e.g. CodePipeline artifact counts). References/dynamics render as null.
fn resolved_to_json_preserving_conditionals(rv: &ResolvedValue) -> serde_json::Value {
    match rv {
        ResolvedValue::Concrete { value: v } => v.0.clone(),
        ResolvedValue::List { items } => {
            serde_json::Value::Array(items.iter().map(resolved_to_json_preserving_conditionals).collect())
        }
        ResolvedValue::Map { entries } => {
            let mut map = serde_json::Map::new();
            for e in entries {
                map.insert(e.key.clone(), resolved_to_json_preserving_conditionals(&e.value));
            }
            serde_json::Value::Object(map)
        }
        ResolvedValue::Conditional { condition, if_true, if_false } => serde_json::json!({
            "Fn::If": [
                condition.clone(),
                resolved_to_json_preserving_conditionals(if_true),
                resolved_to_json_preserving_conditionals(if_false),
            ]
        }),
        ResolvedValue::Enum { variants } => {
            for v in variants {
                if let ResolvedValue::Concrete { value: c } = v {
                    return c.0.clone();
                }
            }
            serde_json::Value::Null
        }
        ResolvedValue::Reference { target, .. } => serde_json::json!({ (FN_REF): target }),
        ResolvedValue::Dynamic { .. } | ResolvedValue::TypedDynamic { .. } => serde_json::Value::Null,
    }
}

/// Enumerates the possible element counts of a CodePipeline artifact list.
/// The list may be a plain array (one count) or an `Fn::If` whose two branches
/// each contribute a count; nested `Fn::If`s recurse. A missing/`AWS::NoValue`
/// branch counts as 0. The result is sorted and deduped so identical branch
/// counts are checked once. Returns `[0]` when the value is absent.
fn artifact_count_scenarios(value: Option<&serde_json::Value>) -> Vec<usize> {
    let mut counts = BTreeSet::new();
    fn walk(value: Option<&serde_json::Value>, out: &mut BTreeSet<usize>) {
        match value {
            None => {
                out.insert(0);
            }
            Some(serde_json::Value::Array(a)) => {
                out.insert(a.len());
            }
            Some(serde_json::Value::Object(o)) if o.len() == 1 && o.contains_key("Fn::If") => {
                // Fn::If: [condition, then, else] - enumerate both value branches.
                if let Some(branches) = o.get("Fn::If").and_then(|v| v.as_array()) {
                    walk(branches.get(1), out);
                    walk(branches.get(2), out);
                }
            }
            // AWS::NoValue (a Ref object) or any non-array scalar contributes no
            // artifacts.
            Some(_) => {
                out.insert(0);
            }
        }
    }
    walk(value, &mut counts);
    counts.into_iter().collect()
}

/// Collects `(block_device_mapping_path, ebs_object)` pairs for a resource,
/// following the `<prefix>` array and, for SpotFleet, the nested
/// `SpotFleetRequestConfigData.LaunchSpecifications[*].BlockDeviceMappings`. The
/// returned path is relative to the resource so a diagnostic can anchor at the
/// specific mapping's `Ebs.Iops`.
fn ebs_block_device_mappings(m: &SemanticModel, rid: &str, prefix: &str) -> Vec<(String, serde_json::Value)> {
    let mut out = Vec::new();
    if let Some(serde_json::Value::Array(mappings)) = resolve_concrete(m, rid, prefix) {
        for (idx, mapping) in mappings.iter().enumerate() {
            if let Some(ebs) = mapping.get("Ebs") {
                out.push((format!("{}.{}", prefix, idx), ebs.clone()));
            }
        }
    }
    out
}

/// Whether a Lambda runtime supports SnapStart, based on rule-tables data.
/// A runtime is supported if it starts with one of the configured prefixes
/// AND is not in the explicitly unsupported prefix/runtime lists.
fn snapstart_runtime_supported(
    runtime: &str,
    runtime_prefixes: &[String],
    unsupported_runtime_prefixes: &[String],
    unsupported_runtimes: &[String],
) -> bool {
    if !runtime_prefixes.iter().any(|prefix| runtime.starts_with(prefix.as_str())) {
        return false;
    }
    if unsupported_runtime_prefixes.iter().any(|prefix| runtime.starts_with(prefix.as_str())) {
        return false;
    }
    !unsupported_runtimes.iter().any(|r| r == runtime)
}

fn resolve_all_json(m: &SemanticModel, rid: &str, path: &str) -> Vec<serde_json::Value> {
    let rv = m.resolve_deep(rid, path).or_else(|| m.resolve(rid, path).cloned());
    let Some(rv) = rv else { return vec![] };
    collect_concrete_values(&rv)
}

fn collect_concrete_values(rv: &ResolvedValue) -> Vec<serde_json::Value> {
    match rv {
        ResolvedValue::Concrete { value: v } => vec![v.0.clone()],
        ResolvedValue::List { items } => {
            vec![serde_json::Value::Array(items.iter().map(resolved_to_json_best_effort).collect())]
        }
        ResolvedValue::Conditional { if_true, if_false, .. } => {
            let mut vals = collect_concrete_values(if_true);
            vals.extend(collect_concrete_values(if_false));
            vals
        }
        _ => {
            let v = resolved_to_json_best_effort(rv);
            if v.is_null() { vec![] } else { vec![v] }
        }
    }
}

/// A concrete HealthCheckPort value as a comparable string, or `None` when the
/// value is not a scalar (an opaque Ref/dynamic reference renders as an object
/// or null and must not be treated as a fixed port). Numbers stringify so an
/// unquoted `8080` and a quoted `"8080"` compare equal.
fn health_check_port_scalar(value: &serde_json::Value) -> Option<String> {
    match value {
        serde_json::Value::String(s) => Some(s.clone()),
        serde_json::Value::Number(n) => Some(n.to_string()),
        _ => None,
    }
}

fn resolve_concrete(m: &SemanticModel, rid: &str, path: &str) -> Option<serde_json::Value> {
    if let Some(resolved) = m.resolve_deep(rid, path).or_else(|| m.resolve(rid, path).cloned()) {
        return match resolved {
            ResolvedValue::Concrete { value: v } => Some(v.into_inner()),
            _ => None,
        };
    }
    // `Properties` wrapped in `Fn::If` stores values only under the synthetic
    // branch path - fall back to scenario resolution so rules that look up
    // properties by name still see per-branch values.
    let scenarios = m.resolve_scenarios_json(rid, path);
    scenarios.into_iter().next().map(|(v, _)| v)
}

/// Number of items in a list-valued property. Known even when the items are
/// references to other resources, which concrete resolution cannot produce;
/// unknown (`None`) when the whole list is a reference or conditional.
fn resolved_item_count(m: &SemanticModel, rid: &str, path: &str) -> Option<usize> {
    resolved_list_items(m, rid, path).map(|items| items.len())
}

/// The items of a list-valued property, rendered best-effort so that items
/// containing references are still visible to rules that only read their
/// literal members. `None` when the whole list is a reference or conditional.
fn resolved_list_items(m: &SemanticModel, rid: &str, path: &str) -> Option<Vec<serde_json::Value>> {
    match m.resolve_deep(rid, path).or_else(|| m.resolve(rid, path).cloned())? {
        ResolvedValue::Concrete { value } => value.as_array().cloned(),
        ResolvedValue::List { items } => Some(items.iter().map(resolved_to_json_best_effort).collect()),
        _ => None,
    }
}

const ALB_MINIMUM_SUBNETS: usize = 2;

const ALB_SUBNET_MINIMUM_MESSAGES: [(&str, &str); 2] = [
    ("Subnets", "Application load balancer requires at least 2 subnets"),
    ("SubnetMappings", "Application load balancer requires at least 2 subnet mappings"),
];

/// A load balancer without a Type is an application load balancer. A Type that
/// is present but does not resolve to a string (a parameter reference) is left
/// alone rather than assumed.
fn alb_is_application_type(m: &SemanticModel, rid: &str) -> bool {
    if !m.has_property(rid, "Type") {
        return true;
    }
    resolve_concrete(m, rid, "Properties.Type").as_ref().and_then(|v| v.as_str()) == Some("application")
}

/// The keys of a Stage method setting that configure caching, logging, metrics
/// or throttling. Once any of them is present, API Gateway resolves the
/// setting's ResourcePath as an absolute path, so it must start with '/'.
const STAGE_METHOD_SETTING_KEYS: [&str; 8] = [
    "CacheDataEncrypted",
    "CacheTtlInSeconds",
    "CachingEnabled",
    "DataTraceEnabled",
    "LoggingLevel",
    "MetricsEnabled",
    "ThrottlingBurstLimit",
    "ThrottlingRateLimit",
];

fn stage_method_setting_configures_something(setting: &serde_json::Value) -> bool {
    STAGE_METHOD_SETTING_KEYS.iter().any(|key| setting.get(key).is_some_and(|value| !value.is_null()))
}

/// Enhanced Monitoring on a DB cluster is configured by two properties that
/// only work together: a MonitoringRoleArn is used only when MonitoringInterval
/// is greater than 0, and a non-zero interval needs a role to publish with.
const DBCLUSTER_MONITORING_MESSAGE: &str =
    "MonitoringRoleArn and a MonitoringInterval greater than 0 must be specified together";

/// The property path an inconsistent monitoring configuration is reported at,
/// or `None` when the configuration is consistent or not decidable.
fn dbcluster_monitoring_mismatch_path(m: &SemanticModel, rid: &str) -> Option<&'static str> {
    let has_role = m.has_property(rid, "MonitoringRoleArn");
    let has_interval = m.has_property(rid, "MonitoringInterval");
    let interval = resolve_concrete(m, rid, "Properties.MonitoringInterval").as_ref().and_then(coerce_to_integer);
    match (has_role, has_interval, interval) {
        (true, false, _) => Some(KEY_PROPERTIES),
        (true, true, Some(interval)) if interval <= 0 => Some("Properties.MonitoringInterval"),
        (false, true, Some(interval)) if interval > 0 => Some(KEY_PROPERTIES),
        _ => None,
    }
}

/// Runs the shared identity-policy structural validator against a resolved
/// document and converts its findings into engine diagnostics. The `substituted`
/// set is derived from the reference graph: any outgoing edge whose source_path
/// is a descendant of the document path identifies a value produced by resolving
/// an intrinsic rather than being authored literally.
fn push_identity_policy_findings(
    out: &mut Vec<Diagnostic>,
    model: &SemanticModel,
    resource_id: &str,
    document_path: &str,
) {
    for finding in validate_identity_policy_scenarios(model, resource_id, document_path) {
        let effective_path = if finding.path.is_empty() {
            document_path.to_string()
        } else {
            format!("{}.{}", document_path, finding.path)
        };
        // Use the authored source_path (branch-qualified) for span resolution
        // when available, falling back to the effective path.
        let source_path = if finding.source_path.is_empty() { effective_path.clone() } else { finding.source_path };
        out.push(make_resource_diagnostic_at_source(
            "E3510",
            &finding.message,
            model,
            resource_id,
            &effective_path,
            &source_path,
            None,
        ));
    }
}

pub(super) fn scenario_is_reachable(m: &SemanticModel, resource_id: &str, conditions: &HashMap<String, bool>) -> bool {
    let mut assumptions: Vec<(String, bool)> = conditions.iter().map(|(name, value)| (name.clone(), *value)).collect();
    if let Some(resource_condition) = m.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() || m.conditions.is_satisfiable(&assumptions)
}

pub(super) fn merge_reachable_scenario_conditions(
    model: &SemanticModel,
    resource_id: &str,
    left: &HashMap<String, bool>,
    right: &HashMap<String, bool>,
) -> Option<HashMap<String, bool>> {
    let mut combined = left.clone();
    for (condition, value) in right {
        if combined.get(condition).is_some_and(|existing| existing != value) {
            return None;
        }
        combined.insert(condition.clone(), *value);
    }
    scenario_is_reachable(model, resource_id, &combined).then_some(combined)
}

fn scenario_conditions_overlap(
    model: &SemanticModel,
    resource_id: &str,
    left: &HashMap<String, bool>,
    right: &HashMap<String, bool>,
) -> bool {
    merge_reachable_scenario_conditions(model, resource_id, left, right).is_some()
}

fn scenario_overlaps_any(
    model: &SemanticModel,
    resource_id: &str,
    conditions: &HashMap<String, bool>,
    applicable_scenarios: &[HashMap<String, bool>],
) -> bool {
    applicable_scenarios
        .iter()
        .any(|applicable| scenario_conditions_overlap(model, resource_id, conditions, applicable))
}

pub(super) fn fargate_condition_scenarios(model: &SemanticModel, resource_id: &str) -> Vec<HashMap<String, bool>> {
    model
        .resolve_scenarios_json(resource_id, "Properties.RequiresCompatibilities")
        .into_iter()
        .filter_map(|(value, conditions)| {
            let is_fargate = value
                .as_array()
                .is_some_and(|compatibilities| compatibilities.iter().any(|item| item.as_str() == Some("FARGATE")));
            (is_fargate && scenario_is_reachable(model, resource_id, &conditions)).then_some(conditions)
        })
        .collect()
}

pub(super) fn scenario_has_effective_property(properties: &ResolvedValue, property_name: &str) -> bool {
    match properties {
        ResolvedValue::Concrete { value } => {
            value.get(property_name).is_some_and(|property_value| !property_value.is_null())
        }
        ResolvedValue::Map { entries } => entries.iter().any(|entry| {
            entry.key == property_name && !matches!(&entry.value, ResolvedValue::Concrete { value } if value.is_null())
        }),
        _ => false,
    }
}

fn resolved_to_route53_scenario_json(value: &ResolvedValue) -> serde_json::Value {
    match value {
        ResolvedValue::Concrete { value } => value.0.clone(),
        ResolvedValue::List { items } => {
            serde_json::Value::Array(items.iter().map(resolved_to_route53_scenario_json).collect())
        }
        ResolvedValue::Map { entries } => serde_json::Value::Object(
            entries.iter().map(|entry| (entry.key.clone(), resolved_to_route53_scenario_json(&entry.value))).collect(),
        ),
        ResolvedValue::Enum { variants } => variants
            .iter()
            .find_map(|variant| match variant {
                ResolvedValue::Concrete { value } => Some(value.0.clone()),
                _ => None,
            })
            .unwrap_or(serde_json::Value::Null),
        ResolvedValue::Conditional { if_true, .. } => resolved_to_route53_scenario_json(if_true),
        ResolvedValue::Reference { target, .. } => serde_json::json!({(FN_REF): target}),
        ResolvedValue::Dynamic { reason } => serde_json::json!({(MARKER_DYNAMIC): reason}),
        ResolvedValue::TypedDynamic { reason, param_type } => {
            serde_json::json!({(MARKER_DYNAMIC): reason, (MARKER_PARAM_TYPE): param_type})
        }
    }
}

fn scenario_property_json(properties: &ResolvedValue, property_name: &str) -> Option<serde_json::Value> {
    match properties {
        ResolvedValue::Concrete { value } => {
            value.get(property_name).filter(|property_value| !property_value.is_null()).cloned()
        }
        ResolvedValue::Map { entries } => {
            let property_value = &entries.iter().find(|entry| entry.key == property_name)?.value;
            if matches!(property_value, ResolvedValue::Concrete { value } if value.is_null()) {
                None
            } else {
                Some(resolved_to_route53_scenario_json(property_value))
            }
        }
        _ => None,
    }
}

fn route53_scenario_source_path(
    m: &SemanticModel,
    resource_id: &str,
    effective_path: &str,
    conditions: &HashMap<String, bool>,
) -> String {
    m.scenario_source_path(resource_id, effective_path, conditions).unwrap_or_else(|| effective_path.to_string())
}

fn push_route53_record_value_diagnostics<F>(
    diagnostics: &mut Vec<Diagnostic>,
    m: &SemanticModel,
    resource_id: &str,
    record_type: &str,
    records: &[serde_json::Value],
    records_path: &str,
    source_path_for: F,
) where
    F: Fn(&str) -> String,
{
    for (record_index, record) in records.iter().enumerate() {
        let Some(record) = record.as_str() else {
            continue;
        };
        let message = match record_type {
            "A" if record.parse::<Ipv4Addr>().is_err() => {
                Some(format!("'{}' is not a valid IPv4 address for record type 'A'", record))
            }
            "AAAA" if record.parse::<Ipv6Addr>().is_err() => {
                Some(format!("'{}' is not a valid IPv6 address for record type 'AAAA'", record))
            }
            "TXT" if !TXT_RECORD_RE.is_match(record) => {
                Some(format!("TXT record value '{}' must be enclosed in double quotes", record))
            }
            "CAA" if !CAA_RECORD_RE.is_match(record) => {
                Some(format!("CAA record value '{}' must match format: flag tag 'value'", record))
            }
            "MX" if !MX_RECORD_RE.is_match(record) => {
                Some(format!("MX record value '{}' must match format: priority domain", record))
            }
            _ => None,
        };
        if let Some(message) = message {
            let property_path = format!("{}.{}", records_path, record_index);
            let source_path = source_path_for(&property_path);
            diagnostics.push(make_resource_diagnostic_at_source(
                "E3023",
                &message,
                m,
                resource_id,
                &property_path,
                &source_path,
                None,
            ));
        }
    }

    let effective_record_count = records.iter().filter(|record| !record.is_null()).count();
    if record_type == "CNAME" && effective_record_count > 1 {
        let source_path = source_path_for(records_path);
        diagnostics.push(make_resource_diagnostic_at_source(
            "E3023",
            "CNAME records must have at most 1 ResourceRecord",
            m,
            resource_id,
            records_path,
            &source_path,
            None,
        ));
    }
}

/// The authored JSON form of a resource property, reconstructed from the resolved
/// value: `{"Ref": target}` for a `Ref`, `{"Fn::GetAtt": [target, attr]}` for a
/// `GetAtt`, or the literal for a concrete value. A `Ref` to a parameter resolves
/// to a dynamic value rather than a reference, so the reference graph is consulted
/// to recover the authored `Ref`/`GetAtt`. `None` when the property is absent or
/// its form is not one of these (e.g. an opaque function). Used to compare and
/// render `RestApiId` values as the author wrote them.
fn authored_form(m: &SemanticModel, rid: &str, path: &str) -> Option<serde_json::Value> {
    match m.resolve(rid, path) {
        Some(ResolvedValue::Reference { target, kind }) => {
            return authored_ref_form(target, kind);
        }
        Some(ResolvedValue::Concrete { value }) => return Some(value.0.clone()),
        _ => {}
    }
    // A Ref/GetAtt to a parameter (or another unresolved target) is a dynamic
    // value; recover the authored form from the reference graph edge anchored at
    // this property path.
    let prop_path = path.strip_prefix("Properties.").unwrap_or(path);
    let edge = m.graph.outgoing(rid).into_iter().find(|e| {
        e.source_path == path || e.source_path == prop_path || e.source_path == format!("Properties.{}", prop_path)
    })?;
    authored_ref_form(&edge.target, &edge.kind)
}

fn authored_ref_form(target: &str, kind: &RefKind) -> Option<serde_json::Value> {
    match kind {
        RefKind::Ref => Some(serde_json::json!({ FN_REF: target })),
        RefKind::GetAtt { attr } => Some(serde_json::json!({ "Fn::GetAtt": [target, attr] })),
        _ => None,
    }
}

/// Whether a Method's `RestApiId` and its Authorizer's `RestApiId` refer to
/// different REST APIs. Identity is compared first: two values that follow to the
/// same resource are equal even when authored differently (e.g. `Ref RestApi` and
/// `Fn::GetAtt RestApi.RestApiId` both resolve to that API). When either side is
/// not a reference to a resource (a literal id or a `Ref` to a parameter), the
/// authored values are compared structurally, matching how CloudFormation would
/// treat them at deploy time. A side whose value is absent is treated as
/// non-conflicting (a missing required `RestApiId` is reported elsewhere).
fn rest_api_ids_conflict(m: &SemanticModel, method_name: &str, authorizer_name: &str) -> bool {
    let method_target = m.follow_ref(method_name, "Properties.RestApiId");
    let authorizer_target = m.follow_ref(authorizer_name, "Properties.RestApiId");
    if let (Some(mt), Some(at)) = (method_target, authorizer_target) {
        return mt != at;
    }
    match (
        authored_form(m, method_name, "Properties.RestApiId"),
        authored_form(m, authorizer_name, "Properties.RestApiId"),
    ) {
        (Some(m_val), Some(a_val)) => m_val != a_val,
        // A missing RestApiId on either side is not a mismatch to report here.
        _ => false,
    }
}

/// Whether an S3 bucket has an *effective* OwnershipControls configuration: the
/// `OwnershipControls.Rules` array is present with at least one entry. An absent
/// `OwnershipControls`, an absent `Rules`, or an empty `Rules: []` is not
/// effective (S3 Object Ownership defaults to BucketOwnerEnforced, which disables
/// ACLs, so an `AccessControl` value would be ignored). The schema requires at
/// least one entry in `Rules` (`minItems: 1`).
fn has_effective_ownership_controls(m: &SemanticModel, rid: &str) -> bool {
    resolve_concrete(m, rid, "Properties.OwnershipControls.Rules")
        .as_ref()
        .and_then(|v| v.as_array().map(|rules| !rules.is_empty()))
        .unwrap_or(false)
}

fn is_valid_cidr_strict(s: &str) -> bool {
    s.parse::<ipnetwork::IpNetwork>()
        .map(|net| match net {
            ipnetwork::IpNetwork::V4(n) => {
                let ip: u32 = n.ip().into();
                let mask = !0u32 << (32 - n.prefix()) as u32;
                ip & !mask == 0
            }
            ipnetwork::IpNetwork::V6(_) => true,
        })
        .unwrap_or(false)
}

/// FromPort/ToPort form an ordered port range only for the TCP and UDP
/// protocols (named case-insensitively or by IP protocol number 6/17). For
/// icmp/icmpv6 the two fields carry the ICMP type and code, where -1 is a
/// wildcard, and every other protocol ignores the ports entirely, so no
/// ordering constraint applies.
fn sg_protocol_has_ordered_port_range(protocol: Option<&serde_json::Value>) -> bool {
    match protocol.and_then(coerce_to_string) {
        Some(proto) => {
            proto.eq_ignore_ascii_case("tcp") || proto.eq_ignore_ascii_case("udp") || proto == "6" || proto == "17"
        }
        None => false,
    }
}

const SG_INVERTED_PORT_RANGE_FIX: &str = "Set FromPort to a value less than or equal to ToPort";

fn sg_inverted_port_range_message(from_port: i64, to_port: i64) -> String {
    format!("FromPort {} is greater than ToPort {}", from_port, to_port)
}

/// The `(FromPort, ToPort)` pair when the protocol's ports form an ordered
/// range, both resolve to integers, and the range is inverted.
fn sg_inverted_port_range(
    protocol: Option<&serde_json::Value>,
    from_port: Option<&serde_json::Value>,
    to_port: Option<&serde_json::Value>,
) -> Option<(i64, i64)> {
    if !sg_protocol_has_ordered_port_range(protocol) {
        return None;
    }
    let from_port = from_port.and_then(coerce_to_integer)?;
    let to_port = to_port.and_then(coerce_to_integer)?;
    (from_port > to_port).then_some((from_port, to_port))
}

/// Protocols whose FromPort/ToPort must be present: TCP and UDP take a port
/// range and ICMP takes a type/code pair. ICMPv6 is deliberately absent - its
/// type/code are optional, and omitting them allows every type and code.
const SG_PORT_REQUIRED_PROTOCOL_NAMES: [&str; 9] = ["1", "icmp", "6", "tcp", "17", "udp", "TCP", "UDP", "ICMP"];
const SG_PORT_REQUIRED_PROTOCOL_NUMBERS: [i64; 3] = [1, 6, 17];

/// Protocols for which FromPort/ToPort carry meaning: a port range for
/// TCP/UDP, an ICMP type/code for ICMP and ICMPv6. Any other protocol -
/// including the all-protocols wildcard -1 - allows traffic on every port
/// regardless of the range given, so the ports are ignored.
const SG_PORT_MEANINGFUL_PROTOCOL_NAMES: [&str; 13] =
    ["1", "icmp", "6", "tcp", "17", "udp", "TCP", "UDP", "ICMP", "58", "icmpv6", "ICMPv6", "ICMPV6"];
const SG_PORT_MEANINGFUL_PROTOCOL_NUMBERS: [i64; 4] = [1, 6, 17, 58];

/// `None` when the protocol is not a resolved scalar (absent, still a
/// reference, or a non-integral number), so callers can neither require nor
/// flag ports for a value they cannot classify.
fn sg_protocol_is_listed(protocol: &serde_json::Value, names: &[&str], numbers: &[i64]) -> Option<bool> {
    match protocol {
        serde_json::Value::String(name) => Some(names.contains(&name.as_str())),
        serde_json::Value::Number(number) => number.as_i64().map(|value| numbers.contains(&value)),
        _ => None,
    }
}

fn sg_protocol_requires_ports(protocol: Option<&serde_json::Value>) -> bool {
    protocol
        .and_then(|value| {
            sg_protocol_is_listed(value, &SG_PORT_REQUIRED_PROTOCOL_NAMES, &SG_PORT_REQUIRED_PROTOCOL_NUMBERS)
        })
        .unwrap_or(false)
}

fn sg_protocol_ignores_ports(protocol: Option<&serde_json::Value>) -> bool {
    protocol
        .and_then(|value| {
            sg_protocol_is_listed(value, &SG_PORT_MEANINGFUL_PROTOCOL_NAMES, &SG_PORT_MEANINGFUL_PROTOCOL_NUMBERS)
        })
        .map(|is_meaningful| !is_meaningful)
        .unwrap_or(false)
}

pub fn eval_extra_resources(ctx: &EvalContext) -> Vec<Diagnostic> {
    let mut out = Vec::new();
    let m = ctx.model;
    let input = ctx.input;

    for (name, res) in &m.resources {
        // An AWS-namespaced type absent from the compiled schema set is a
        // typo or nonexistent type - CloudFormation owns the reserved
        // `AWS::` namespace, so the embedded catalog is authoritative for
        // it. Types in any other namespace (private registry types,
        // `Custom::` resources, modules, hook-shaped names) may be
        // registered per account/region, so they are skipped entirely
        // rather than guessed at.
        if res.resource_type.starts_with("AWS::")
            && !res.resource_type.ends_with("::MODULE")
            && !ctx.cached_data.known_types.contains(&res.resource_type)
        {
            out.push(make_resource_diagnostic(
                "F3006",
                &format!("Unknown resource type '{}'", res.resource_type),
                m,
                name,
                "",
                None,
            ));
        }
    }

    for name in m.resources_of_type("AWS::ECS::TaskDefinition") {
        if let Some(serde_json::Value::Array(cdefs)) = resolve_concrete(m, name, "Properties.ContainerDefinitions")
            && cdefs.len() > 1
            && cdefs.iter().all(|c| c.get("Essential").and_then(|e| e.as_bool()) == Some(false))
        {
            out.push(make_resource_diagnostic(
                "E3042",
                "At least one container definition must have Essential set to true",
                m,
                name,
                "",
                None,
            ));
        }
    }

    for name in m.resources_of_type("AWS::IAM::Policy") {
        push_identity_policy_findings(&mut out, m, name, "Properties.PolicyDocument");
    }
    let single_document_types = [
        ("AWS::IAM::ManagedPolicy", "Properties.PolicyDocument"),
        ("AWS::IAM::UserPolicy", "Properties.PolicyDocument"),
        ("AWS::IAM::RolePolicy", "Properties.PolicyDocument"),
        ("AWS::IAM::GroupPolicy", "Properties.PolicyDocument"),
        ("AWS::SSO::PermissionSet", "Properties.InlinePolicy"),
    ];
    for (resource_type, document_path) in single_document_types {
        for name in m.resources_of_type(resource_type) {
            push_identity_policy_findings(&mut out, m, name, document_path);
        }
    }
    for resource_type in ["AWS::IAM::Role", "AWS::IAM::User", "AWS::IAM::Group"] {
        for name in m.resources_of_type(resource_type) {
            for document_path in inline_identity_policy_document_paths(m, name, "Properties.Policies") {
                push_identity_policy_findings(&mut out, m, name, &document_path);
            }
        }
    }

    for name in m.resources_of_type("AWS::ECS::Service") {
        if resolve_concrete(m, name, "Properties.LaunchType").as_ref().and_then(|v| v.as_str()) == Some("FARGATE")
            && resolve_concrete(m, name, "Properties.SchedulingStrategy").as_ref().and_then(|v| v.as_str())
                == Some("DAEMON")
        {
            out.push(make_resource_diagnostic(
                "E3044",
                "Fargate launch type does not support DAEMON scheduling strategy",
                m,
                name,
                "Properties.SchedulingStrategy",
                Some("Use REPLICA scheduling strategy with Fargate"),
            ));
        }
    }

    for name in m.resources_of_type("AWS::EC2::SecurityGroup") {
        // Raw properties are used so that rule arrays containing dynamic Refs
        // (which concrete resolution skips) still have their scalar ports read.
        let Some(properties) = input
            .get(FIELD_RESOURCES)
            .and_then(|resources| resources.get(name.as_str()))
            .and_then(|r| r.get(FIELD_PROPERTIES))
        else {
            continue;
        };
        for direction in ["SecurityGroupIngress", "SecurityGroupEgress"] {
            let Some(rules) = properties.get(direction).and_then(|s| s.as_array()) else {
                continue;
            };
            for (idx, rule) in rules.iter().enumerate() {
                if let Some((from_port, to_port)) =
                    sg_inverted_port_range(rule.get("IpProtocol"), rule.get("FromPort"), rule.get("ToPort"))
                {
                    out.push(make_resource_diagnostic(
                        "E9002",
                        &sg_inverted_port_range_message(from_port, to_port),
                        m,
                        name,
                        &format!("Properties.{}.{}", direction, idx),
                        Some(SG_INVERTED_PORT_RANGE_FIX),
                    ));
                }
            }
        }
    }

    for rule_type in ["AWS::EC2::SecurityGroupIngress", "AWS::EC2::SecurityGroupEgress"] {
        for name in m.resources_of_type(rule_type) {
            let protocol = resolve_concrete(m, name, "Properties.IpProtocol");
            let from_port = resolve_concrete(m, name, "Properties.FromPort");
            let to_port = resolve_concrete(m, name, "Properties.ToPort");
            if let Some((from_port, to_port)) =
                sg_inverted_port_range(protocol.as_ref(), from_port.as_ref(), to_port.as_ref())
            {
                out.push(make_resource_diagnostic(
                    "E9002",
                    &sg_inverted_port_range_message(from_port, to_port),
                    m,
                    name,
                    "Properties.FromPort",
                    Some(SG_INVERTED_PORT_RANGE_FIX),
                ));
            }
        }
    }

    {
        // Inline SecurityGroup ingress/egress rules - access raw properties
        // to handle arrays containing dynamic Refs that resolve_concrete skips
        for name in m.resources_of_type("AWS::EC2::SecurityGroup") {
            let Some(res) = m.resources.get(name.as_str()) else {
                continue;
            };
            for direction in &["SecurityGroupIngress", "SecurityGroupEgress"] {
                let Some(rv) = res.properties.get(*direction) else {
                    continue;
                };
                let items: Vec<serde_json::Value> = match rv {
                    ResolvedValue::Concrete { value: v } => match v.as_array() {
                        Some(a) => a.clone(),
                        None => continue,
                    },
                    ResolvedValue::List { items } => items.iter().map(resolved_to_json_best_effort).collect(),
                    _ => continue,
                };
                for (idx, rule) in items.iter().enumerate() {
                    let proto = rule.get("IpProtocol");
                    let has_port = rule.get("FromPort").is_some() || rule.get("ToPort").is_some();
                    let val = proto.map(|p| p.to_string()).unwrap_or_default();
                    let val_display = val.trim_matches('"');
                    if sg_protocol_requires_ports(proto) && !has_port {
                        out.push(make_resource_diagnostic(
                            "E3687",
                            &format!(
                                "['FromPort', 'ToPort'] are required properties when using 'IpProtocol' value {}",
                                val_display
                            ),
                            m,
                            name,
                            &format!("Properties.{}.{}", direction, idx),
                            None,
                        ));
                    }
                    if sg_protocol_ignores_ports(proto) && has_port {
                        out.push(make_resource_diagnostic(
                            "W3687",
                            &format!(
                                "['FromPort', 'ToPort'] are ignored when using 'IpProtocol' value '{}'",
                                val_display
                            ),
                            m,
                            name,
                            &format!("Properties.{}.{}.FromPort", direction, idx),
                            None,
                        ));
                    }
                }
            }
        }

        // Standalone SecurityGroupIngress/SecurityGroupEgress resources
        for rtype in &["AWS::EC2::SecurityGroupIngress", "AWS::EC2::SecurityGroupEgress"] {
            for name in m.resources_of_type(rtype) {
                let proto = resolve_concrete(m, name, "Properties.IpProtocol");
                let has_port = m.has_property(name, "Properties.FromPort") || m.has_property(name, "Properties.ToPort");
                let val = proto.as_ref().map(|p| p.to_string()).unwrap_or_default();
                let val_display = val.trim_matches('"');
                if sg_protocol_requires_ports(proto.as_ref()) && !has_port {
                    out.push(make_resource_diagnostic(
                        "E3687",
                        &format!(
                            "['FromPort', 'ToPort'] are required properties when using 'IpProtocol' value {}",
                            val_display
                        ),
                        m,
                        name,
                        KEY_PROPERTIES,
                        None,
                    ));
                }
                if sg_protocol_ignores_ports(proto.as_ref()) && has_port {
                    out.push(make_resource_diagnostic(
                        "W3687",
                        &format!("['FromPort', 'ToPort'] are ignored when using 'IpProtocol' value '{}'", val_display),
                        m,
                        name,
                        "Properties.FromPort",
                        None,
                    ));
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::ECS::TaskDefinition") {
        if resolve_concrete(m, name, "Properties.NetworkMode").as_ref().and_then(|v| v.as_str()) == Some("awsvpc")
            && let Some(serde_json::Value::Array(cdefs)) = resolve_concrete(m, name, "Properties.ContainerDefinitions")
        {
            for (ci, cdef) in cdefs.iter().enumerate() {
                if let Some(pms) = cdef.get("PortMappings").and_then(|p| p.as_array()) {
                    for (pi, pm) in pms.iter().enumerate() {
                        let hp = pm.get("HostPort").and_then(coerce_port_to_string);
                        let cp = pm.get("ContainerPort").and_then(coerce_port_to_string);
                        if let (Some(h), Some(c)) = (hp, cp)
                            && h != c
                        {
                            out.push(make_resource_diagnostic(
                                "E3053",
                                &format!("HostPort {} must equal ContainerPort {} when NetworkMode is awsvpc", h, c),
                                m,
                                name,
                                &format!("Properties.ContainerDefinitions[{}].PortMappings[{}].HostPort", ci, pi),
                                Some("Set HostPort equal to ContainerPort or remove HostPort"),
                            ));
                        }
                    }
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::DynamoDB::Table") {
        let analysis = analyze_dynamodb_table_scenarios(m, name);
        for mismatch in &analysis.attribute_mismatches {
            let mut parts = Vec::new();
            if !mismatch.missing.is_empty() {
                parts.push(format!("missing definitions: [{}]", mismatch.missing.join(", ")));
            }
            if !mismatch.unused.is_empty() {
                parts.push(format!("unused definitions: [{}]", mismatch.unused.join(", ")));
            }
            let message = format!("AttributeDefinitions does not match KeySchema attributes. {}", parts.join("; "));
            out.push(make_resource_diagnostic("E3039", &message, m, name, KEY_PROPERTIES, None));
        }
        if analysis.explicit_provisioned_missing_throughput {
            out.push(make_resource_diagnostic(
                "E3639",
                "ProvisionedThroughput is required when BillingMode is 'PROVISIONED'",
                m,
                name,
                "Properties.ProvisionedThroughput",
                Some("Add ProvisionedThroughput or set BillingMode to 'PAY_PER_REQUEST'"),
            ));
        }
        if analysis.default_provisioned_missing_throughput {
            out.push(make_resource_diagnostic(
                "E3639",
                "ProvisionedThroughput is required when BillingMode defaults to 'PROVISIONED'",
                m,
                name,
                "Properties.ProvisionedThroughput",
                Some("Add ProvisionedThroughput or set BillingMode to 'PAY_PER_REQUEST'"),
            ));
        }
    }

    for name in m.resources_of_type("AWS::CloudFront::Distribution") {
        if let Some(dist) = resolve_concrete(m, name, "Properties.DistributionConfig") {
            // A TargetOriginId is valid if it matches the Id of any Origin or any
            // OriginGroup item. Only the DefaultCacheBehavior is validated.
            let mut origin_ids: HashSet<&str> = dist
                .get("Origins")
                .and_then(|o| o.as_array())
                .map(|arr| arr.iter().filter_map(|o| o.get("Id").and_then(|i| i.as_str())).collect())
                .unwrap_or_default();
            if let Some(items) = dist.get("OriginGroups").and_then(|g| g.get("Items")).and_then(|i| i.as_array()) {
                origin_ids.extend(items.iter().filter_map(|g| g.get("Id").and_then(|i| i.as_str())));
            }
            if let Some(target) =
                dist.get("DefaultCacheBehavior").and_then(|d| d.get("TargetOriginId")).and_then(|t| t.as_str())
                && !origin_ids.contains(target)
            {
                out.push(make_resource_diagnostic(
                    "E3057",
                    &format!("TargetOriginId '{}' does not match any Origin Id in the distribution", target),
                    m,
                    name,
                    "Properties.DistributionConfig.DefaultCacheBehavior.TargetOriginId",
                    Some("Set TargetOriginId to match one of the Origin Ids defined in Origins"),
                ));
            }
        }
    }

    for rpp in &ctx.cached_data.rule_tables.resource_policy_paths {
        let path = full_property_path(rpp);
        for name in m.resources_of_type(&rpp.resource_type) {
            if let Some(doc) = resolve_concrete(m, name, &path)
                && doc.is_object()
                && doc.get("Statement").is_none()
            {
                out.push(make_resource_diagnostic(
                    "E3512",
                    "Resource-based policy must have a Statement property",
                    m,
                    name,
                    &path,
                    Some("Add a Statement array to the policy document"),
                ));
            }
        }
    }

    for name in m.resources_of_type("AWS::IAM::Role") {
        if let Some(doc) = resolve_concrete(m, name, "Properties.AssumeRolePolicyDocument") {
            if doc.is_object() && doc.get("Statement").is_none() {
                out.push(make_resource_diagnostic(
                    "E3530",
                    "'Statement' is a required property",
                    m,
                    name,
                    "Properties.AssumeRolePolicyDocument",
                    Some("Add a Statement array to the AssumeRolePolicyDocument"),
                ));
            }
            if let Some(stmts) = doc.get("Statement").and_then(|s| s.as_array()) {
                for (idx, stmt) in stmts.iter().enumerate() {
                    if !stmt.is_object() {
                        continue;
                    }
                    let path = format!("Properties.AssumeRolePolicyDocument.Statement.{}", idx);
                    if stmt.get("Effect").is_none() {
                        out.push(make_resource_diagnostic(
                            "E3530",
                            "'Effect' is a required property in trust policy statement",
                            m,
                            name,
                            &path,
                            Some("Add Effect (Allow or Deny) to the statement"),
                        ));
                    }
                    if stmt.get("Principal").is_none() {
                        out.push(make_resource_diagnostic(
                            "E3530",
                            "'Principal' is a required property in trust policy statement",
                            m,
                            name,
                            &path,
                            Some("Add Principal to the statement"),
                        ));
                    }
                    if stmt.get("Action").is_none() && stmt.get("NotAction").is_none() {
                        out.push(make_resource_diagnostic(
                            "E3530",
                            "'Action' or 'NotAction' is a required property in trust policy statement",
                            m,
                            name,
                            &path,
                            Some("Add Action or NotAction to the statement"),
                        ));
                    }
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::ECR::Repository") {
        if let Some(doc) = resolve_concrete(m, name, "Properties.RepositoryPolicyText")
            && doc.is_object()
            && doc.get("Statement").is_none()
        {
            out.push(make_resource_diagnostic(
                "E3513",
                "ECR repository policy must have a Statement property",
                m,
                name,
                "Properties.RepositoryPolicyText",
                Some("Add a Statement array to the RepositoryPolicyText"),
            ));
        }
    }

    let policy_doc_types: &[(&str, &str)] = &[
        ("AWS::S3::BucketPolicy", "Properties.PolicyDocument.Version"),
        ("AWS::SNS::TopicPolicy", "Properties.PolicyDocument.Version"),
        ("AWS::SQS::QueuePolicy", "Properties.PolicyDocument.Version"),
        ("AWS::KMS::Key", "Properties.KeyPolicy.Version"),
        ("AWS::IAM::Role", "Properties.AssumeRolePolicyDocument.Version"),
        ("AWS::IAM::Policy", "Properties.PolicyDocument.Version"),
        ("AWS::IAM::ManagedPolicy", "Properties.PolicyDocument.Version"),
    ];
    for (rtype, path) in policy_doc_types {
        for name in m.resources_of_type(rtype) {
            let scenarios = m.resolve_scenarios_json(name, path);
            for (val, _conds) in &scenarios {
                // Only the older-but-valid policy version is worth a portability
                // warning (suggesting an upgrade). Any other string is either the
                // current version or an invalid value; invalid enum values are a
                // schema error, not this best-practice warning, so they are left
                // to the policy-document schema check.
                if val.as_str() == Some("2008-10-17") {
                    out.push(make_resource_diagnostic(
                        "W2511",
                        "IAM Policy Version should be updated to '2012-10-17'",
                        m,
                        name,
                        path,
                        Some("Update the policy document Version to '2012-10-17'"),
                    ));
                    break;
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::CodeBuild::Project") {
        if resolve_concrete(m, name, "Properties.Source.Type").as_ref().and_then(|v| v.as_str()) == Some("S3")
            && let Some(serde_json::Value::String(loc)) = resolve_concrete(m, name, "Properties.Source.Location")
            && !loc.contains('/')
        {
            out.push(make_resource_diagnostic(
                "E3636",
                &format!("CodeBuild S3 source location '{}' must be in 'bucket/key' format", loc),
                m,
                name,
                "Properties.Source.Location",
                Some("Use format: my-bucket/path/to/source.zip"),
            ));
        }
    }

    for name in m.resources_of_type("AWS::CodePipeline::Pipeline") {
        if let Some(serde_json::Value::Array(stages)) = resolve_concrete(m, name, "Properties.Stages")
            && let Some(first) = stages.first()
        {
            let has_source = first
                .get("Actions")
                .and_then(|a| a.as_array())
                .map(|actions| {
                    actions.iter().any(|a| {
                        a.get("ActionTypeId").and_then(|at| at.get("Category")).and_then(|c| c.as_str())
                            == Some("Source")
                    })
                })
                .unwrap_or(false);
            if !has_source {
                out.push(make_resource_diagnostic(
                    "E3700",
                    "First stage of a pipeline must contain at least one Source action",
                    m,
                    name,
                    "Properties.Stages[0]",
                    Some("Add an action with ActionTypeId.Category=Source to the first stage"),
                ));
            }
        }
    }

    for name in m.resources_of_type("AWS::Lambda::Function") {
        if let Some(serde_json::Value::Object(code)) = resolve_concrete(m, name, "Properties.Code")
            && code.contains_key("ZipFile")
            && let Some(serde_json::Value::String(rt)) = resolve_concrete(m, name, "Properties.Runtime")
            && !rt.starts_with("nodejs")
            && !rt.starts_with("python")
        {
            out.push(make_resource_diagnostic(
                "E3677",
                &format!("Runtime '{}' is not supported with Code.ZipFile - use nodejs or python", rt),
                m,
                name,
                "",
                None,
            ));
        }
    }

    for name in m.resources_of_type("AWS::SQS::Queue") {
        if let Some(serde_json::Value::Bool(true)) = resolve_concrete(m, name, "Properties.FifoQueue")
            && let Some(serde_json::Value::String(qname)) = resolve_concrete(m, name, "Properties.QueueName")
            && !qname.ends_with(".fifo")
        {
            out.push(make_resource_diagnostic(
                "E3501",
                &format!("FIFO queue name '{}' must end with '.fifo'", qname),
                m,
                name,
                "Properties.QueueName",
                Some("Append .fifo to the queue name"),
            ));
        }
    }

    if let Some(resources) = input.get(FIELD_RESOURCES).and_then(|r| r.as_object()) {
        for name in m.resources_of_type("AWS::SQS::Queue") {
            let is_fifo =
                resolve_concrete(m, name, "Properties.FifoQueue").as_ref().and_then(|v| v.as_bool()).unwrap_or(false);
            if let Some(edges) =
                resources.get(name.as_str()).and_then(|r| r.get(FIELD_OUTGOING_REFS)).and_then(|r| r.as_array())
            {
                for edge in edges {
                    let sp = edge.get(FIELD_SOURCE_PATH).and_then(|p| p.as_str()).unwrap_or("");
                    if !sp.contains("RedrivePolicy.deadLetterTargetArn") {
                        continue;
                    }
                    let target = match edge.get(FIELD_TARGET).and_then(|t| t.as_str()) {
                        Some(t) => t,
                        None => continue,
                    };
                    let dlq_res = match m.resources.get(target) {
                        Some(r) => r,
                        None => continue,
                    };
                    if dlq_res.resource_type != "AWS::SQS::Queue" {
                        continue;
                    }
                    let dlq_fifo = resolve_concrete(m, target, "Properties.FifoQueue")
                        .as_ref()
                        .and_then(|v| v.as_bool())
                        .unwrap_or(false);
                    if is_fifo != dlq_fifo {
                        let src_type = if is_fifo { "FIFO" } else { "standard" };
                        let dlq_type = if dlq_fifo { "FIFO" } else { "standard" };
                        out.push(make_resource_diagnostic(
                            "E3502",
                            &format!(
                                "Source queue type '{}' does not match destination queue type '{}'",
                                src_type, dlq_type
                            ),
                            m,
                            name,
                            "Properties.RedrivePolicy",
                            None,
                        ));
                    }
                }
            }
        }
    }

    for finding in duplicate_subnet_associations(m) {
        out.push(make_resource_diagnostic(
            "E3022",
            &finding.message,
            m,
            &finding.resource_id,
            "Properties.SubnetId",
            Some("Associate each subnet with exactly one route table"),
        ));
    }

    let creation_policy_types = [
        "AWS::AppStream::Fleet",
        "AWS::AutoScaling::AutoScalingGroup",
        "AWS::CloudFormation::WaitCondition",
        "AWS::EC2::Instance",
    ];
    let creation_policy_fix =
        format!("Remove CreationPolicy or change resource type to one of: {}", creation_policy_types.join(", "));
    let update_policy_types = &ctx.cached_data.rule_tables.update_policy_resource_types;
    let mut update_policy_types_for_message = update_policy_types.clone();
    update_policy_types_for_message.sort();
    let update_policy_fix = format!(
        "Remove UpdatePolicy or change resource type to one of: {}",
        update_policy_types_for_message.join(", ")
    );
    if let Some(resources) = input.get(FIELD_RESOURCES).and_then(|r| r.as_object()) {
        for (name, res) in resources {
            let creation_policy_status = m.lifecycle_attribute_status(name, KEY_CREATION_POLICY);
            if creation_policy_status.may_be_present {
                let rtype = res.get(FIELD_RESOURCE_TYPE).and_then(|t| t.as_str()).unwrap_or("");
                if !creation_policy_types.contains(&rtype) {
                    out.push(make_resource_diagnostic(
                        "E3055",
                        &format!("CreationPolicy is not supported on resource type '{}'", rtype),
                        m,
                        name,
                        KEY_CREATION_POLICY,
                        Some(&creation_policy_fix),
                    ));
                }
            }
            let update_policy_status = m.lifecycle_attribute_status(name, KEY_UPDATE_POLICY);
            if update_policy_status.may_be_present {
                let rtype = res.get(FIELD_RESOURCE_TYPE).and_then(|t| t.as_str()).unwrap_or("");
                if !update_policy_types.iter().any(|t| t == rtype) {
                    out.push(make_resource_diagnostic(
                        "E3016",
                        &format!("UpdatePolicy is not supported on resource type '{}'", rtype),
                        m,
                        name,
                        KEY_UPDATE_POLICY,
                        Some(&update_policy_fix),
                    ));
                } else if let Some(invalid_value) = update_policy_status.invalid_value.as_deref() {
                    out.push(make_resource_diagnostic(
                        "E3016",
                        &format!("{} is not of type 'object'", invalid_value),
                        m,
                        name,
                        KEY_UPDATE_POLICY,
                        None,
                    ));
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::S3::Bucket") {
        if m.resources.get(name.as_str()).map(|r| r.properties.contains_key("AccessControl")).unwrap_or(false) {
            out.push(make_resource_diagnostic(
                "W3045",
                "AccessControl property is deprecated. Use bucket policies instead",
                m,
                name,
                "Properties.AccessControl",
                Some("Remove AccessControl and use an AWS::S3::BucketPolicy resource"),
            ));
        }
    }

    for name in m.resources_of_type("AWS::S3::Bucket") {
        if let Some(res) = m.resources.get(name.as_str())
            && res.properties.contains_key("AccessControl")
            && let Some(serde_json::Value::String(ac)) = resolve_concrete(m, name, "Properties.AccessControl")
            // OwnershipControls is only required for ACLs that grant access to
            // other accounts. These owner-scoped ACLs need no OwnershipControl.
            && !matches!(ac.as_str(), "Private" | "BucketOwnerFullControl" | "BucketOwnerRead")
            && !has_effective_ownership_controls(m, name)
        {
            out.push(make_resource_diagnostic(
                "E3045",
                "A bucket with 'AccessControl' set should also have at least one 'OwnershipControl' configured",
                m,
                name,
                KEY_PROPERTIES,
                Some("Add OwnershipControls to the bucket when using AccessControl"),
            ));
        }
    }

    // W3663: a Lambda Permission needs SourceAccount when its SourceArn cannot by
    // itself pin the calling account. That is the case when the SourceArn either
    // references an S3 bucket (bucket ARNs carry no account id) or is a literal
    // ARN string with no ':<12-digit-account>:' segment. The Principal is
    // irrelevant - keying on it (e.g. "s3.amazonaws.com") both misses the string
    // case and over-fires when SourceArn is absent entirely.
    for name in m.resources_of_type("AWS::Lambda::Permission") {
        let source_account_set =
            m.resources.get(name.as_str()).map(|r| r.properties.contains_key("SourceAccount")).unwrap_or(false);
        if source_account_set {
            continue;
        }
        let source_arn_is_s3_bucket = m
            .follow_ref(name, "Properties.SourceArn")
            .and_then(|target| m.resources.get(target))
            .map(|r| r.resource_type == "AWS::S3::Bucket")
            .unwrap_or(false);
        // The string branch only tests a genuinely literal SourceArn. A value
        // supplied via a Ref/GetAtt/Sub (even one that resolves to a concrete
        // account-less string) is not folded into the pattern check - only a
        // literal string is pattern-matched here.
        let source_arn_string_without_account = !m.is_from_intrinsic(name, "Properties.SourceArn")
            && resolve_concrete(m, name, "Properties.SourceArn")
                .as_ref()
                .and_then(|v| v.as_str())
                .map(|arn| !ARN_HAS_ACCOUNT_RE.is_match(arn))
                .unwrap_or(false);
        if source_arn_is_s3_bucket || source_arn_string_without_account {
            out.push(make_resource_diagnostic(
                "W3663",
                "Lambda Permission with a SourceArn that has no account id should also specify SourceAccount",
                m,
                name,
                KEY_PROPERTIES,
                Some("Add SourceAccount property"),
            ));
        }
    }

    let sub_filters = m.resources_of_type("AWS::Logs::SubscriptionFilter");
    let mut lg_counts: HashMap<String, Vec<&str>> = HashMap::new();
    for name in sub_filters {
        if let Some(serde_json::Value::String(lg)) = resolve_concrete(m, name, "Properties.LogGroupName") {
            lg_counts.entry(lg).or_default().push(name);
        } else if let Some(ref_target) = m.follow_ref(name, "Properties.LogGroupName") {
            let key = format!("__ref:{}", ref_target);
            lg_counts.entry(key).or_default().push(name);
        }
    }
    for (lg, names) in &lg_counts {
        if names.len() > 2 {
            let display = lg.strip_prefix("__ref:").unwrap_or(lg);
            out.push(make_resource_diagnostic(
                "E2529",
                &format!("Log group '{}' has {} subscription filters, maximum is 2", display, names.len()),
                m,
                names[0],
                "",
                None,
            ));
        }
    }

    for name in m.resources_of_type("AWS::Lambda::Function") {
        if let Some(snap) = resolve_concrete(m, name, "Properties.SnapStart")
            && snap.get("ApplyOn").and_then(|a| a.as_str()) == Some("PublishedVersions")
            && let Some(serde_json::Value::String(rt)) = resolve_concrete(m, name, "Properties.Runtime")
            && !snapstart_runtime_supported(
                &rt,
                &ctx.cached_data.rule_tables.snapstart_runtime_prefixes,
                &ctx.cached_data.rule_tables.snapstart_unsupported_runtime_prefixes,
                &ctx.cached_data.rule_tables.snapstart_unsupported_runtimes,
            )
        {
            out.push(make_resource_diagnostic(
                "E2530",
                &format!("SnapStart is not supported with runtime '{}'", rt),
                m,
                name,
                "Properties.SnapStart.ApplyOn",
                Some("Use a supported Python, Java, or .NET runtime"),
            ));
        }
    }

    // SnapStart enabled in an unsupported region. Only fires when a
    // region is explicitly configured and absent from the supported regions list.
    let snapstart_supported_regions = &ctx.cached_data.rule_tables.snapstart_supported_regions;
    if let Some(region) = ctx.region.as_ref()
        && !snapstart_supported_regions.iter().any(|sr| sr == region)
    {
        for name in m.resources_of_type("AWS::Lambda::Function") {
            if let Some(snap) = resolve_concrete(m, name, "Properties.SnapStart")
                && snap.get("ApplyOn").and_then(|a| a.as_str()) == Some("PublishedVersions")
            {
                out.push(make_resource_diagnostic(
                    "E2530",
                    &format!("SnapStart is not supported in region '{}'", region),
                    m,
                    name,
                    "Properties.SnapStart.ApplyOn",
                    Some("Deploy to a region that supports SnapStart or disable SnapStart"),
                ));
            }
        }
    }

    for name in m.resources_of_type("AWS::CloudFront::Distribution") {
        if let Some(serde_json::Value::Array(aliases)) =
            resolve_concrete(m, name, "Properties.DistributionConfig.Aliases")
        {
            for (i, alias) in aliases.iter().enumerate() {
                if let Some(s) = alias.as_str() {
                    let path = format!("Properties.DistributionConfig.Aliases.{}", i);
                    // Wildcard must only appear as the leftmost label (e.g. *.example.com).
                    // `email.*.example.com` or any `.*.` in the middle is invalid.
                    if s.contains(".*.") {
                        out.push(make_resource_diagnostic(
                            "E3013",
                            &format!("CloudFront alias '{}' has wildcard in invalid position", s),
                            m,
                            name,
                            &path,
                            None,
                        ));
                    } else if !DOMAIN_RE.is_match(s) {
                        out.push(make_resource_diagnostic(
                            "E3013",
                            &format!("CloudFront alias '{}' is not a valid domain name", s),
                            m,
                            name,
                            &path,
                            None,
                        ));
                    }
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::CodeBuild::Project") {
        if let Some(target) = m.follow_ref(name, "Properties.ServiceRole")
            && let Some(target_res) = m.resources.get(target)
            && target_res.resource_type == "AWS::IAM::Role"
            && let Some(serde_json::Value::String(path)) = resolve_concrete(m, target, "Properties.Path")
            && path != "/"
        {
            out.push(make_resource_diagnostic(
                "E3050",
                &format!("Ref to IAM role '{}' with Path '{}' - use GetAtt {}.Arn instead", target, path, target),
                m,
                name,
                "Properties.ServiceRole",
                Some("Switch from Ref to !GetAtt <Role>.Arn when Path is not '/'"),
            ));
        }
    }

    for (name, res) in &m.resources {
        if res.properties.len() > 50 {
            out.push(make_resource_diagnostic(
                "E3010",
                &format!("Resource has {} properties, maximum is 50", res.properties.len()),
                m,
                name,
                "",
                None,
            ));
        }
    }

    // Advisory for duplicate scalar items in a list that permits duplicates. A
    // list whose schema requires uniqueItems is covered by the Fatal uniqueItems
    // check instead, so it is excluded here. The `Command` property of
    // run-command resources legitimately repeats values and is exempt.
    {
        let schema_metadata = ctx.cached_data.schema_metadata_catalog();
        for (name, res) in &m.resources {
            let Some(type_meta) = schema_metadata.get(&res.resource_type) else {
                continue;
            };
            for prop in res.properties.keys() {
                if prop == "Command" {
                    continue;
                }
                // Only a property the schema actually defines can be a "list that
                // permits duplicates"; an unknown property is a structural error
                // handled elsewhere.
                if !type_meta.property_types.contains_key(prop) {
                    continue;
                }
                let requires_unique = type_meta
                    .property_constraints
                    .get(prop)
                    .and_then(|constraints| constraints.unique_items)
                    .unwrap_or(false);
                if requires_unique {
                    continue;
                }
                let path = format!("Properties.{}", prop);
                let Some(rv) = m.resolve_deep(name, &path).or_else(|| m.resolve(name, &path).cloned()) else {
                    continue;
                };
                let ResolvedValue::Concrete { value: arr } = &rv else {
                    continue;
                };
                let Some(items) = arr.as_array() else {
                    continue;
                };
                // Distinct intrinsics (e.g. two different Refs) can each resolve to
                // the same placeholder and look like duplicates; only literal items
                // are advisory, so skip the list if any element came from an
                // intrinsic function.
                let any_intrinsic_item =
                    (0..items.len()).any(|i| m.is_from_intrinsic(name, &format!("{}.{}", path, i)));
                if m.is_from_intrinsic(name, &path) || any_intrinsic_item {
                    continue;
                }
                if let Some(dup) = first_duplicate_scalar(items) {
                    out.push(make_resource_diagnostic(
                        "I3037",
                        &format!("Array property '{}' contains duplicate value: {}", prop, dup),
                        m,
                        name,
                        &path,
                        None,
                    ));
                }
            }
        }
    }

    let prev_gen_re = &ctx.cached_data.previous_generation_instance_regex;
    for rpp in &ctx.cached_data.rule_tables.previous_generation_instance_property_paths {
        let path = full_property_path(rpp);
        for name in m.resources_of_type(&rpp.resource_type) {
            // Only literal string instance types are checked; a value that comes
            // from a parameter Ref or other intrinsic is left alone because its
            // deploy-time value is not known here (the default is just one of many
            // possible values). Skip those to avoid flagging a parameter default
            // the author may never use.
            if m.is_from_parameter(name, &path) || m.is_from_intrinsic(name, &path) {
                continue;
            }
            if let Some(serde_json::Value::String(val)) = resolve_concrete(m, name, &path)
                && prev_gen_re.is_match(&val)
            {
                out.push(make_resource_diagnostic(
                    "I3100",
                    &format!("Previous generation instance type '{}' - consider upgrading", val),
                    m,
                    name,
                    &path,
                    Some("Upgrade to a current generation instance type"),
                ));
            }
        }
    }

    for name in m.resources_of_type("AWS::ECS::Service") {
        if let Some(target) = m.follow_ref(name, "Properties.TaskDefinition")
            && let Some(_td) = m.resources.get(target)
            && resolve_concrete(m, target, "Properties.NetworkMode").as_ref().and_then(|v| v.as_str()) == Some("awsvpc")
            && !m
                .resources
                .get(name.as_str())
                .map(|r| r.properties.contains_key("NetworkConfiguration"))
                .unwrap_or(false)
        {
            out.push(make_resource_diagnostic(
                "E3052",
                "NetworkConfiguration required when TaskDefinition NetworkMode is 'awsvpc'",
                m,
                name,
                "",
                None,
            ));
        }
    }

    // Fires both when the TaskDefinition omits RequiresCompatibilities entirely and
    // when it's present but missing the FARGATE value.
    for name in m.resources_of_type("AWS::ECS::Service") {
        if resolve_concrete(m, name, "Properties.LaunchType").as_ref().and_then(|v| v.as_str()) != Some("FARGATE") {
            continue;
        }
        let Some(target) = m.follow_ref(name, "Properties.TaskDefinition") else {
            continue;
        };
        // FARGATE is only required in RequiresCompatibilities when the task
        // definition uses a non-awsvpc network mode. An awsvpc task definition is
        // already Fargate-compatible, so it is exempt.
        if resolve_concrete(m, target, "Properties.NetworkMode").as_ref().and_then(|v| v.as_str()) == Some("awsvpc") {
            continue;
        }
        let compat_rv = m
            .resolve_deep(target, "Properties.RequiresCompatibilities")
            .or_else(|| m.resolve(target, "Properties.RequiresCompatibilities").cloned());
        let compat_strings: Vec<&str> = match compat_rv.as_ref() {
            Some(ResolvedValue::Concrete { value: v }) => {
                v.as_array().map(|arr| arr.iter().filter_map(|x| x.as_str()).collect()).unwrap_or_default()
            }
            Some(ResolvedValue::List { items }) => items
                .iter()
                .filter_map(|it| match it {
                    ResolvedValue::Concrete { value: v } => v.as_str(),
                    _ => None,
                })
                .collect(),
            _ => Vec::new(),
        };
        if compat_strings.contains(&"FARGATE") {
            continue;
        }
        let rendered = if compat_strings.is_empty() { render_str_list([""]) } else { render_str_list(&compat_strings) };
        out.push(make_resource_diagnostic(
            "E3054",
            &format!("{} does not contain items matching 'FARGATE'", rendered),
            m,
            target,
            "Properties.RequiresCompatibilities",
            Some("Add 'FARGATE' to RequiresCompatibilities"),
        ));
    }

    for name in m.resources_of_type("AWS::CodePipeline::Pipeline") {
        if let Some(serde_json::Value::Array(stages)) = resolve_concrete(m, name, "Properties.Stages") {
            let mut seen_outputs = HashSet::new();
            for (si, stage) in stages.iter().enumerate() {
                let stage_name = stage.get("Name").and_then(|n| n.as_str()).unwrap_or("unknown");
                if let Some(actions) = stage.get("Actions").and_then(|a| a.as_array()) {
                    for action in actions {
                        let aname = action.get("Name").and_then(|n| n.as_str()).unwrap_or("unknown");
                        if let Some(outs) = action.get("OutputArtifacts").and_then(|o| o.as_array()) {
                            for o in outs {
                                if let Some(n) = o.get("Name").and_then(|n| n.as_str())
                                    && !seen_outputs.insert(n.to_string())
                                {
                                    out.push(make_resource_diagnostic(
                                        "E3701",
                                        &format!("Duplicate OutputArtifact name '{}'", n),
                                        m,
                                        name,
                                        "",
                                        None,
                                    ));
                                }
                            }
                        }
                        if si > 0
                            && let Some(ins) = action.get("InputArtifacts").and_then(|i| i.as_array())
                        {
                            for i in ins {
                                if let Some(n) = i.get("Name").and_then(|n| n.as_str())
                                    && !seen_outputs.contains(n)
                                {
                                    out.push(make_resource_diagnostic("E3701", &format!("InputArtifact '{}' in stage '{}' action '{}' does not reference a previously defined OutputArtifact", n, stage_name, aname), m, name, "", None));
                                }
                            }
                        }
                    }
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::CodePipeline::Pipeline") {
        // Resolve Stages preserving Fn::If so artifact lists authored behind a
        // conditional keep both branches; the count check enumerates them.
        let stages_json =
            m.resolve_deep(name, "Properties.Stages").map(|rv| resolved_to_json_preserving_conditionals(&rv));
        if let Some(serde_json::Value::Array(stages)) = stages_json {
            for stage in &stages {
                if let Some(actions) = stage.get("Actions").and_then(|a| a.as_array()) {
                    for action in actions {
                        let action_type_id = action.get("ActionTypeId");
                        let owner = action_type_id.and_then(|a| a.get("Owner")).and_then(|c| c.as_str());
                        let category = action_type_id.and_then(|a| a.get("Category")).and_then(|c| c.as_str());
                        let provider = action_type_id.and_then(|a| a.get("Provider")).and_then(|c| c.as_str());
                        // Artifact-count constraints are keyed on the full
                        // Owner/Category/Provider tuple. A category alone is
                        // ambiguous (e.g. AWS/Deploy/CloudFormation allows 0 input
                        // artifacts while AWS/Deploy/CodeDeploy requires 1). Skip
                        // when the tuple is unknown.
                        let (Some(owner), Some(category), Some(provider)) = (owner, category, provider) else {
                            continue;
                        };
                        let aname = action.get("Name").and_then(|n| n.as_str()).unwrap_or("unknown");
                        let key = format!("{owner}/{category}/{provider}");
                        if let Some(counts) = ctx.cached_data.codepipeline_artifact_counts.get(&key) {
                            // An artifact list may be authored directly or wrapped
                            // in an Fn::If; enumerate every branch's count (walk each
                            // path) so a violation in ANY branch is reported, and
                            // dedupe so equal counts are not double-reported.
                            for actual_in in artifact_count_scenarios(action.get("InputArtifacts")) {
                                if actual_in < counts.min_input {
                                    out.push(make_resource_diagnostic(
                                        "E3702",
                                        &format!(
                                            "Action '{}' ({}) has {} input artifacts, expected at least {}",
                                            aname, key, actual_in, counts.min_input
                                        ),
                                        m,
                                        name,
                                        "",
                                        None,
                                    ));
                                }
                                if actual_in > counts.max_input {
                                    out.push(make_resource_diagnostic(
                                        "E3702",
                                        &format!(
                                            "Action '{}' ({}) has {} input artifacts, expected at most {}",
                                            aname, key, actual_in, counts.max_input
                                        ),
                                        m,
                                        name,
                                        "",
                                        None,
                                    ));
                                }
                            }
                            for actual_out in artifact_count_scenarios(action.get("OutputArtifacts")) {
                                if actual_out < counts.min_output {
                                    out.push(make_resource_diagnostic(
                                        "E3702",
                                        &format!(
                                            "Action '{}' ({}) has {} output artifacts, expected at least {}",
                                            aname, key, actual_out, counts.min_output
                                        ),
                                        m,
                                        name,
                                        "",
                                        None,
                                    ));
                                }
                                if actual_out > counts.max_output {
                                    out.push(make_resource_diagnostic(
                                        "E3702",
                                        &format!(
                                            "Action '{}' ({}) has {} output artifacts, expected at most {}",
                                            aname, key, actual_out, counts.max_output
                                        ),
                                        m,
                                        name,
                                        "",
                                        None,
                                    ));
                                }
                            }
                        }
                    }
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::CodePipeline::Pipeline") {
        if let Some(serde_json::Value::Array(stages)) = resolve_concrete(m, name, "Properties.Stages") {
            for stage in &stages {
                if let Some(actions) = stage.get("Actions").and_then(|a| a.as_array()) {
                    for action in actions {
                        if let Some(tp) =
                            action.get("Configuration").and_then(|c| c.get("TemplatePath")).and_then(|t| t.as_str())
                            && tp.contains("::")
                        {
                            let artifact = tp.split("::").next().unwrap_or("");
                            let input_names: HashSet<&str> = action
                                .get("InputArtifacts")
                                .and_then(|i| i.as_array())
                                .map(|arr| arr.iter().filter_map(|a| a.get("Name").and_then(|n| n.as_str())).collect())
                                .unwrap_or_default();
                            if !input_names.contains(artifact) {
                                out.push(make_resource_diagnostic(
                                    "E3703",
                                    &format!("TemplatePath artifact '{}' is not one of the InputArtifacts", artifact),
                                    m,
                                    name,
                                    "",
                                    None,
                                ));
                            }
                        }
                    }
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::Lambda::EventSourceMapping") {
        if let Some(target) = m.follow_ref(name, "Properties.EventSourceArn")
            && let Some(sqs) = m.resources.get(target)
            && sqs.resource_type == "AWS::SQS::Queue"
        {
            let vis = resolve_concrete(m, target, "Properties.VisibilityTimeout").as_ref().and_then(coerce_to_integer);
            if let Some(fn_name) = m.follow_ref(name, "Properties.FunctionName") {
                let timeout = resolve_concrete(m, fn_name, "Properties.Timeout")
                    .as_ref()
                    .and_then(coerce_to_integer)
                    .unwrap_or(3);
                // CloudFormation requires the queue's VisibilityTimeout to be at
                // least the function Timeout. The finding is anchored on the
                // queue's VisibilityTimeout property (the value that must be raised).
                if let Some(v) = vis
                    && v < timeout
                {
                    out.push(make_resource_diagnostic(
                        "E3505",
                        &format!(
                            "Queue visibility timeout ({}) is less than Function timeout ({}) seconds",
                            v, timeout
                        ),
                        m,
                        target,
                        "Properties.VisibilityTimeout",
                        Some("Set the SQS VisibilityTimeout to at least the Lambda function Timeout"),
                    ));
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::SSM::Document") {
        if let Some(content) = resolve_concrete(m, name, "Properties.Content")
            && content.is_object()
            && content.get("schemaVersion").is_none()
        {
            out.push(make_resource_diagnostic(
                "E3051",
                "SSM Document Content must include 'schemaVersion'",
                m,
                name,
                "Properties.Content",
                None,
            ));
        }
    }

    for name in m.resources_of_type("AWS::S3::Bucket") {
        if let Some(serde_json::Value::Array(configs)) =
            resolve_concrete(m, name, "Properties.IntelligentTieringConfigurations")
        {
            for (config_idx, config) in configs.iter().enumerate() {
                if let Some(tierings) = config.get("Tierings").and_then(|t| t.as_array()) {
                    for (tier_idx, tier) in tierings.iter().enumerate() {
                        if let Some(days) = tier.get("Days").and_then(|d| d.as_i64()) {
                            let access_tier = tier.get("AccessTier").and_then(|a| a.as_str()).unwrap_or("");
                            let path = format!(
                                "Properties.IntelligentTieringConfigurations[{}].Tierings[{}].Days",
                                config_idx, tier_idx
                            );
                            if access_tier == "ARCHIVE_ACCESS" && !(90..=730).contains(&days) {
                                out.push(make_resource_diagnostic(
                                    "E3061",
                                    &format!("Days {} for ARCHIVE_ACCESS must be between 90 and 730", days),
                                    m,
                                    name,
                                    &path,
                                    Some("Set Days between 90 and 730"),
                                ));
                            }
                            if access_tier == "DEEP_ARCHIVE_ACCESS" && !(180..=730).contains(&days) {
                                out.push(make_resource_diagnostic(
                                    "E3061",
                                    &format!("Days {} for DEEP_ARCHIVE_ACCESS must be between 180 and 730", days),
                                    m,
                                    name,
                                    &path,
                                    Some("Set Days between 180 and 730"),
                                ));
                            }
                        }
                    }
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::Lambda::Function") {
        if let Some(snap) = resolve_concrete(m, name, "Properties.SnapStart")
            && snap.get("ApplyOn").and_then(|a| a.as_str()) == Some("PublishedVersions")
        {
            let version_refs = m.graph.ref_sources(name);
            let has_ver = version_refs
                .iter()
                .any(|src| m.resources.get(&**src).map(|r| r.resource_type == "AWS::Lambda::Version").unwrap_or(false));
            if !has_ver {
                out.push(make_resource_diagnostic(
                    "W2530",
                    "SnapStart is enabled but no AWS::Lambda::Version resource is attached",
                    m,
                    name,
                    "Properties.SnapStart",
                    Some("Add an AWS::Lambda::Version resource that references this function"),
                ));
            }
        }
    }

    for rpp in &ctx.cached_data.rule_tables.iam_role_arn_property_paths {
        let path = full_property_path(rpp);
        for name in m.resources_of_type(&rpp.resource_type) {
            if let Some(serde_json::Value::String(val)) = resolve_concrete(m, name, &path)
                && !ARN_RE.is_match(&val)
            {
                out.push(make_resource_diagnostic(
                    "E3511",
                    &format!("IAM Role ARN '{}' does not match expected pattern", val),
                    m,
                    name,
                    &path,
                    None,
                ));
            }
        }
    }

    for (name, res) in &m.resources {
        for (prop, val) in &res.properties {
            if !prop.ends_with("Arn") {
                continue;
            }
            // Collapse an Fn::If-wrapped value to its resolved (true-branch) form
            // before inspecting it, so a hardcoded ARN behind Fn::If is still
            // flagged. Matching the raw `val` against Concrete alone would skip a
            // Conditional. Every matching property is reported, not just the first.
            let resolved = resolved_to_json_best_effort(val);
            if let Some(s) = resolved.as_str()
                && s.starts_with("arn:")
                && !crate::functions::contains_unresolvable_content(val)
            {
                out.push(make_resource_diagnostic(
                    "W9002",
                    &format!("Property '{}' has a hardcoded ARN - use Ref, GetAtt, or a parameter instead", prop),
                    m,
                    name,
                    &format!("Properties.{}", prop),
                    None,
                ));
            }
        }
    }

    let has_sam_for_w1020 = m.transforms.iter().any(|t| t == TRANSFORM_SERVERLESS);
    if !has_sam_for_w1020 {
        for (name, res) in &m.resources {
            for p in &res.diagnostics.redundant_subs {
                out.push(make_resource_diagnostic(
                    "W1020",
                    "Fn::Sub isn't needed because there are no variables",
                    m,
                    name,
                    p,
                    None,
                ));
            }
        }
    }

    for (name, res) in &m.resources {
        for (prop, val) in &res.properties {
            check_dynamic_ref_spaces(&mut out, m, name, &format!("Properties.{}", prop), val);
        }
    }

    if let Some(resources) = input.get(FIELD_RESOURCES).and_then(|r| r.as_object()) {
        for (name, res) in resources {
            if let Some(edges) = res.get(FIELD_OUTGOING_REFS).and_then(|r| r.as_array()) {
                for edge in edges {
                    let target = edge.get(FIELD_TARGET).and_then(|t| t.as_str()).unwrap_or("");
                    let sp = edge.get(FIELD_SOURCE_PATH).and_then(|p| p.as_str()).unwrap_or("");
                    if sp.starts_with("Metadata")
                        && let Some(param) = m.parameters.get(target)
                        && param.no_echo
                    {
                        out.push(make_resource_diagnostic(
                            "W2010",
                            &format!("Don't use 'NoEcho' parameter '{}' in resource metadata", target),
                            m,
                            name,
                            sp,
                            Some("Move the parameter reference out of Metadata or remove NoEcho"),
                        ));
                    }
                }
            }
        }
    }

    if let Some(resources) = input.get(FIELD_RESOURCES).and_then(|r| r.as_object()) {
        for (name, res) in resources {
            if let Some(edges) = res.get(FIELD_OUTGOING_REFS).and_then(|r| r.as_array()) {
                for edge in edges {
                    if edge.get(FIELD_KIND).and_then(|k| k.as_str()) == Some(EDGE_KIND_SELECT)
                        && let Some(idx) = edge.get("index").and_then(|i| i.as_i64())
                        && idx < 0
                    {
                        out.push(make_resource_diagnostic(
                            "F1050",
                            "Fn::Select index must be a non-negative integer",
                            m,
                            name,
                            "",
                            None,
                        ));
                    }
                }
            }
        }
    }

    const UNIQUE_ARRAY_PROPS: &[&str] = &[
        "AvailabilityZones",
        "SecurityGroupIds",
        "SecurityGroups",
        "SubnetIds",
        "Subnets",
        "RequiresCompatibilities",
        "PlacementConstraints",
    ];
    // A property whose items must be unique must not repeat a value. Two entries
    // are only reported when they are provably one value: an entry that resolves
    // before deployment is settled by its contents, and an entry that stays opaque
    // is settled by the expression that produces it, since two entries written the
    // same way always read the same thing. When any entry offers neither, the
    // property is left alone - entries that merely look alike because their values
    // are unknowable are not a duplicate.
    for (name, res) in &m.resources {
        for prop in res.properties.keys() {
            if !UNIQUE_ARRAY_PROPS.contains(&prop.as_str()) {
                continue;
            }
            let path = format!("Properties.{}", prop);
            let resolved = m.resolve_deep(name, &path).or_else(|| m.resolve(name, &path).cloned());
            let Some(rv) = resolved else {
                continue;
            };
            let item_count = match &rv {
                ResolvedValue::Concrete { value: v } => match v.as_array() {
                    Some(arr) => arr.len(),
                    None => continue,
                },
                ResolvedValue::List { items } => items.len(),
                _ => continue,
            };
            if item_count < 2 {
                continue;
            }
            let identities: Vec<String> =
                (0..item_count).filter_map(|index| m.value_identity(name, &format!("{}.{}", path, index))).collect();
            if identities.len() != item_count {
                continue;
            }
            let mut seen = HashSet::new();
            let has_dup = identities.iter().any(|identity| !seen.insert(identity.clone()));
            if has_dup {
                out.push(make_resource_diagnostic(
                    "W9007",
                    &format!("Array property '{}' contains duplicate values", prop),
                    m,
                    name,
                    &path,
                    None,
                ));
            }
        }
    }

    if let Some(mappings) = input.get(FIELD_MAPPINGS).and_then(|m| m.as_object()) {
        for (map_name, level1) in mappings {
            if !level1.is_object() {
                out.push(make_resource_diagnostic(
                    "F0050",
                    &format!("Mapping '{}' must be a map", map_name),
                    m,
                    "",
                    &format!("Mappings/{}", map_name),
                    None,
                ));
            } else if let Some(obj) = level1.as_object() {
                for (k1, level2) in obj {
                    if !level2.is_object() {
                        out.push(make_resource_diagnostic(
                            "F0050",
                            &format!("Mapping '{}' second level key '{}' must be a map", map_name, k1),
                            m,
                            "",
                            &format!("Mappings/{}/{}", map_name, k1),
                            None,
                        ));
                    }
                }
            }
        }
    }

    {
        if let Some(resources) = input.get(FIELD_RESOURCES).and_then(|r| r.as_object()) {
            for (name, res) in resources {
                if let Some(edges) = res.get(FIELD_OUTGOING_REFS).and_then(|r| r.as_array()) {
                    for edge in edges {
                        if edge.get(FIELD_KIND).and_then(|k| k.as_str()) != Some(EDGE_KIND_REF) {
                            continue;
                        }
                        let target = edge.get(FIELD_TARGET).and_then(|t| t.as_str()).unwrap_or("");
                        let sp = edge.get(FIELD_SOURCE_PATH).and_then(|p| p.as_str()).unwrap_or("");
                        if let Some(param) = m.parameters.get(target) {
                            if let Some(ref def) = param.default {
                                // ImageId: only fire if Default fails AMI pattern
                                if sp.ends_with("ImageId")
                                    && param.param_type == PARAM_TYPE_STRING
                                    && !AMI_ID_RE.is_match(def)
                                {
                                    out.push(make_resource_diagnostic(
                                        "W1030",
                                        &format!(
                                            "{{'Ref': '{}'}} is not a 'AWS::EC2::Image.Id' when 'Ref' is resolved",
                                            target
                                        ),
                                        m,
                                        name,
                                        sp,
                                        Some("Use parameter type AWS::EC2::Image::Id"),
                                    ));
                                }
                                // CidrBlock: fire if Default fails strict CIDR validation (host bits set)
                                if (sp.ends_with("CidrBlock") || sp.ends_with("DestinationCidrBlock"))
                                    && param.param_type == PARAM_TYPE_STRING
                                    && !is_valid_cidr_strict(def)
                                {
                                    out.push(make_resource_diagnostic(
                                        "W1030",
                                        &format!(
                                            "{{'Ref': '{}'}} is not a 'ipv4-network' when 'Ref' is resolved",
                                            target
                                        ),
                                        m,
                                        name,
                                        sp,
                                        Some("Validate the parameter value matches CIDR format"),
                                    ));
                                }
                            }

                            // SecurityGroup.Id: not validated without concrete default
                            // Subnet.Id: not validated without concrete default
                            // VPC.Id: not validated without concrete default

                            // VPC.Id: parameter default fails VPC ID pattern
                            if sp.ends_with("VpcId")
                                && param.param_type == "AWS::EC2::VPC::Id"
                                && let Some(ref def) = param.default
                                && !W1030_VPC_RE.is_match(def)
                            {
                                out.push(make_resource_diagnostic("W1030",
                                            &format!("{{'Ref': '{}'}} is not a 'AWS::EC2::VPC.Id' with pattern '^vpc-(([0-9A-Fa-f]{{8}})|([0-9A-Fa-f]{{17}}))$' when 'Ref' is resolved", target),
                                            m, name, sp, None));
                            }
                        }
                    }
                }
            }
        }
    }

    let getatt_format: HashMap<(&str, &str), &str> = [
        (("AWS::EC2::SecurityGroup", "GroupId"), "AWS::EC2::SecurityGroup.Id"),
        (("AWS::EC2::SecurityGroup", "GroupName"), "AWS::EC2::SecurityGroup.Name"),
        (("AWS::EC2::VPC", "DefaultSecurityGroup"), "AWS::EC2::SecurityGroup.Id"),
        (("AWS::Logs::LogGroup", "Arn"), "AWS::Logs::LogGroup.Arn"),
    ]
    .into_iter()
    .collect();
    if let Some(resources) = input.get(FIELD_RESOURCES).and_then(|r| r.as_object()) {
        for (name, res) in resources {
            if let Some(edges) = res.get(FIELD_OUTGOING_REFS).and_then(|r| r.as_array()) {
                for edge in edges {
                    if edge.get(FIELD_KIND).and_then(|k| k.as_str()) != Some(EDGE_KIND_GET_ATT) {
                        continue;
                    }
                    let target = edge.get(FIELD_TARGET).and_then(|t| t.as_str()).unwrap_or("");
                    let attr = edge.get(FIELD_ATTR).and_then(|a| a.as_str()).unwrap_or("");
                    let sp = edge.get(FIELD_SOURCE_PATH).and_then(|p| p.as_str()).unwrap_or("");
                    if let Some(target_res) = m.resources.get(target) {
                        let src_fmt =
                            getatt_format.get(&(target_res.resource_type.as_str(), attr)).copied().unwrap_or("");
                        let dest_fmt = if sp.contains("GroupSet") {
                            "AWS::EC2::SecurityGroup.Id"
                        } else if sp.contains("awslogs-group") {
                            "AWS::Logs::LogGroup.Name"
                        } else {
                            ""
                        };
                        if !src_fmt.is_empty() && !dest_fmt.is_empty() && src_fmt != dest_fmt {
                            out.push(make_resource_diagnostic(
                                "E1040",
                                &format!(
                                    "{{'Fn::GetAtt': ['{}', '{}']}} does not match destination format of '{}'",
                                    target, attr, dest_fmt
                                ),
                                m,
                                name,
                                sp,
                                Some("Use the correct GetAtt attribute"),
                            ));
                        }
                    }
                }
            }
        }
    }

    let ref_type_ok: HashMap<(&str, &str), bool> = [
        (("AWS::EC2::VPC", "AWS::EC2::VPC.Id"), true),
        (("AWS::EC2::Subnet", "AWS::EC2::Subnet.Id"), true),
        (("AWS::EC2::SecurityGroup", "AWS::EC2::SecurityGroup.Id"), true),
        (("AWS::EC2::NetworkInterface", "AWS::EC2::NetworkInterface.Id"), true),
    ]
    .into_iter()
    .collect();
    if let Some(resources) = input.get(FIELD_RESOURCES).and_then(|r| r.as_object()) {
        for (name, res) in resources {
            if let Some(edges) = res.get(FIELD_OUTGOING_REFS).and_then(|r| r.as_array()) {
                for edge in edges {
                    if edge.get(FIELD_KIND).and_then(|k| k.as_str()) != Some(EDGE_KIND_REF) {
                        continue;
                    }
                    let target = edge.get(FIELD_TARGET).and_then(|t| t.as_str()).unwrap_or("");
                    let sp = edge.get(FIELD_SOURCE_PATH).and_then(|p| p.as_str()).unwrap_or("");
                    if let Some(target_res) = m.resources.get(target) {
                        let dest_fmt = if sp.ends_with("VpcId") {
                            "AWS::EC2::VPC.Id"
                        } else if sp.ends_with("SubnetId") {
                            "AWS::EC2::Subnet.Id"
                        } else if sp.ends_with("NetworkInterfaceId") {
                            "AWS::EC2::NetworkInterface.Id"
                        } else if sp.ends_with("SecurityGroupId") {
                            "AWS::EC2::SecurityGroup.Id"
                        } else {
                            ""
                        };
                        if !dest_fmt.is_empty()
                            && !ref_type_ok.contains_key(&(target_res.resource_type.as_str(), dest_fmt))
                        {
                            out.push(make_resource_diagnostic(
                                "E1041",
                                &format!("{{'Ref': '{}'}} does not match destination format of '{}'", target, dest_fmt),
                                m,
                                name,
                                sp,
                                Some("Use a Ref to a resource whose type matches the expected format"),
                            ));
                        }
                        // Non-VPC SecurityGroup Ref used where SecurityGroup.Id expected
                        if target_res.resource_type == "AWS::EC2::SecurityGroup"
                            && sp.ends_with("SecurityGroupId")
                            && !target_res.properties.contains_key("VpcId")
                        {
                            out.push(make_resource_diagnostic(
                                "E1041",
                                &format!(
                                    "{{'Ref': '{}'}} with formats ['AWS::EC2::SecurityGroup.Name'] does not match destination format of 'AWS::EC2::SecurityGroup.Id'",
                                    target
                                ),
                                m,
                                name,
                                sp,
                                Some("Use a Ref to a resource whose type matches the expected format"),
                            ));
                        }
                    }
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::CloudFront::Distribution") {
        let scenarios = m.resolve_scenarios_json(name, "Properties.DistributionConfig.DefaultCacheBehavior");
        for (val, conds) in &scenarios {
            if !conds.is_empty()
                && !m.conditions.is_satisfiable(&conds.iter().map(|(k, v)| (k.clone(), *v)).collect::<Vec<_>>())
            {
                continue;
            }
            if val.is_null() {
                out.push(make_resource_diagnostic(
                    "E3003",
                    "'DefaultCacheBehavior' is a required property",
                    m,
                    name,
                    "Properties.DistributionConfig",
                    Some("Add DefaultCacheBehavior to the Fn::If branch"),
                ));
            } else if val.is_object() {
                let path = "Properties.DistributionConfig.DefaultCacheBehavior.Fn::If.2";
                if val.get("TargetOriginId").is_none() {
                    out.push(make_resource_diagnostic(
                        "E3003",
                        "'TargetOriginId' is a required property",
                        m,
                        name,
                        path,
                        Some("Add TargetOriginId to the DefaultCacheBehavior"),
                    ));
                }
                if val.get("ViewerProtocolPolicy").is_none() {
                    out.push(make_resource_diagnostic(
                        "E3003",
                        "'ViewerProtocolPolicy' is a required property",
                        m,
                        name,
                        path,
                        Some("Add ViewerProtocolPolicy to the DefaultCacheBehavior"),
                    ));
                }
            }
        }
    }

    if let Some(resources) = input.get(FIELD_RESOURCES).and_then(|r| r.as_object()) {
        for (name, res) in resources {
            if let Some(tags) = res.get(FIELD_PROPERTIES).and_then(|p| p.get("Tags")).and_then(|t| t.as_array()) {
                for tag in tags {
                    if tag.is_object() && tag.get("Value").is_some() {
                        let key = tag.get("Key");
                        if key.is_none() || key == Some(&serde_json::Value::Null) {
                            out.push(make_resource_diagnostic(
                                "E3003",
                                "'Key' is a required property",
                                m,
                                name,
                                "Properties.Tags",
                                Some("Tag Key cannot be null or AWS::NoValue"),
                            ));
                            out.push(make_resource_diagnostic(
                                "E3024",
                                "'Key' is a required property",
                                m,
                                name,
                                "Properties.Tags",
                                Some("Tag Key cannot be null or AWS::NoValue"),
                            ));
                        }
                    }
                }
            }
        }
    }

    // emitted false positives when a conditional was nested inside a specific property
    // (e.g., Stack.Properties.Parameters = Fn::If). Nested-stack parameter validation
    // is covered by the dedicated schema/guard checks for AWS::CloudFormation::Stack.

    for svc_name in m.resources_of_type("AWS::ECS::Service") {
        let td_id = match m.follow_ref(svc_name, "Properties.TaskDefinition") {
            Some(id) => id.to_string(),
            None => continue,
        };
        // Use best-effort JSON conversion so partially-resolved (List, Map) values are
        // still traversable. resolve_concrete would reject anything that isn't a flat
        // Concrete serde value, which is the usual shape for multi-level resource props.
        let lbs = match m
            .resolve_deep(svc_name, "Properties.LoadBalancers")
            .or_else(|| m.resolve(svc_name, "Properties.LoadBalancers").cloned())
        {
            Some(rv) => match resolved_to_json_best_effort(&rv) {
                serde_json::Value::Array(a) => a,
                _ => continue,
            },
            None => continue,
        };
        let cdefs = match m
            .resolve_deep(&td_id, "Properties.ContainerDefinitions")
            .or_else(|| m.resolve(&td_id, "Properties.ContainerDefinitions").cloned())
        {
            Some(rv) => match resolved_to_json_best_effort(&rv) {
                serde_json::Value::Array(a) => a,
                _ => continue,
            },
            None => continue,
        };
        for (i, lb) in lbs.iter().enumerate() {
            let cn = lb.get("ContainerName").and_then(|v| v.as_str()).unwrap_or("");
            let cp = match lb.get("ContainerPort").and_then(coerce_port_to_string) {
                Some(port) => port,
                None => continue,
            };
            if cn.is_empty() {
                continue;
            }
            let has_dynamic = cdefs.iter().any(|c| {
                c.get("Name").and_then(|n| n.as_str()) == Some(cn)
                    && c.get("PortMappings")
                        .and_then(|p| p.as_array())
                        .map(|pms| {
                            pms.iter().any(|pm| {
                                pm.get("ContainerPort").and_then(coerce_port_to_string).as_deref() == Some(cp.as_str())
                                    && pm.get("HostPort").and_then(coerce_port_to_string).as_deref() == Some("0")
                            })
                        })
                        .unwrap_or(false)
            });
            if !has_dynamic {
                continue;
            }
            let tg_path = format!("Properties.LoadBalancers.{}.TargetGroupArn", i);
            if let Some(tg_id) = m.follow_ref(svc_name, &tg_path) {
                let present = m.resources.get(tg_id).is_some_and(|tg| tg.properties.contains_key("HealthCheckPort"));
                let wrong_ports: BTreeSet<String> = resolve_all_json(m, tg_id, "Properties.HealthCheckPort")
                    .iter()
                    .filter_map(health_check_port_scalar)
                    .filter(|p| p != "traffic-port")
                    .collect();
                let mut findings: Vec<(&str, String)> = wrong_ports
                    .into_iter()
                    .map(|p| {
                        (
                            "W3049",
                            format!(
                                "Container '{cn}' uses dynamic host port 0, so each task registers on an ephemeral port, but TargetGroup '{tg_id}' health-checks the fixed port '{p}'. The health check will not follow the traffic port unless '{p}' is separately served on every target. Use HealthCheckPort 'traffic-port' to health-check the port each target actually receives traffic on"
                            ),
                        )
                    })
                    .collect();
                if findings.is_empty() && !present {
                    findings.push((
                        "I3049",
                        format!(
                            "Container '{cn}' uses dynamic host port 0; TargetGroup '{tg_id}' omits HealthCheckPort, which defaults to 'traffic-port', so the health check follows each task's ephemeral port - the correct behavior for dynamic port mapping"
                        ),
                    ));
                }
                for (rule_id, message) in findings {
                    let mut diag =
                        make_resource_diagnostic(rule_id, &message, m, tg_id, "Properties.HealthCheckPort", None);
                    let svc_span = m.resource_span(svc_name, "Properties.LoadBalancers");
                    diag.related_resources.get_or_insert_with(Vec::new).push(RelatedResource {
                        resource: Some(ResourceRef {
                            id: Some(svc_name.to_string()),
                            resource_type: m.resources.get(svc_name).map(|r| r.resource_type.clone()),
                        }),
                        location: Some(SourceSpan {
                            start_line: svc_span.start_line,
                            start_column: svc_span.start_column,
                            end_line: svc_span.end_line,
                            end_column: svc_span.end_column,
                        }),
                        message: "Dynamic host port defined here".into(),
                    });
                    out.push(diag);
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::IAM::Policy") {
        let doc_rv = m
            .resolve_deep(name, "Properties.PolicyDocument")
            .or_else(|| m.resolve(name, "Properties.PolicyDocument").cloned());
        let Some(rv) = doc_rv else {
            continue;
        };
        // Preserve Fn::If as `{"Fn::If": [cond, then, else]}` (rather
        // than collapsing to one branch) so the action-format check
        // can consider every branch's ARN - a resource is acceptable
        // if any reachable branch matches the action.
        let doc = resolved_to_json_preserving_conditionals(&rv);
        check_iam_action_resources(&mut out, m, name, &doc, &ctx.cached_data.iam_action_resource_patterns);
    }

    for finding in subnets_outside_vpc(m) {
        out.push(make_resource_diagnostic(
            "E3059",
            &finding.message,
            m,
            &finding.subnet_id,
            "Properties.CidrBlock",
            None,
        ));
    }

    {
        let mut resource_types: Vec<&String> = m
            .resources_by_type
            .keys()
            .filter(|resource_type| ctx.cached_data.primary_identifiers.contains_key(*resource_type))
            .collect();
        resource_types.sort_unstable();
        for resource_type in resource_types {
            let identifier_properties = &ctx.cached_data.primary_identifiers[resource_type];
            let conflicts = m.primary_identifier_conflicts(resource_type, identifier_properties);
            for (tuple, resources) in &conflicts {
                let message =
                    primary_identifier_conflict_message(resource_type, identifier_properties, tuple, resources);
                let path = if identifier_properties.len() == 1 {
                    format!("Properties.{}", identifier_properties[0])
                } else {
                    KEY_PROPERTIES.to_string()
                };
                for resource_id in resources {
                    out.push(make_resource_diagnostic("E3019", &message, m, resource_id, &path, None));
                }
            }
        }
    }

    for (name, res) in &m.resources {
        for s in &res.diagnostics.unsubstituted_variables {
            out.push(make_resource_diagnostic(
                "E1029",
                &format!("Found an embedded parameter '{}' outside of an 'Fn::Sub' at {}", s.value, s.path),
                m,
                name,
                &s.path,
                Some("Wrap the string with Fn::Sub"),
            ));
        }
    }

    {
        let schema_metadata = ctx.cached_data.schema_metadata_catalog();
        for (name, res) in &m.resources {
            if let Some(type_meta) = schema_metadata.get(&res.resource_type) {
                for (prop, constraints) in &type_meta.property_constraints {
                    let is_string = type_meta.property_types.get(prop).map(|t| t == "string").unwrap_or(false);
                    let max_len = constraints.max_length.or_else(|| {
                        if is_string { constraints.maximum.as_ref().and_then(|v| v.as_u64()) } else { None }
                    });
                    let min_len = constraints.min_length.or_else(|| {
                        if is_string { constraints.minimum.as_ref().and_then(|v| v.as_u64()) } else { None }
                    });
                    if max_len.is_none() && min_len.is_none() {
                        continue;
                    }
                    let path = format!("Properties.{}", prop);
                    // Only reported when the constraint is broken whichever
                    // value the deployment picks: the shortest possibility
                    // still too long, or the longest still too short. A value
                    // the template states literally is checked against the
                    // constraint by schema validation instead, and a value with
                    // any unknown possibility yields no bounds at all.
                    if let Some((shortest, longest)) = m.estimated_string_length_bounds(name, &path) {
                        if let Some(max) = max_len
                            && shortest as u64 > max
                        {
                            out.push(make_resource_diagnostic(
                                "W9006",
                                &format!("String length {} exceeds maximum {} for property '{}'", shortest, max, prop),
                                m,
                                name,
                                &path,
                                None,
                            ));
                        }
                        if let Some(min) = min_len
                            && (longest as u64) < min
                        {
                            out.push(make_resource_diagnostic(
                                "W9006",
                                &format!("String length {} is below minimum {} for property '{}'", longest, min, prop),
                                m,
                                name,
                                &path,
                                None,
                            ));
                        }
                    }
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::Lambda::EventSourceMapping") {
        if let Some(target) = m.follow_ref(name, "Properties.EventSourceArn")
            && let Some(target_res) = m.resources.get(target as &str)
            && target_res.resource_type == "AWS::SQS::Queue"
        {
            let is_fifo = resolve_concrete(m, target, "Properties.FifoQueue")
                .is_some_and(|v| scalar_eq(&v, &serde_json::Value::Bool(true)));
            if is_fifo
                && let Some(batch) =
                    resolve_concrete(m, name, "Properties.BatchSize").as_ref().and_then(coerce_to_integer)
                && batch > 10
            {
                out.push(make_resource_diagnostic(
                    "E3705",
                    &format!("BatchSize {} exceeds maximum of 10 for SQS FIFO queue event source", batch),
                    m,
                    name,
                    "Properties.BatchSize",
                    None,
                ));
            }
        }
    }

    for name in m.resources_of_type("AWS::RDS::DBInstance") {
        if let Some(cluster_name) = m.follow_ref(name, "Properties.DBClusterIdentifier")
            && let (Some(serde_json::Value::String(inst_engine)), Some(serde_json::Value::String(cluster_engine))) =
                (resolve_concrete(m, name, "Properties.Engine"), resolve_concrete(m, cluster_name, "Properties.Engine"))
            && inst_engine != cluster_engine
        {
            let mut diag = make_resource_diagnostic(
                "E3707",
                &format!("DBInstance Engine '{}' does not match DBCluster Engine '{}'", inst_engine, cluster_engine),
                m,
                name,
                "Properties.Engine",
                None,
            );
            let cluster_span = m.resource_span(cluster_name, "Properties.Engine");
            diag.related_resources.get_or_insert_with(Vec::new).push(RelatedResource {
                resource: Some(ResourceRef {
                    id: Some(cluster_name.to_string()),
                    resource_type: m.resources.get(cluster_name).map(|r| r.resource_type.clone()),
                }),
                location: Some(SourceSpan {
                    start_line: cluster_span.start_line,
                    start_column: cluster_span.start_column,
                    end_line: cluster_span.end_line,
                    end_column: cluster_span.end_column,
                }),
                message: "cluster engine".into(),
            });
            out.push(diag);
        }
    }

    for name in m.resources_of_type("AWS::ApiGateway::Method") {
        if let Some(auth_id) = m.follow_ref(name, "Properties.AuthorizerId")
            && let (Some(serde_json::Value::String(auth_type)), Some(serde_json::Value::String(authorizer_type))) = (
                resolve_concrete(m, name, "Properties.AuthorizationType"),
                resolve_concrete(m, auth_id, "Properties.Type"),
            )
        {
            let expected = match auth_type.as_str() {
                "CUSTOM" => vec!["TOKEN", "REQUEST"],
                "COGNITO_USER_POOLS" => vec!["COGNITO_USER_POOLS"],
                _ => vec![],
            };
            if !expected.is_empty() && !expected.contains(&authorizer_type.as_str()) {
                out.push(make_resource_diagnostic(
                    "E3708",
                    &format!("'{}' is not one of {}", authorizer_type, render_str_list(&expected)),
                    m,
                    auth_id,
                    "Properties.Type",
                    None,
                ));
            }
        }
    }

    for name in m.resources_of_type("AWS::ApiGateway::Stage") {
        // The check applies only when the Stage references a Deployment that
        // resolves to a resource.
        let Some(deployment_name) = m.follow_ref(name, "Properties.DeploymentId") else {
            continue;
        };
        // A Deployment's RestApiId must match the Stage's RestApiId; a mismatch
        // fails deployment. Compare by identity first (a `Ref` and a `GetAtt` of
        // the same RestApi are equal), else by authored value. The finding is
        // anchored on the Deployment's RestApiId and renders the Stage's own
        // authored RestApiId as "<value> was expected".
        if rest_api_ids_conflict(m, name, deployment_name) {
            let expected = authored_form(m, name, "Properties.RestApiId")
                .map(|v| render_value(&v))
                .unwrap_or_else(|| "null".to_string());
            out.push(make_resource_diagnostic(
                "E3698",
                &format!("{} was expected", expected),
                m,
                deployment_name,
                "Properties.RestApiId",
                None,
            ));
        }
    }

    for name in m.resources_of_type("AWS::ApiGateway::Method") {
        // The check applies only when the Method references an Authorizer that
        // resolves to a resource (a Ref/GetAtt to an Authorizer) - a Method whose
        // AuthorizerId is a literal or opaque function has no in-template
        // Authorizer to compare against.
        let Some(authorizer_name) = m.follow_ref(name, "Properties.AuthorizerId") else {
            continue;
        };
        // A Method and the Authorizer it references must reference the same
        // RestApi; a mismatch fails deployment. Compare by identity: two values
        // that follow to the same resource match even when authored differently
        // (e.g. `Ref` vs `GetAtt` of the same RestApi resolve to one API), and
        // otherwise the raw authored values must be structurally equal.
        if rest_api_ids_conflict(m, name, authorizer_name) {
            // The finding is anchored on the Authorizer's RestApiId and renders
            // the Method's own authored RestApiId as "<value> was expected".
            let expected = authored_form(m, name, "Properties.RestApiId")
                .map(|v| render_value(&v))
                .unwrap_or_else(|| "null".to_string());
            out.push(make_resource_diagnostic(
                "E3699",
                &format!("{} was expected", expected),
                m,
                authorizer_name,
                "Properties.RestApiId",
                None,
            ));
        }
    }

    for name in m.resources_of_type("AWS::AutoScaling::AutoScalingGroup") {
        // MinSize/MaxSize are typically authored as strings ('10'), which
        // CloudFormation coerces to numbers; compare on the coerced integers so
        // string forms are checked too. Anchor the finding at MaxSize and render
        // it as the constraint that is violated (the max is below the minimum).
        let min_raw = resolve_concrete(m, name, "Properties.MinSize");
        let max_raw = resolve_concrete(m, name, "Properties.MaxSize");
        if let (Some(min_raw), Some(max_raw)) = (min_raw.as_ref(), max_raw.as_ref())
            && let (Some(min_val), Some(max_val)) = (coerce_to_integer(min_raw), coerce_to_integer(max_raw))
            && min_val > max_val
        {
            out.push(make_resource_diagnostic(
                "E3706",
                &format!("{} is less than the minimum of {}", render_value(max_raw), min_val),
                m,
                name,
                "Properties.MaxSize",
                None,
            ));
        }
    }

    for name in m.resources_of_type("AWS::ElasticLoadBalancingV2::Listener") {
        if let Some(serde_json::Value::String(proto)) = resolve_concrete(m, name, "Properties.Protocol")
            && ctx.cached_data.load_balancer_v2_certificate_protocols.contains(&proto)
            && !m.has_property(name, "Properties.Certificates")
        {
            out.push(make_resource_diagnostic(
                "E3676",
                &format!("{} listener requires Certificates", proto),
                m,
                name,
                "Properties.Certificates",
                None,
            ));
        }
    }

    for name in m.resources_of_type("AWS::ElasticLoadBalancing::LoadBalancer") {
        if let Some(listeners) = resolved_list_items(m, name, "Properties.Listeners") {
            for (i, listener) in listeners.iter().enumerate() {
                let proto = listener.get("Protocol").and_then(|p| p.as_str()).unwrap_or("");
                if ctx.cached_data.classic_load_balancer_certificate_protocols.contains(proto)
                    && listener.get("SSLCertificateId").is_none()
                {
                    out.push(make_resource_diagnostic(
                        "E3679",
                        &format!("{} listener requires SSLCertificateId", proto),
                        m,
                        name,
                        &format!("Properties.Listeners.{}", i),
                        None,
                    ));
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::Lambda::Function") {
        if let Some(serde_json::Value::Object(env_vars)) = resolve_concrete(m, name, "Properties.Environment.Variables")
        {
            for key in env_vars.keys() {
                if ctx.cached_data.lambda_reserved_environment_keys.contains(key) {
                    out.push(make_resource_diagnostic(
                        "E3663",
                        &format!("Environment variable '{}' is a Lambda reserved key", key),
                        m,
                        name,
                        "Properties.Environment.Variables",
                        None,
                    ));
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::Lambda::Function") {
        if resolve_concrete(m, name, "Properties.PackageType").as_ref().and_then(|v| v.as_str()) == Some("Image") {
            // A container-image function must not set Handler, Runtime, or Layers.
            // Report this once with a fixed message, anchored at the first
            // offending property present (schema order), no matter how many are
            // set - so collapse to a single diagnostic. Presence is keyed on the
            // property being SET (key-presence), not on it resolving to a concrete
            // value - so an excluded property whose value is an unresolved
            // Ref/intrinsic still anchors the finding.
            let props = m.resources.get(name.as_str()).map(|r| &r.properties);
            if let Some(first_excluded) = ctx
                .cached_data
                .lambda_image_excluded_properties
                .iter()
                .find(|excluded| props.map(|p| p.contains_key(excluded.as_str())).unwrap_or(false))
            {
                out.push(make_resource_diagnostic(
                    "E3685",
                    "Container image functions cannot specify Handler, Runtime, or Layers properties",
                    m,
                    name,
                    &format!("Properties.{}", first_excluded),
                    None,
                ));
            }
        }
    }

    for name in m.resources_of_type("AWS::ApiGateway::RestApi") {
        let has_body = m.has_property(name, "Properties.Body") || m.has_property(name, "Properties.BodyS3Location");
        if !has_body && !m.has_property(name, "Properties.Name") {
            out.push(make_resource_diagnostic(
                "E3660",
                "'Name' is required when 'Body' or 'BodyS3Location' is not provided",
                m,
                name,
                "Properties.Name",
                None,
            ));
        }
    }

    // E3671: an EBS block device mapping's Iops must satisfy the per-VolumeType
    // rules. io1/io2 require Iops and bound it (io1 100..=64000, io2
    // 100..=256000); gp3 does not require Iops but bounds it (3000..=16000).
    // This applies to the `BlockDeviceMappings[*].Ebs` blocks of launch
    // configurations, instances, launch templates, spot fleets, and OpsWorks
    // instances - NOT to a standalone AWS::EC2::Volume (whose Iops rules are
    // enforced by the resource schema directly).
    let emit_ebs_iops = |name: &str, bdm_path: &str, ebs: &serde_json::Value, out: &mut Vec<Diagnostic>| {
        let Some(vtype) = ebs.get("VolumeType").and_then(|v| v.as_str()) else {
            return;
        };
        // (required, min, max) per volume type; None => no Iops constraints.
        let bounds = match vtype {
            "io1" => Some((true, 100i64, 64000i64)),
            "io2" => Some((true, 100, 256000)),
            "gp3" => Some((false, 3000, 16000)),
            _ => None,
        };
        let Some((required, min, max)) = bounds else {
            return;
        };
        let iops_path = format!("{}.Ebs.Iops", bdm_path);
        // The numeric bounds only apply to a literal Iops. An Iops supplied via a
        // parameter Ref (whose default happens to be out of range) or another
        // intrinsic is not flagged - the parameter default is not folded into the
        // bound check. The required-when-absent check still holds regardless,
        // since absence is unambiguous.
        let iops_is_literal = !m.is_from_parameter(name, &iops_path) && !m.is_from_intrinsic(name, &iops_path);
        match ebs.get("Iops").and_then(coerce_to_integer) {
            None => {
                if required {
                    out.push(make_resource_diagnostic(
                        "E3671",
                        &format!("'Iops' is a required property when 'VolumeType' has a value of '{}'", vtype),
                        m,
                        name,
                        &iops_path,
                        None,
                    ));
                }
            }
            Some(iops) if iops_is_literal && iops < min => {
                out.push(make_resource_diagnostic(
                    "E3671",
                    &format!("{} is less than the minimum of {}", iops, min),
                    m,
                    name,
                    &iops_path,
                    None,
                ));
            }
            Some(iops) if iops_is_literal && iops > max => {
                out.push(make_resource_diagnostic(
                    "E3671",
                    &format!("{} is greater than the maximum of {}", iops, max),
                    m,
                    name,
                    &iops_path,
                    None,
                ));
            }
            Some(_) => {}
        }
    };
    for rpp in &ctx.cached_data.rule_tables.ebs_iops_property_paths {
        let wc_count = wildcard_count(rpp);
        if wc_count == 1 {
            // Single-wildcard: iterate over the array at the prefix path
            let prefix = property_path_prefix(rpp);
            for name in m.resources_of_type(&rpp.resource_type) {
                for (bdm_path, ebs) in ebs_block_device_mappings(m, name, &prefix) {
                    emit_ebs_iops(name, &bdm_path, &ebs, &mut out);
                }
            }
        } else if wc_count == 2 {
            // Double-wildcard (SpotFleet pattern): outer array then inner
            // BlockDeviceMappings. The prefix is the path up to the FIRST
            // wildcard; the middle segments are between the two wildcards.
            let Some(first_wc) = rpp.segments.iter().position(|s| matches!(s, PathSegment::Wildcard)) else {
                continue;
            };
            let outer_prefix: String = rpp.segments[..first_wc]
                .iter()
                .filter_map(|s| match s {
                    PathSegment::Literal(l) => Some(l.as_str()),
                    _ => None,
                })
                .collect::<Vec<_>>()
                .join(".");
            // Middle segments between first and second wildcard
            let Some(second_wc) = rpp.segments.iter().rposition(|s| matches!(s, PathSegment::Wildcard)) else {
                continue;
            };
            let middle_segments: String = rpp.segments[first_wc + 1..second_wc]
                .iter()
                .filter_map(|s| match s {
                    PathSegment::Literal(l) => Some(l.as_str()),
                    _ => None,
                })
                .collect::<Vec<_>>()
                .join(".");
            for name in m.resources_of_type(&rpp.resource_type) {
                if let Some(serde_json::Value::Array(specs)) = resolve_concrete(m, name, &outer_prefix) {
                    for (spec_idx, spec) in specs.iter().enumerate() {
                        let inner_val = if middle_segments.is_empty() {
                            Some(spec)
                        } else {
                            navigate_json_path(spec, &middle_segments)
                        };
                        let Some(serde_json::Value::Array(mappings)) = inner_val else {
                            continue;
                        };
                        for (idx, mapping) in mappings.iter().enumerate() {
                            if let Some(ebs) = mapping.get("Ebs") {
                                let path = format!("{}.{}.{}.{}", outer_prefix, spec_idx, middle_segments, idx);
                                emit_ebs_iops(name, &path, ebs, &mut out);
                            }
                        }
                    }
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::ElastiCache::ReplicationGroup") {
        if resolve_concrete(m, name, "Properties.Engine").as_ref().and_then(|v| v.as_str()) == Some("valkey")
            && !m.has_property(name, "Properties.TransitEncryptionEnabled")
        {
            out.push(make_resource_diagnostic(
                "E3704",
                "TransitEncryptionEnabled must be explicitly set when Engine is 'valkey'",
                m,
                name,
                "Properties.TransitEncryptionEnabled",
                None,
            ));
        }
    }

    // AZ_PATHS lists some resource types more than once (a type with multiple AZ
    // paths); `hardcoded_az::find` already scans all of a type's paths, so visit
    // each distinct type once to avoid emitting duplicate findings.
    let mut az_types_seen = HashSet::new();
    for (rtype, _) in hardcoded_az::AZ_PATHS {
        if !az_types_seen.insert(*rtype) {
            continue;
        }
        for name in m.resources_of_type(rtype) {
            for az in hardcoded_az::find(m, name, rtype) {
                out.push(make_resource_diagnostic("W3010", &hardcoded_az::message(&az.zone), m, name, &az.path, None));
            }
        }
    }

    // Instance type enum validation per region. With no region configured the
    // value is validated against the union of all regions (flagged only when
    // invalid everywhere); with a region set, that one region - see
    // `template_model::region_enums`.
    {
        let region = ctx.region.as_deref();
        let enum_checks: &[(&str, &str, &str, &str)] = &[
            ("E3628", "AWS::EC2::Instance", "Properties.InstanceType", "data/aws_ec2_instance_instancetype_enum"),
            (
                "E3641",
                "AWS::GameLift::Fleet",
                "Properties.EC2InstanceType",
                "data/aws_gamelift_fleet_ec2instancetype_enum",
            ),
            (
                "E3675",
                "AWS::EMR::InstanceTypeConfig",
                "Properties.InstanceType",
                "data/aws_emr_cluster_instancetypeconfig_instancetype_enum",
            ),
            (
                "E3617",
                "AWS::ManagedBlockchain::Node",
                "Properties.NodeConfiguration.InstanceType",
                "data/aws_managedblockchain_node_nodeconfiguration_instancetype_enum",
            ),
            (
                "E3620",
                "AWS::DocDB::DBInstance",
                "Properties.DBInstanceClass",
                "data/aws_docdb_dbinstance_dbinstanceclass_enum",
            ),
            ("E3621", "AWS::AppStream::Fleet", "Properties.InstanceType", "data/aws_appstream_fleet_instancetype_enum"),
            (
                "E3647",
                "AWS::ElastiCache::CacheCluster",
                "Properties.CacheNodeType",
                "data/aws_elasticache_cachecluster_cachenodetype_enum",
            ),
            ("E3672", "AWS::DAX::Cluster", "Properties.NodeType", "data/aws_dax_cluster_nodetype_enum"),
            (
                "E3635",
                "AWS::Neptune::DBInstance",
                "Properties.DBInstanceClass",
                "data/aws_neptune_dbinstance_dbinstanceclass_enum",
            ),
            ("E3667", "AWS::Redshift::Cluster", "Properties.NodeType", "data/aws_redshift_cluster_nodetype_enum"),
            (
                "E3640",
                "AWS::SageMaker::DataQualityJobDefinition",
                "Properties.JobResources.ClusterConfig.InstanceType",
                "data/aws_sagemaker_processing_instancetype_enum",
            ),
            (
                "E3640",
                "AWS::SageMaker::ModelBiasJobDefinition",
                "Properties.JobResources.ClusterConfig.InstanceType",
                "data/aws_sagemaker_processing_instancetype_enum",
            ),
            (
                "E3640",
                "AWS::SageMaker::ModelExplainabilityJobDefinition",
                "Properties.JobResources.ClusterConfig.InstanceType",
                "data/aws_sagemaker_processing_instancetype_enum",
            ),
            (
                "E3640",
                "AWS::SageMaker::ModelQualityJobDefinition",
                "Properties.JobResources.ClusterConfig.InstanceType",
                "data/aws_sagemaker_processing_instancetype_enum",
            ),
            (
                "E3640",
                "AWS::SageMaker::MonitoringSchedule",
                "Properties.MonitoringScheduleConfig.MonitoringJobDefinition.MonitoringResources.ClusterConfig.InstanceType",
                "data/aws_sagemaker_processing_instancetype_enum",
            ),
            (
                "E3652",
                "AWS::Elasticsearch::Domain",
                "Properties.ElasticsearchClusterConfig.InstanceType",
                "data/aws_elasticsearch_domain_elasticsearchclusterconfig_instancetype_enum",
            ),
            (
                "E3653",
                "AWS::OpenSearchService::Domain",
                "Properties.ClusterConfig.InstanceType",
                "data/aws_opensearchservice_domain_clusterconfig_instancetype_enum",
            ),
        ];
        for &(rule_id, rtype, prop_path, enum_key) in enum_checks {
            let resources = m.resources_of_type(rtype);
            if resources.is_empty() {
                continue;
            }
            let Some(allowed) = region_flat_allowed(&ctx.cached_data.enum_data, enum_key, region) else {
                continue;
            };
            for name in resources {
                if let Some(val) = resolve_enum_string(m, name, prop_path)
                    && !allowed.contains(val.as_str())
                {
                    out.push(make_resource_diagnostic(
                        rule_id,
                        &region_enums::flat_invalid_message(&val, region),
                        m,
                        name,
                        prop_path,
                        None,
                    ));
                }
            }
        }

        let wildcard_enum_checks: &[(&str, &str, &str, &str, &str)] = &[
            (
                "E3642",
                "AWS::SageMaker::InferenceExperiment",
                "Properties.ModelVariants.{}.InfrastructureConfig.RealTimeInferenceConfig.InstanceType",
                "Properties.ModelVariants.InfrastructureConfig.RealTimeInferenceConfig.InstanceType",
                "data/aws_sagemaker_hosting_instancetype_enum",
            ),
            (
                "E3643",
                "AWS::SageMaker::ModelPackage",
                "Properties.ValidationSpecification.ValidationProfiles.{}.TransformJobDefinition.TransformResources.InstanceType",
                "Properties.ValidationSpecification.ValidationProfiles.TransformJobDefinition.TransformResources.InstanceType",
                "data/aws_sagemaker_transform_instancetype_enum",
            ),
            (
                "E3644",
                "AWS::SageMaker::Cluster",
                "Properties.InstanceGroups.{}.InstanceType",
                "Properties.InstanceGroups.InstanceType",
                "data/aws_sagemaker_cluster_instancetype_enum",
            ),
            (
                "E3644",
                "AWS::SageMaker::Cluster",
                "Properties.RestrictedInstanceGroups.{}.InstanceType",
                "Properties.RestrictedInstanceGroups.InstanceType",
                "data/aws_sagemaker_cluster_instancetype_enum",
            ),
        ];
        for &(rule_id, rtype, wildcard_path, report_path, enum_key) in wildcard_enum_checks {
            let resources = m.resources_of_type(rtype);
            if resources.is_empty() {
                continue;
            }
            let Some(allowed) = region_flat_allowed(&ctx.cached_data.enum_data, enum_key, region) else {
                continue;
            };
            for name in resources {
                let mut reported = HashSet::new();
                for val in resolve_concrete_strings(m, name, wildcard_path) {
                    if allowed.contains(val.as_str()) || !reported.insert(val.clone()) {
                        continue;
                    }
                    out.push(make_resource_diagnostic(
                        rule_id,
                        &region_enums::flat_invalid_message(&val, region),
                        m,
                        name,
                        report_path,
                        None,
                    ));
                }
            }
        }

        if let Some(region_map) =
            region_map_for_key(&ctx.cached_data.enum_data, "data/aws_rds_dbinstance_dbinstanceclass_enum")
        {
            // The RDS document is a conditional schema: each allOf branch's
            // `if.required` consts (Engine/LicenseModel) select which
            // `then.enum` of instance classes applies. Every branch whose consts
            // match applies, so the class must be in ALL matching branches'
            // enums (the intersection); a dynamic or unmatched Engine leaves no
            // matching branch and is not validated. The Engine value is matched
            // case-insensitively for DBInstance (the Engine value is lowercased
            // before evaluating this schema). With no region configured the class
            // is validated against the union of all regions.
            for name in m.resources_of_type("AWS::RDS::DBInstance") {
                let Some(val) = resolve_enum_string(m, name, "Properties.DBInstanceClass") else {
                    continue;
                };
                if let Some(mismatch) =
                    region_enums::conditional_enum_mismatch(region_map, region, "DBInstanceClass", true, &val, |prop| {
                        resolve_enum_string(m, name, &format!("Properties.{}", prop))
                    })
                {
                    out.push(make_resource_diagnostic(
                        "E3025",
                        &region_enums::conditional_mismatch_message(&val, &mismatch, region),
                        m,
                        name,
                        "Properties.DBInstanceClass",
                        None,
                    ));
                }
            }
        }

        // RDS DBCluster DBClusterInstanceClass uses a conditional schema keyed
        // on Engine, but Engine is not lowercased for DBCluster, so match the
        // const case-sensitively.
        if let Some(region_map) =
            region_map_for_key(&ctx.cached_data.enum_data, "data/aws_rds_dbcluster_dbclusterinstanceclass_enum")
        {
            for name in m.resources_of_type("AWS::RDS::DBCluster") {
                let Some(val) = resolve_enum_string(m, name, "Properties.DBClusterInstanceClass") else {
                    continue;
                };
                if let Some(mismatch) = region_enums::conditional_enum_mismatch(
                    region_map,
                    region,
                    "DBClusterInstanceClass",
                    false,
                    &val,
                    |prop| resolve_enum_string(m, name, &format!("Properties.{}", prop)),
                ) {
                    out.push(make_resource_diagnostic(
                        "E3694",
                        &region_enums::conditional_mismatch_message(&val, &mismatch, region),
                        m,
                        name,
                        "Properties.DBClusterInstanceClass",
                        None,
                    ));
                }
            }
        }

        let amazonmq_brokers = m.resources_of_type("AWS::AmazonMQ::Broker");
        if !amazonmq_brokers.is_empty()
            && let Some(allowed) =
                region_flat_allowed(&ctx.cached_data.enum_data, "data/aws_amazonmq_broker_instancetype_enum", region)
        {
            for name in amazonmq_brokers {
                if let Some(val) = resolve_enum_string(m, name, "Properties.HostInstanceType")
                    && !allowed.contains(val.as_str())
                {
                    out.push(make_resource_diagnostic(
                        "E3670",
                        &region_enums::flat_invalid_message(&val, region),
                        m,
                        name,
                        "Properties.HostInstanceType",
                        None,
                    ));
                }
            }
        }

        let emr_instance_fleets = m.resources_of_type("AWS::EMR::InstanceFleetConfig");
        if !emr_instance_fleets.is_empty()
            && let Some(allowed) = region_flat_allowed(
                &ctx.cached_data.enum_data,
                "data/aws_emr_cluster_instancetypeconfig_instancetype_enum",
                region,
            )
        {
            for name in emr_instance_fleets {
                if let Some(val) = resolve_enum_string(m, name, "Properties.InstanceType")
                    && !allowed.contains(val.as_str())
                {
                    out.push(make_resource_diagnostic(
                        "E3675",
                        &region_enums::flat_invalid_message(&val, region),
                        m,
                        name,
                        "Properties.InstanceType",
                        None,
                    ));
                }
            }
        }
    }

    for name in m.resources_of_type("AWS::ElasticLoadBalancingV2::LoadBalancer") {
        if !alb_is_application_type(m, name) {
            continue;
        }
        for (subnet_property, message) in ALB_SUBNET_MINIMUM_MESSAGES {
            let path = format!("Properties.{}", subnet_property);
            if resolved_item_count(m, name, &path).is_some_and(|count| count < ALB_MINIMUM_SUBNETS) {
                out.push(make_resource_diagnostic("E3680", message, m, name, &path, None));
            }
        }
    }

    for name in m.resources_of_type("AWS::RDS::DBCluster") {
        if let Some(path) = dbcluster_monitoring_mismatch_path(m, name) {
            out.push(make_resource_diagnostic("E3689", DBCLUSTER_MONITORING_MESSAGE, m, name, path, None));
        }
    }

    for name in m.resources_of_type("AWS::ApiGateway::Stage") {
        let Some(settings) = resolved_list_items(m, name, "Properties.MethodSettings") else {
            continue;
        };
        for (idx, setting) in settings.iter().enumerate() {
            if !stage_method_setting_configures_something(setting) {
                continue;
            }
            if let Some(resource_path) = setting.get("ResourcePath").and_then(|v| v.as_str())
                && !resource_path.starts_with('/')
            {
                out.push(make_resource_diagnostic(
                    "E3723",
                    &format!(
                        "ResourcePath '{}' must start with '/' when a method setting is configured",
                        resource_path
                    ),
                    m,
                    name,
                    &format!("Properties.MethodSettings.{}.ResourcePath", idx),
                    None,
                ));
            }
        }
    }

    for name in m.resources_of_type("AWS::Route53::RecordSet") {
        if let (Some(serde_json::Value::String(rec_name)), Some(serde_json::Value::String(hz_name))) =
            (resolve_concrete(m, name, "Properties.Name"), resolve_concrete(m, name, "Properties.HostedZoneName"))
        {
            let trimmed_rec = rec_name.trim_end_matches('.');
            let trimmed_hz = hz_name.trim_end_matches('.');
            if trimmed_rec != trimmed_hz && !rec_name.ends_with(&*hz_name) && !trimmed_rec.ends_with(trimmed_hz) {
                out.push(make_resource_diagnostic(
                    "E3041",
                    &format!("RecordSet Name '{}' is not a subdomain of HostedZoneName '{}'", rec_name, hz_name),
                    m,
                    name,
                    "Properties.Name",
                    None,
                ));
            }
        }
    }

    for name in m.resources_of_type("AWS::Backup::BackupPlan") {
        if let Some(serde_json::Value::Array(rules)) = resolve_concrete(m, name, "Properties.BackupPlan.BackupPlanRule")
        {
            for rule in &rules {
                if let Some(lifecycle) = rule.get("Lifecycle") {
                    let move_days = lifecycle.get("MoveToColdStorageAfterDays").and_then(|v| v.as_i64());
                    let delete_days = lifecycle.get("DeleteAfterDays").and_then(|v| v.as_i64());
                    if let (Some(m_d), Some(d_d)) = (move_days, delete_days)
                        && m_d >= d_d
                    {
                        out.push(make_resource_diagnostic(
                            "E3504",
                            &format!(
                                "MoveToColdStorageAfterDays ({}) must be less than DeleteAfterDays ({})",
                                m_d, d_d
                            ),
                            m,
                            name,
                            "Properties.BackupPlanRule",
                            None,
                        ));
                    }
                }
            }
        }
    }

    // IAM ManagedPolicy - Statement should have Resource when Action is present
    for name in m.resources_of_type("AWS::IAM::ManagedPolicy") {
        if let Some(doc) = resolve_concrete(m, name, "Properties.PolicyDocument")
            && let Some(stmts) = doc.get("Statement").and_then(|s| s.as_array())
        {
            for stmt in stmts {
                if !stmt.is_object() {
                    continue;
                }
                if stmt.get("Action").is_some() && stmt.get("Resource").is_none() && stmt.get("NotResource").is_none() {
                    out.push(make_resource_diagnostic(
                        "W3037",
                        "IAM policy statement has Action but no Resource",
                        m,
                        name,
                        "Properties.PolicyDocument",
                        None,
                    ));
                }
            }
        }
    }

    // A mis-cased property name is already reported as an unknown property
    // (F3002, "Additional properties are not allowed"). A separate casing
    // diagnostic under E3011 would be a false positive, since E3011 enforces a
    // name-length limit rather than a casing check, so it is not emitted.

    // Route53 RecordSet validation
    for name in m.resources_of_type("AWS::Route53::RecordSet") {
        for (properties, conditions) in m.resolve_properties_scenarios(name) {
            if !scenario_is_reachable(m, name, &conditions) {
                continue;
            }
            let Some(serde_json::Value::String(record_type)) = scenario_property_json(&properties, "Type") else {
                continue;
            };
            if record_type == "CNAME"
                && let (Some(serde_json::Value::String(record_name)), Some(serde_json::Value::String(hosted_zone_name))) =
                    (scenario_property_json(&properties, "Name"), scenario_property_json(&properties, "HostedZoneName"))
                && record_name.trim_end_matches('.') == hosted_zone_name.trim_end_matches('.')
            {
                let property_path = "Properties.Name";
                let source_path = route53_scenario_source_path(m, name, property_path, &conditions);
                out.push(make_resource_diagnostic_at_source(
                    "E3023",
                    &format!(
                        "CNAME record Name '{}' must not match HostedZoneName '{}' exactly",
                        record_name, hosted_zone_name
                    ),
                    m,
                    name,
                    property_path,
                    &source_path,
                    None,
                ));
            }
            if let Some(serde_json::Value::Array(records)) = scenario_property_json(&properties, "ResourceRecords") {
                push_route53_record_value_diagnostics(
                    &mut out,
                    m,
                    name,
                    &record_type,
                    &records,
                    "Properties.ResourceRecords",
                    |property_path| route53_scenario_source_path(m, name, property_path, &conditions),
                );
            }
        }
    }

    // RecordSetGroup - validate records within RecordSets[]
    for name in m.resources_of_type("AWS::Route53::RecordSetGroup") {
        for (properties, conditions) in m.resolve_properties_scenarios(name) {
            if !scenario_is_reachable(m, name, &conditions) {
                continue;
            }
            let Some(serde_json::Value::Array(record_sets)) = scenario_property_json(&properties, "RecordSets") else {
                continue;
            };
            for (set_index, record_set) in record_sets.iter().enumerate() {
                let Some(record_type) = record_set.get("Type").and_then(|value| value.as_str()) else {
                    continue;
                };
                if let Some(records) = record_set.get("ResourceRecords").and_then(|value| value.as_array()) {
                    push_route53_record_value_diagnostics(
                        &mut out,
                        m,
                        name,
                        record_type,
                        records,
                        &format!("Properties.RecordSets.{}.ResourceRecords", set_index),
                        |property_path| route53_scenario_source_path(m, name, property_path, &conditions),
                    );
                }
            }
        }
    }

    // ElastiCache Redis replication group failover
    for name in m.resources_of_type("AWS::ElastiCache::ReplicationGroup") {
        if resolve_concrete(m, name, "Properties.Engine").as_ref().and_then(|v| v.as_str()) == Some("redis") {
            // NumCacheClusters is ignored when NumNodeGroups is specified
            if m.has_property(name, "Properties.NumNodeGroups") {
                continue;
            }
            if let Some(num) = resolve_concrete(m, name, "Properties.NumCacheClusters").and_then(|v| v.as_i64())
                && num > 1
            {
                let failover_rv = m
                    .resolve_deep(name, "Properties.AutomaticFailoverEnabled")
                    .or_else(|| m.resolve(name, "Properties.AutomaticFailoverEnabled").cloned());
                let is_definitely_not_true = match &failover_rv {
                    None => true,
                    Some(ResolvedValue::Concrete { value: v }) => v.as_bool() != Some(true),
                    _ => false,
                };
                if is_definitely_not_true {
                    out.push(make_resource_diagnostic(
                        "E3026",
                        "AutomaticFailoverEnabled must be true when NumCacheClusters > 1 and Engine is 'redis'",
                        m,
                        name,
                        "Properties.AutomaticFailoverEnabled",
                        None,
                    ));
                }
            }
        }
    }

    // Events Rule ScheduleExpression validation: a value that is invalid service-side (e.g. a plural
    // unit with value 1, a non-integer value, or a cron pinning both day fields) is caught here. A
    // value built from a parameter or intrinsic is only known at deploy time, so it is skipped -
    // validating the pre-resolution literal would flag a value the service never sees.
    for name in m.resources_of_type("AWS::Events::Rule") {
        let path = "Properties.ScheduleExpression";
        if m.is_from_parameter(name, path) || m.is_from_intrinsic(name, path) {
            continue;
        }
        if let Some(serde_json::Value::String(val)) = resolve_concrete(m, name, path) {
            for message in schedule_expression_errors(&val) {
                out.push(make_resource_diagnostic("E3027", &message, m, name, path, None));
            }
        }
    }

    // Route53 RecordSet Alias validation
    for name in m.resources_of_type("AWS::Route53::RecordSet") {
        let property_scenarios = m.resolve_properties_scenarios(name);
        let has_alias_ttl_conflict = property_scenarios.iter().any(|(properties, conditions)| {
            scenario_is_reachable(m, name, conditions)
                && scenario_has_effective_property(properties, "AliasTarget")
                && scenario_has_effective_property(properties, "TTL")
        });
        if has_alias_ttl_conflict {
            out.push(make_resource_diagnostic(
                "E3029",
                "TTL must not be set when AliasTarget is specified",
                m,
                name,
                "Properties.TTL",
                None,
            ));
        }

        let invalid_alias_types: BTreeSet<String> = property_scenarios
            .iter()
            .filter_map(|(properties, conditions)| {
                if !scenario_is_reachable(m, name, conditions)
                    || !scenario_has_effective_property(properties, "AliasTarget")
                {
                    return None;
                }
                let record_type = scenario_property_json(properties, "Type")?;
                let record_type = record_type.as_str()?;
                matches!(record_type, "NS" | "SOA").then(|| record_type.to_string())
            })
            .collect();
        for record_type in invalid_alias_types {
            out.push(make_resource_diagnostic(
                "E3029",
                &format!("AliasTarget cannot be used with record type '{}'", record_type),
                m,
                name,
                "Properties.AliasTarget",
                None,
            ));
        }
    }

    // RDS DBInstance class validation by Engine/EngineVersion
    if let Some(rds_data) = ctx
        .cached_data
        .enum_data
        .get("data/aws_rds_dbinstance_db_instance_class")
        .and_then(|v| v.as_object())
        .and_then(|o| o.values().next())
        .and_then(|v| v.get("allOf"))
        .and_then(|v| v.as_array())
    {
        for name in m.resources_of_type("AWS::RDS::DBInstance") {
            let engine = resolve_concrete(m, name, "Properties.Engine")
                .and_then(|v| if let serde_json::Value::String(s) = v { Some(s) } else { None });
            let engine_ver = resolve_concrete(m, name, "Properties.EngineVersion").as_ref().and_then(coerce_to_string);
            let db_class = resolve_concrete(m, name, "Properties.DBInstanceClass")
                .and_then(|v| if let serde_json::Value::String(s) = v { Some(s) } else { None });
            if let (Some(eng), Some(ver), Some(cls)) = (engine, engine_ver, db_class) {
                for entry in rds_data {
                    let cond = match entry.get("if").and_then(|v| v.get("properties")) {
                        Some(p) => p,
                        None => continue,
                    };
                    let eng_match = cond
                        .get("Engine")
                        .and_then(|e| e.get("const"))
                        .and_then(|c| c.as_str())
                        .map(|c| c == eng)
                        .unwrap_or(false);
                    if !eng_match {
                        continue;
                    }
                    let ver_match = cond
                        .get("EngineVersion")
                        .and_then(|e| e.get("pattern"))
                        .and_then(|p| p.as_str())
                        .and_then(|p| regex::Regex::new(p).ok())
                        .map(|re| re.is_match(&ver))
                        .unwrap_or(false);
                    if !ver_match {
                        continue;
                    }
                    if let Some(allowed) = entry
                        .get("then")
                        .and_then(|t| t.get("properties"))
                        .and_then(|p| p.get("DBInstanceClass"))
                        .and_then(|d| d.get("enum"))
                        .and_then(|e| e.as_array())
                    {
                        let allowed_set: HashSet<&str> = allowed.iter().filter_map(|v| v.as_str()).collect();
                        if !allowed_set.contains(cls.as_str()) {
                            out.push(make_resource_diagnostic(
                                "E3062",
                                &format!(
                                    "DBInstanceClass '{}' is not valid for Engine '{}' EngineVersion '{}'",
                                    cls, eng, ver
                                ),
                                m,
                                name,
                                "Properties.DBInstanceClass",
                                None,
                            ));
                        }
                    }
                    break;
                }
            }
        }
    }

    // Warns on properties that only work with `aws cloudformation package`. The
    // parent property (e.g. Code, Content, TemplateURL) is checked as a string;
    // if its value is a string not starting with s3:// or https://, it warns.
    // SAM templates are excluded entirely.
    if !m.transforms.iter().any(|t| t == TRANSFORM_SERVERLESS) {
        for rpp in &ctx.cached_data.rule_tables.package_property_paths {
            let path = full_property_path(rpp);
            for name in m.resources_of_type(&rpp.resource_type) {
                // Only string literals are inspected here; a value wrapped in
                // an intrinsic (Fn::Join/Fn::Sub building an S3 URL) resolves
                // at deploy time and is left alone, so skip intrinsic-sourced
                // values.
                if m.is_from_intrinsic(name, &path) {
                    continue;
                }
                if let Some(serde_json::Value::String(val)) = resolve_concrete(m, name, &path) {
                    if val.starts_with("s3://") || val.starts_with("https://") {
                        continue;
                    }
                    out.push(make_resource_diagnostic(
                        "W3002",
                        "This code may only work with 'package' cli command",
                        m,
                        name,
                        &path,
                        None,
                    ));
                }
            }
        }
    }

    // API Gateway mixing inline definitions with related resources
    {
        let apigw_resource_types = &ctx.cached_data.rule_tables.api_gateway_mixing_resource_types;
        let mut rest_api_refs: HashMap<String, Vec<String>> = HashMap::new();
        for rtype in apigw_resource_types {
            for name in m.resources_of_type(rtype) {
                if let Some(api_id) = m.follow_ref(name, "Properties.RestApiId") {
                    rest_api_refs.entry(api_id.to_string()).or_default().push(name.to_string());
                }
            }
        }
        for (api_id, referrers) in &rest_api_refs {
            let Some(api) = m.resources.get(api_id) else {
                continue;
            };
            for property in ["Body", "BodyS3Location"] {
                if !api.properties.contains_key(property) {
                    continue;
                }
                for referrer in referrers {
                    let resource_type = &m.resources[referrer].resource_type;
                    out.push(make_resource_diagnostic(
                        "W3660",
                        &format!(
                            "Defining '{property}' with a relation to resource '{referrer}' of type '{resource_type}' may result in drift and orphaned resources"
                        ),
                        m,
                        api_id,
                        &format!("Properties.{property}"),
                        None,
                    ));
                }
            }
        }
    }

    // Lambda Permission Principal/SourceArn consistency
    for name in m.resources_of_type("AWS::Lambda::Permission") {
        if let Some(serde_json::Value::String(principal)) = resolve_concrete(m, name, "Properties.Principal")
            && let Some(target) = m.follow_ref(name, "Properties.SourceArn")
            && let Some(target_res) = m.resources.get(target)
        {
            match principal.as_str() {
                "sns.amazonaws.com" if target_res.resource_type != "AWS::SNS::Topic" => {
                    out.push(make_resource_diagnostic(
                                "W3664",
                                &format!(
                                    "SourceArn references '{}' (type '{}') but Principal 'sns.amazonaws.com' expects an SNS Topic",
                                    target, target_res.resource_type
                                ),
                                m, name, "Properties.SourceArn", None,
                            ));
                }
                "s3.amazonaws.com" if target_res.resource_type != "AWS::S3::Bucket" => {
                    out.push(make_resource_diagnostic(
                                "W3664",
                                &format!(
                                    "SourceArn references '{}' (type '{}') but Principal 's3.amazonaws.com' expects an S3 Bucket",
                                    target, target_res.resource_type
                                ),
                                m, name, "Properties.SourceArn", None,
                            ));
                }
                _ => {}
            }
        }
    }

    // EBS Iops silently ignored for certain volume types
    {
        let iops_ignored_types = &ctx.cached_data.rule_tables.ebs_iops_ignored_volume_types;
        let iops_ignored_refs: Vec<&str> = iops_ignored_types.iter().map(|s| s.as_str()).collect();
        for rpp in &ctx.cached_data.rule_tables.ebs_iops_property_paths {
            let wc_count = wildcard_count(rpp);
            if wc_count == 1 {
                let prefix = property_path_prefix(rpp);
                for name in m.resources_of_type(&rpp.resource_type) {
                    if let Some(serde_json::Value::Array(bdms)) = resolve_concrete(m, name, &prefix) {
                        check_bdm_iops_ignored(&mut out, m, name, &bdms, &prefix, "W3671", &iops_ignored_refs);
                    }
                }
            } else if wc_count == 2 {
                // SpotFleet-style double-wildcard
                let Some(first_wc) = rpp.segments.iter().position(|s| matches!(s, PathSegment::Wildcard)) else {
                    continue;
                };
                let outer_prefix: String = rpp.segments[..first_wc]
                    .iter()
                    .filter_map(|s| match s {
                        PathSegment::Literal(l) => Some(l.as_str()),
                        _ => None,
                    })
                    .collect::<Vec<_>>()
                    .join(".");
                let Some(second_wc) = rpp.segments.iter().rposition(|s| matches!(s, PathSegment::Wildcard)) else {
                    continue;
                };
                let middle_segments: String = rpp.segments[first_wc + 1..second_wc]
                    .iter()
                    .filter_map(|s| match s {
                        PathSegment::Literal(l) => Some(l.as_str()),
                        _ => None,
                    })
                    .collect::<Vec<_>>()
                    .join(".");
                for name in m.resources_of_type(&rpp.resource_type) {
                    if let Some(serde_json::Value::Array(specs)) = resolve_concrete(m, name, &outer_prefix) {
                        for (si, spec) in specs.iter().enumerate() {
                            let bdms_val = if middle_segments.is_empty() {
                                Some(spec)
                            } else {
                                navigate_json_path(spec, &middle_segments)
                            };
                            if let Some(serde_json::Value::Array(bdms)) = bdms_val {
                                let base = format!("{}.{}.{}", outer_prefix, si, middle_segments);
                                check_bdm_iops_ignored(&mut out, m, name, bdms, &base, "W3671", &iops_ignored_refs);
                            }
                        }
                    }
                }
            }
        }
    }

    // RDS DBCluster - SnapshotIdentifier makes MasterUsername ignored
    for name in m.resources_of_type("AWS::RDS::DBCluster") {
        if m.has_property(name, "Properties.SnapshotIdentifier") && m.has_property(name, "Properties.MasterUsername") {
            out.push(make_resource_diagnostic(
                "W3688",
                "MasterUsername is ignored when SnapshotIdentifier is present",
                m,
                name,
                "Properties.MasterUsername",
                None,
            ));
        }
    }

    // RDS DBCluster - SourceDBClusterIdentifier makes several properties ignored
    for name in m.resources_of_type("AWS::RDS::DBCluster") {
        if m.has_property(name, "Properties.SourceDBClusterIdentifier") {
            for ignored in &["MasterUserPassword", "MasterUsername", "StorageEncrypted"] {
                if m.has_property(name, &format!("Properties.{}", ignored)) {
                    out.push(make_resource_diagnostic(
                        "W3689",
                        &format!("'{}' is ignored when SourceDBClusterIdentifier is present", ignored),
                        m,
                        name,
                        &format!("Properties.{}", ignored),
                        None,
                    ));
                }
            }
        }
    }

    // RDS DBCluster - Aurora serverless ignores PerformanceInsights properties
    for name in m.resources_of_type("AWS::RDS::DBCluster") {
        let engine = resolve_concrete(m, name, "Properties.Engine")
            .and_then(|v| if let serde_json::Value::String(s) = v { Some(s) } else { None });
        let engine_mode = resolve_concrete(m, name, "Properties.EngineMode")
            .and_then(|v| if let serde_json::Value::String(s) = v { Some(s) } else { None });
        if let (Some(eng), Some(mode)) = (engine, engine_mode)
            && (eng == "aurora-mysql" || eng == "aurora-postgresql")
            && mode == "serverless"
        {
            for ignored in
                &["PerformanceInsightsEnabled", "PerformanceInsightsKmsKeyId", "PerformanceInsightsRetentionPeriod"]
            {
                if m.has_property(name, &format!("Properties.{}", ignored)) {
                    out.push(make_resource_diagnostic(
                        "W3693",
                        &format!("'{}' is ignored when EngineMode is 'serverless'", ignored),
                        m,
                        name,
                        &format!("Properties.{}", ignored),
                        None,
                    ));
                }
            }
        }
    }

    // SNS Subscription Protocol/Endpoint consistency
    for name in m.resources_of_type("AWS::SNS::Subscription") {
        if let Some(serde_json::Value::String(protocol)) = resolve_concrete(m, name, "Properties.Protocol")
            && let Some(target) = m.follow_ref(name, "Properties.Endpoint")
            && let Some(target_res) = m.resources.get(target)
        {
            match protocol.as_str() {
                "sqs" if target_res.resource_type != "AWS::SQS::Queue" => {
                    out.push(make_resource_diagnostic(
                        "W3694",
                        &format!(
                            "Endpoint references '{}' (type '{}') but Protocol 'sqs' expects an SQS Queue",
                            target, target_res.resource_type
                        ),
                        m,
                        name,
                        "Properties.Endpoint",
                        None,
                    ));
                }
                "lambda" if target_res.resource_type != "AWS::Lambda::Function" => {
                    out.push(make_resource_diagnostic(
                        "W3694",
                        &format!(
                            "Endpoint references '{}' (type '{}') but Protocol 'lambda' expects a Lambda Function",
                            target, target_res.resource_type
                        ),
                        m,
                        name,
                        "Properties.Endpoint",
                        None,
                    ));
                }
                _ => {}
            }
        }
    }

    // VirtualName ignored when Ebs is specified in block device mappings
    {
        for rpp in &ctx.cached_data.rule_tables.ebs_iops_property_paths {
            let wc_count = wildcard_count(rpp);
            if wc_count == 1 {
                let prefix = property_path_prefix(rpp);
                for name in m.resources_of_type(&rpp.resource_type) {
                    // Concrete resolution handles the common non-conditional case.
                    if let Some(serde_json::Value::Array(bdms)) = resolve_concrete(m, name, &prefix) {
                        check_bdm_virtualname_ignored(&mut out, m, name, &bdms, &prefix);
                    } else {
                        // Fall back to conditional branch traversal for the
                        // virtual-name-ignored check only
                        let bdm_arrays = resolve_all_json(m, name, &prefix);
                        for bdms_val in &bdm_arrays {
                            if let serde_json::Value::Array(bdms) = bdms_val {
                                for (i, bdm) in bdms.iter().enumerate() {
                                    if let Some(vname) = bdm.get("VirtualName").and_then(|v| v.as_str())
                                        && bdm.get("Ebs").is_none()
                                        && !EPHEMERAL_RE.is_match(vname)
                                    {
                                        out.push(make_resource_diagnostic(
                                                "E3715",
                                                &format!("'{}' is not a valid ephemeral device name. Expected format is 'ephemeralN' where N is 0-23", vname),
                                                m, name,
                                                &format!("{}.{}.VirtualName", prefix, i),
                                                None,
                                            ));
                                    }
                                }
                            }
                        }
                    }
                }
            } else if wc_count == 2 {
                // SpotFleet-style double-wildcard
                let Some(first_wc) = rpp.segments.iter().position(|s| matches!(s, PathSegment::Wildcard)) else {
                    continue;
                };
                let outer_prefix: String = rpp.segments[..first_wc]
                    .iter()
                    .filter_map(|s| match s {
                        PathSegment::Literal(l) => Some(l.as_str()),
                        _ => None,
                    })
                    .collect::<Vec<_>>()
                    .join(".");
                let Some(second_wc) = rpp.segments.iter().rposition(|s| matches!(s, PathSegment::Wildcard)) else {
                    continue;
                };
                let middle_segments: String = rpp.segments[first_wc + 1..second_wc]
                    .iter()
                    .filter_map(|s| match s {
                        PathSegment::Literal(l) => Some(l.as_str()),
                        _ => None,
                    })
                    .collect::<Vec<_>>()
                    .join(".");
                for name in m.resources_of_type(&rpp.resource_type) {
                    if let Some(serde_json::Value::Array(specs)) = resolve_concrete(m, name, &outer_prefix) {
                        for (si, spec) in specs.iter().enumerate() {
                            let bdms_val = if middle_segments.is_empty() {
                                spec.as_array()
                            } else {
                                navigate_json_path(spec, &middle_segments).and_then(|v| v.as_array())
                            };
                            if let Some(bdms) = bdms_val {
                                check_bdm_virtualname_ignored(
                                    &mut out,
                                    m,
                                    name,
                                    bdms,
                                    &format!("{}.{}.{}", outer_prefix, si, middle_segments),
                                );
                            }
                        }
                    }
                }
            }
        }
    }

    // Validate ECS Fargate tasks have required properties and values.
    // Each property scenario is paired with a compatible Fargate scenario so
    // values from mutually exclusive EC2 deployments cannot create findings.
    {
        let supported_log_drivers = &ctx.cached_data.fargate_supported_log_drivers;
        let supported_log_drivers_rendered = render_str_list(supported_log_drivers);
        let supported_log_driver_fix = &ctx.cached_data.fargate_supported_log_driver_fix;

        for name in m.resources_of_type("AWS::ECS::TaskDefinition") {
            let fargate_scenarios = fargate_condition_scenarios(m, name);
            if fargate_scenarios.is_empty() {
                continue;
            }

            let properties_scenarios = m.resolve_properties_scenarios(name);
            let property_can_be_missing = |property_name: &str| {
                properties_scenarios.iter().any(|(properties, conditions)| {
                    !scenario_has_effective_property(properties, property_name)
                        && scenario_overlaps_any(m, name, conditions, &fargate_scenarios)
                })
            };

            let network_mode_scenarios = m.resolve_scenarios_json(name, "Properties.NetworkMode");
            let network_mode_can_be_missing = property_can_be_missing("NetworkMode");
            if network_mode_can_be_missing {
                out.push(make_resource_diagnostic(
                    "E3048",
                    "Fargate requires NetworkMode to be specified as 'awsvpc'",
                    m,
                    name,
                    KEY_PROPERTIES,
                    Some("Set NetworkMode to 'awsvpc'"),
                ));
            }
            let mut invalid_network_modes = HashSet::new();
            for (value, conditions) in &network_mode_scenarios {
                if let Some(network_mode) = value.as_str()
                    && network_mode != "awsvpc"
                    && scenario_overlaps_any(m, name, conditions, &fargate_scenarios)
                    && invalid_network_modes.insert(network_mode.to_string())
                {
                    out.push(make_resource_diagnostic(
                        "E3048",
                        &format!("Fargate requires NetworkMode 'awsvpc', got '{}'", network_mode),
                        m,
                        name,
                        "Properties.NetworkMode",
                        Some("Set NetworkMode to 'awsvpc'"),
                    ));
                }
            }

            let cpu_scenarios = m.resolve_scenarios_json(name, "Properties.Cpu");
            let cpu_can_be_missing = property_can_be_missing("Cpu");
            if cpu_can_be_missing {
                out.push(make_resource_diagnostic(
                    "E3048",
                    "Fargate requires Cpu to be specified",
                    m,
                    name,
                    KEY_PROPERTIES,
                    Some("Set Cpu to a valid Fargate value (256, 512, 1024, 2048, 4096, 8192, 16384, or 32768)"),
                ));
            }
            let mut invalid_cpu_values = HashSet::new();
            for (value, conditions) in &cpu_scenarios {
                if scenario_overlaps_any(m, name, conditions, &fargate_scenarios)
                    && matches!(cpu_is_offered(value), Some(false))
                {
                    let rendered = render_value(value);
                    if invalid_cpu_values.insert(rendered.clone()) {
                        out.push(make_resource_diagnostic(
                            "E3048",
                            &format!(
                                "Fargate Cpu value {} is not valid. Valid sizes are {} CPU units or {} vCPU.",
                                rendered,
                                render_str_list(CPU_UNIT_LABELS),
                                render_str_list(VCPU_SIZES.iter().map(|(label, _)| *label)),
                            ),
                            m,
                            name,
                            "Properties.Cpu",
                            Some("Use a valid Fargate Cpu size in CPU units or vCPU"),
                        ));
                    }
                }
            }

            let memory_can_be_missing = property_can_be_missing("Memory");
            if memory_can_be_missing {
                out.push(make_resource_diagnostic(
                    "E3048",
                    "Fargate requires Memory to be specified",
                    m,
                    name,
                    KEY_PROPERTIES,
                    Some("Set Memory to a valid Fargate value"),
                ));
            }

            let placement_is_unsupported = properties_scenarios.iter().any(|(properties, conditions)| {
                scenario_has_effective_property(properties, "PlacementConstraints")
                    && scenario_overlaps_any(m, name, conditions, &fargate_scenarios)
            });
            if placement_is_unsupported {
                out.push(make_resource_diagnostic(
                    "E3048",
                    "Fargate does not support PlacementConstraints",
                    m,
                    name,
                    "Properties.PlacementConstraints",
                    Some("Remove PlacementConstraints for Fargate tasks"),
                ));
            }

            // Expanding the whole list preserves both condition branches even
            // when `ContainerDefinitions` itself is wrapped in `Fn::If`; every
            // returned scenario is concrete enough to inspect by index.
            let mut reported_drivers: HashSet<(usize, String)> = HashSet::new();
            for (container_value, container_conditions) in
                m.resolve_scenarios_json(name, "Properties.ContainerDefinitions")
            {
                if !scenario_overlaps_any(m, name, &container_conditions, &fargate_scenarios) {
                    continue;
                }
                let Some(container_definitions) = container_value.as_array() else {
                    continue;
                };
                for (container_index, container_definition) in container_definitions.iter().enumerate() {
                    let Some(driver) = container_definition
                        .get("LogConfiguration")
                        .and_then(|configuration| configuration.get("LogDriver"))
                        .and_then(serde_json::Value::as_str)
                    else {
                        continue;
                    };
                    if supported_log_drivers.iter().any(|supported| supported == driver)
                        || !reported_drivers.insert((container_index, driver.to_string()))
                    {
                        continue;
                    }
                    let driver_path =
                        format!("Properties.ContainerDefinitions.{container_index}.LogConfiguration.LogDriver");
                    out.push(make_resource_diagnostic(
                        "E3048",
                        &format!(
                            "Fargate does not support log driver '{}'. Supported drivers: {}",
                            driver, supported_log_drivers_rendered,
                        ),
                        m,
                        name,
                        &driver_path,
                        Some(supported_log_driver_fix),
                    ));
                }
            }
        }
    }

    out
}

fn check_bdm_iops_ignored(
    out: &mut Vec<Diagnostic>,
    m: &Arc<SemanticModel>,
    name: &str,
    bdms: &[serde_json::Value],
    base_path: &str,
    rule_id: &str,
    ignored_types: &[&str],
) {
    for (i, bdm) in bdms.iter().enumerate() {
        if let Some(ebs) = bdm.get("Ebs")
            && ebs.get("Iops").is_some()
            && let Some(vtype) = ebs.get("VolumeType").and_then(|v| v.as_str())
            && ignored_types.contains(&vtype)
        {
            out.push(make_resource_diagnostic(
                rule_id,
                &format!("Iops is ignored when VolumeType is '{}'", vtype),
                m,
                name,
                &format!("{}.{}.Ebs.Iops", base_path, i),
                None,
            ));
        }
    }
}

static EPHEMERAL_RE: LazyLock<regex::Regex> =
    LazyLock::new(|| regex::Regex::new(r"^ephemeral([0-9]|1[0-9]|2[0-3])$").expect("Invalid EPHEMERAL_RE"));

fn check_bdm_virtualname_ignored(
    out: &mut Vec<Diagnostic>,
    m: &Arc<SemanticModel>,
    name: &str,
    bdms: &[serde_json::Value],
    base_path: &str,
) {
    for (i, bdm) in bdms.iter().enumerate() {
        let has_vname = bdm.get("VirtualName");
        let has_ebs = bdm.get("Ebs").is_some();
        if let Some(vname_val) = has_vname {
            if has_ebs {
                out.push(make_resource_diagnostic(
                    "W3698",
                    "VirtualName is ignored when Ebs is specified",
                    m,
                    name,
                    &format!("{}.{}.VirtualName", base_path, i),
                    None,
                ));
            } else if let Some(vname) = vname_val.as_str()
                && !EPHEMERAL_RE.is_match(vname)
            {
                out.push(make_resource_diagnostic(
                    "E3715",
                    &format!(
                        "'{}' is not a valid ephemeral device name. Expected format is 'ephemeralN' where N is 0-23",
                        vname
                    ),
                    m,
                    name,
                    &format!("{}.{}.VirtualName", base_path, i),
                    None,
                ));
            }
        }
    }
}

/// The inner `{ "<region>": ... }` map of an embedded enum document, reached by
/// unwrapping the single top-level document key. `None` when the key is absent or
/// the shape is unexpected. Both flat and conditional documents share this shape;
/// the region-scoping logic lives in [`template_model::region_enums`].
fn region_map_for_key<'a>(
    enum_data: &'a HashMap<String, serde_json::Value>,
    enum_key: &str,
) -> Option<&'a serde_json::Map<String, serde_json::Value>> {
    enum_data.get(enum_key)?.as_object()?.values().next()?.as_object()
}

/// Allowed instance-type/node-type values from a flat enum document for the
/// effective scope: the single `region` when configured, or the union of all
/// regions when not (`None` region). `None` when the document is absent or has no
/// entry for the effective scope, so the caller skips validation.
fn region_flat_allowed<'a>(
    enum_data: &'a HashMap<String, serde_json::Value>,
    enum_key: &str,
    region: Option<&str>,
) -> Option<HashSet<&'a str>> {
    region_enums::flat_allowed_value_set(region_map_for_key(enum_data, enum_key)?, region)
}

/// A scalar string property value, collapsing an `Fn::If`-wrapped value to its
/// true branch, so the region-scoped enum rules that key on this (instance class
/// plus the Engine/LicenseModel consts) see the concrete value. `resolve_concrete`
/// alone is `Conditional`-blind and would silently skip `Fn::If`-wrapped values.
fn resolve_enum_string(m: &SemanticModel, rid: &str, path: &str) -> Option<String> {
    let resolved = m.resolve_deep(rid, path).or_else(|| m.resolve(rid, path).cloned())?;
    match resolved_to_json_best_effort(&resolved) {
        serde_json::Value::String(s) => Some(s),
        _ => None,
    }
}

fn resolve_concrete_strings(m: &SemanticModel, rid: &str, path: &str) -> Vec<String> {
    let Some(resolved) = m.resolve_deep(rid, path).or_else(|| m.resolve(rid, path).cloned()) else {
        return Vec::new();
    };
    let mut out = Vec::new();
    collect_concrete_strings(&resolved, &mut out);
    out
}

fn collect_concrete_strings(value: &ResolvedValue, out: &mut Vec<String>) {
    match value {
        ResolvedValue::Concrete { value: v } => {
            if let Some(s) = v.0.as_str() {
                out.push(s.to_string());
            }
        }
        ResolvedValue::Enum { variants } => {
            for variant in variants {
                collect_concrete_strings(variant, out);
            }
        }
        ResolvedValue::List { items } => {
            for item in items {
                collect_concrete_strings(item, out);
            }
        }
        ResolvedValue::Conditional { if_true, if_false, .. } => {
            collect_concrete_strings(if_true, out);
            collect_concrete_strings(if_false, out);
        }
        _ => {}
    }
}

/// First scalar (string/number/bool) value that repeats in the array, formatted
/// the way the diagnostic renders it. Non-scalar items and nulls are ignored -
/// only repeated primitives are advisory.
fn first_duplicate_scalar(items: &[serde_json::Value]) -> Option<String> {
    let mut seen = HashSet::new();
    for item in items {
        if item.is_null() || item.is_array() || item.is_object() {
            continue;
        }
        let key = item.to_string();
        if !seen.insert(key) {
            return Some(item.to_string());
        }
    }
    None
}

fn resolved_to_json_best_effort(rv: &ResolvedValue) -> serde_json::Value {
    match rv {
        ResolvedValue::Concrete { value: v } => v.0.clone(),
        ResolvedValue::List { items } => {
            serde_json::Value::Array(items.iter().map(resolved_to_json_best_effort).collect())
        }
        ResolvedValue::Map { entries } => {
            let mut map = serde_json::Map::new();
            for e in entries {
                map.insert(e.key.clone(), resolved_to_json_best_effort(&e.value));
            }
            serde_json::Value::Object(map)
        }
        ResolvedValue::Enum { variants } => {
            for v in variants {
                if let ResolvedValue::Concrete { value: c } = v {
                    return c.0.clone();
                }
            }
            serde_json::Value::Null
        }
        ResolvedValue::Conditional { if_true, .. } => resolved_to_json_best_effort(if_true),
        ResolvedValue::Reference { target, .. } => serde_json::json!({(FN_REF): target}),
        ResolvedValue::Dynamic { .. } | ResolvedValue::TypedDynamic { .. } => serde_json::Value::Null,
    }
}

fn arn_matches_format(resource_arn: &str, format_arn: &str) -> bool {
    // An ARN has up to six colon-delimited parts; a shorter one is padded with
    // "*" wildcards to six, so a truncated ARN
    // (e.g. one missing region/account) still compares rather than being treated
    // as a non-match (which would produce a spurious finding).
    let pad_to_six = |arn: &str| -> Vec<String> {
        let mut parts: Vec<String> = arn.splitn(6, ':').map(str::to_string).collect();
        while parts.len() < 6 {
            parts.push("*".to_string());
        }
        parts
    };
    let r_parts = pad_to_six(resource_arn);
    let f_parts = pad_to_six(format_arn);
    for i in 0..5 {
        if r_parts[i] == "*" {
            continue;
        }
        if matches!(f_parts[i].as_str(), "${Partition}" | "${Region}" | "${Account}") {
            continue;
        }
        if r_parts[i] != f_parts[i] {
            return false;
        }
    }
    if r_parts[5] == "*" {
        return true;
    }
    let delimiter = if f_parts[5].contains(':') {
        ':'
    } else if f_parts[5].contains('/') {
        '/'
    } else {
        return true;
    };
    for (r_seg, f_seg) in r_parts[5].split(delimiter).zip(f_parts[5].split(delimiter)) {
        if r_seg == f_seg {
            continue;
        }
        if r_seg == "*" || r_seg.starts_with('*') || f_seg.is_empty() || f_seg == ".*" {
            return true;
        }
        if r_seg.starts_with(f_seg) && r_seg.contains('*') {
            return true;
        }
        return false;
    }
    true
}

/// The outcome of classifying a statement's `Resource`/`NotResource` entries for
/// the action-format check: the literal ARNs to match against, whether any entry
/// was a function whose ARN is unknowable (so the format check is skipped), and
/// whether the whole statement is exempt (a `*`, a dynamic reference, or a
/// `Ref` to a parameter).
#[derive(Default)]
struct IamResourceClassification {
    resources: Vec<String>,
    using_functions: bool,
    skip_statement: bool,
}

/// Classifies one `Resource`/`NotResource` entry. When `at_top_level` (the entry
/// is the property's scalar value, not a list element), an `Fn::If` is enumerated
/// so every reachable branch contributes - a resource is acceptable if any branch
/// matches. As a list element, an `Fn::If` is instead an opaque function whose
/// ARN is unknowable. A concrete ARN is a match candidate unless it was
/// synthesized by an intrinsic (its real ARN is only known at deploy time); a
/// `*`/dynamic-reference/`Ref`-to-parameter exempts the statement.
fn classify_iam_resource_entry(
    m: &Arc<SemanticModel>,
    name: &str,
    item_path: &str,
    val: &serde_json::Value,
    at_top_level: bool,
    out: &mut IamResourceClassification,
) {
    match val {
        serde_json::Value::String(s) => {
            if s.contains("{{resolve:") || s == "*" {
                out.skip_statement = true;
                return;
            }
            if m.is_from_intrinsic(name, item_path) {
                out.using_functions = true;
            } else {
                out.resources.push(s.clone());
            }
        }
        serde_json::Value::Array(arr) => {
            // A scalar `Fn::If` branch may itself be a list of ARNs; classify
            // each element as a (non-top-level) entry.
            for (i, v) in arr.iter().enumerate() {
                classify_iam_resource_entry(m, name, &format!("{}.{}", item_path, i), v, false, out);
                if out.skip_statement {
                    return;
                }
            }
        }
        serde_json::Value::Object(obj) => {
            // `Fn::If` survives preserved as `{"Fn::If": [cond, then, else]}`.
            // Enumerate both branches only for a scalar value (mirroring how the
            // reference model descends conditions); an `Fn::If` list element is an
            // opaque function.
            if at_top_level
                && let Some(serde_json::Value::Array(branches)) = obj.get(FN_IF)
                && branches.len() == 3
            {
                classify_iam_resource_entry(m, name, &format!("{}.{}.1", item_path, FN_IF), &branches[1], true, out);
                if out.skip_statement {
                    return;
                }
                classify_iam_resource_entry(m, name, &format!("{}.{}.2", item_path, FN_IF), &branches[2], true, out);
                return;
            }
            if let Some(ref_target) = obj.get(FN_REF).and_then(|v| v.as_str())
                && m.parameters.contains_key(ref_target)
            {
                out.skip_statement = true;
                return;
            }
            out.using_functions = true;
        }
        serde_json::Value::Null => {
            out.using_functions = true;
        }
        _ => {}
    }
}

fn check_iam_action_resources(
    out: &mut Vec<Diagnostic>,
    m: &Arc<SemanticModel>,
    name: &str,
    doc: &serde_json::Value,
    patterns: &HashMap<String, Vec<String>>,
) {
    let stmts = match doc.get("Statement").and_then(|s| s.as_array()) {
        Some(s) => s,
        None => return,
    };
    for (stmt_idx, stmt) in stmts.iter().enumerate() {
        if !stmt.is_object() {
            continue;
        }
        let mut classification = IamResourceClassification::default();

        for key in &["Resource", "NotResource"] {
            let field = match stmt.get(*key) {
                Some(v) => v,
                None => continue,
            };
            // A scalar field is anchored at `...<key>`; array elements at
            // `...<key>.<index>`. A scalar field value may be an `Fn::If` whose
            // branches are enumerated (`at_top_level`); a list element that is an
            // `Fn::If` is instead treated as an opaque function.
            let items: Vec<(String, &serde_json::Value, bool)> = match field {
                serde_json::Value::Array(arr) => arr
                    .iter()
                    .enumerate()
                    .map(|(i, v)| (format!("Properties.PolicyDocument.Statement.{}.{}.{}", stmt_idx, key, i), v, false))
                    .collect(),
                other => vec![(format!("Properties.PolicyDocument.Statement.{}.{}", stmt_idx, key), other, true)],
            };
            for (item_path, val, at_top_level) in items {
                classify_iam_resource_entry(m, name, &item_path, val, at_top_level, &mut classification);
                if classification.skip_statement {
                    break;
                }
            }
            if classification.skip_statement {
                break;
            }
        }
        if classification.skip_statement {
            continue;
        }
        let using_functions = classification.using_functions;
        let all_resources: Vec<&str> = classification.resources.iter().map(String::as_str).collect();

        let actions: Vec<&str> = match stmt.get("Action") {
            Some(serde_json::Value::String(s)) => vec![s.as_str()],
            Some(serde_json::Value::Array(arr)) => arr.iter().filter_map(|v| v.as_str()).collect(),
            _ => continue,
        };
        for action in &actions {
            if action.contains('*') || action.contains('?') || !action.contains(':') {
                continue;
            }
            let key = action.to_lowercase();
            let Some(candidate_formats) = patterns.get(&key) else {
                continue;
            };
            if using_functions {
                continue;
            }
            let matched = candidate_formats.iter().any(|fmt| all_resources.iter().any(|r| arn_matches_format(r, fmt)));
            if !matched {
                let formats_display =
                    candidate_formats.iter().map(|s| format!("'{}'", s)).collect::<Vec<_>>().join(", ");
                let resource_field = if stmt.get("Resource").is_some() { "Resource" } else { "NotResource" };
                out.push(make_resource_diagnostic(
                    "I3510",
                    &format!("action '{}' requires a resource of [{}]", action, formats_display),
                    m,
                    name,
                    &format!("Properties.PolicyDocument.Statement.{}.{}", stmt_idx, resource_field),
                    None,
                ));
            }
        }
    }
}

fn check_dynamic_ref_spaces(
    out: &mut Vec<Diagnostic>,
    m: &Arc<SemanticModel>,
    name: &str,
    path: &str,
    val: &ResolvedValue,
) {
    match val {
        ResolvedValue::Concrete { value: v } => {
            if let Some(s) = v.as_str() {
                // Detect strings that look like dynamic reference attempts but have spaces
                // that prevent CloudFormation from resolving them.
                // Valid: {{resolve:ssm:...}}  Invalid: {{ resolve:ssm:...}}
                if s.contains("resolve:") && s.contains("{{") && s.contains("}}") {
                    // Has a valid dynamic ref - no warning needed
                    if s.contains("{{resolve:") {
                        return;
                    }
                    // Looks like a dynamic ref attempt with spaces
                    out.push(make_resource_diagnostic(
                    "W1053",
                    &format!(
                        "'{}' has spaces and will not be resolved as a dynamic reference. Remove spaces from '{{{{resolve:...}}}}'",
                        s
                    ),
                    m,
                    name,
                    path,
                    Some("Remove spaces from the dynamic reference"),
                ));
                }
            }
        }
        ResolvedValue::List { items } => {
            for (i, item) in items.iter().enumerate() {
                check_dynamic_ref_spaces(out, m, name, &format!("{}.{}", path, i), item);
            }
        }
        ResolvedValue::Map { entries } => {
            for e in entries {
                check_dynamic_ref_spaces(out, m, name, &format!("{}.{}", path, e.key), &e.value);
            }
        }
        _ => {}
    }
}

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

    #[test]
    fn ordered_port_range_protocol_tcp_udp_names_any_case() {
        for proto in ["tcp", "TCP", "Tcp", "udp", "UDP"] {
            let value = serde_json::Value::String(proto.to_string());
            assert!(sg_protocol_has_ordered_port_range(Some(&value)), "{} should have an ordered port range", proto);
        }
    }

    #[test]
    fn ordered_port_range_protocol_tcp_udp_numbers() {
        for proto in [serde_json::json!(6), serde_json::json!(17), serde_json::json!("6"), serde_json::json!("17")] {
            assert!(sg_protocol_has_ordered_port_range(Some(&proto)), "{} should have an ordered port range", proto);
        }
    }

    #[test]
    fn ordered_port_range_protocol_rejects_icmp_type_code_protocols() {
        for proto in [
            serde_json::json!("icmp"),
            serde_json::json!("ICMP"),
            serde_json::json!("icmpv6"),
            serde_json::json!(1),
            serde_json::json!("1"),
            serde_json::json!(58),
            serde_json::json!("58"),
        ] {
            assert!(!sg_protocol_has_ordered_port_range(Some(&proto)), "{} carries ICMP type/code, not a range", proto);
        }
    }

    #[test]
    fn ordered_port_range_protocol_rejects_all_protocols_wildcard_and_unknown() {
        for proto in [serde_json::json!("-1"), serde_json::json!(-1), serde_json::json!("esp"), serde_json::json!(50)] {
            assert!(!sg_protocol_has_ordered_port_range(Some(&proto)), "{} ignores ports, no ordering applies", proto);
        }
    }

    #[test]
    fn ordered_port_range_protocol_rejects_missing_or_unresolved_protocol() {
        assert!(!sg_protocol_has_ordered_port_range(None), "absent protocol should not have an ordered port range");
        let unresolved_ref = serde_json::json!({"Ref": "ProtocolParam"});
        assert!(
            !sg_protocol_has_ordered_port_range(Some(&unresolved_ref)),
            "unresolved protocol should not have an ordered port range"
        );
    }

    #[test]
    fn ports_are_meaningful_for_icmpv6_in_every_spelling() {
        for proto in [
            serde_json::json!("icmpv6"),
            serde_json::json!("ICMPv6"),
            serde_json::json!("ICMPV6"),
            serde_json::json!("58"),
            serde_json::json!(58),
        ] {
            assert!(
                !sg_protocol_ignores_ports(Some(&proto)),
                "{} carries an ICMP type/code, ports are not ignored",
                proto
            );
        }
    }

    #[test]
    fn ports_are_optional_for_icmpv6_but_required_for_tcp_udp_icmp() {
        for proto in
            [serde_json::json!("icmpv6"), serde_json::json!("ICMPv6"), serde_json::json!("58"), serde_json::json!(58)]
        {
            assert!(!sg_protocol_requires_ports(Some(&proto)), "{} may omit its type/code", proto);
        }
        for proto in [
            serde_json::json!("tcp"),
            serde_json::json!("UDP"),
            serde_json::json!("icmp"),
            serde_json::json!("6"),
            serde_json::json!(17),
            serde_json::json!(1),
        ] {
            assert!(sg_protocol_requires_ports(Some(&proto)), "{} must state its ports", proto);
        }
    }

    #[test]
    fn ports_are_ignored_for_wildcard_and_other_protocols_only() {
        for proto in [serde_json::json!("-1"), serde_json::json!(-1), serde_json::json!("esp"), serde_json::json!(50)] {
            assert!(sg_protocol_ignores_ports(Some(&proto)), "{} allows every port, so a range is ignored", proto);
        }
        for proto in
            [serde_json::json!("tcp"), serde_json::json!("ICMP"), serde_json::json!(6), serde_json::json!("17")]
        {
            assert!(!sg_protocol_ignores_ports(Some(&proto)), "{} uses its ports", proto);
        }
    }

    #[test]
    fn unresolved_protocol_neither_requires_nor_ignores_ports() {
        let unresolved_ref = serde_json::json!({"Ref": "ProtocolParam"});
        assert!(!sg_protocol_requires_ports(None));
        assert!(!sg_protocol_ignores_ports(None));
        assert!(!sg_protocol_requires_ports(Some(&unresolved_ref)));
        assert!(!sg_protocol_ignores_ports(Some(&unresolved_ref)));
    }

    #[test]
    fn cidr_strict_valid_network() {
        assert!(is_valid_cidr_strict("10.0.0.0/16"));
    }

    #[test]
    fn cidr_strict_invalid_host_bits() {
        assert!(!is_valid_cidr_strict("10.0.0.1/16"));
    }

    #[test]
    fn cidr_strict_invalid_format() {
        assert!(!is_valid_cidr_strict("not-a-cidr"));
    }

    #[test]
    fn cidr_strict_slash_32_any_host() {
        assert!(is_valid_cidr_strict("192.168.1.1/32"));
    }

    #[test]
    fn arn_matches_exact() {
        assert!(arn_matches_format("arn:aws:s3:::my-bucket", "arn:aws:s3:::my-bucket"));
    }

    #[test]
    fn arn_matches_resource_wildcard() {
        assert!(arn_matches_format("arn:*:s3:::my-bucket", "arn:${Partition}:s3:::my-bucket"));
    }

    #[test]
    fn arn_matches_placeholder_partition() {
        assert!(arn_matches_format("arn:aws:iam::123456:user/test", "arn:${Partition}:iam::${Account}:user/.*"));
    }

    #[test]
    fn arn_matches_resource_star_sixth() {
        assert!(arn_matches_format("arn:aws:s3:::*", "arn:${Partition}:s3:::.*"));
    }

    #[test]
    fn arn_no_match_different_service() {
        assert!(!arn_matches_format("arn:aws:ec2:::instance", "arn:${Partition}:s3:::.*"));
    }

    #[test]
    fn arn_short_is_padded_with_wildcards() {
        // A resource ARN with fewer than six parts is padded with "*", so the
        // missing service/region/account are treated as wildcards on the
        // resource side and the ARN still matches a format.
        assert!(arn_matches_format("arn:aws", "arn:aws:s3:::*"));
    }

    #[test]
    fn arn_short_still_rejects_service_mismatch() {
        // Padding does not mask a concrete mismatch in a present part.
        assert!(!arn_matches_format("arn:aws:ec2", "arn:aws:s3:::*"));
    }
}