formualizer_eval/

function_registry.rs

use crate::function::{FnCaps, Function};
use crate::function_contract::{
    FunctionDependencySemantics, FunctionEnvironmentSemantics, FunctionEvaluationSemantics,
    FunctionResultSemantics, FunctionSemanticContract, FunctionSemanticIdentity,
};
use once_cell::sync::Lazy;
use std::collections::{HashMap, VecDeque};
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};

type RegistryKey = (String, String);

#[derive(Clone)]
struct RegistryEntry {
    function: Arc<dyn Function>,
    generation: u64,
    trusted_builtin: bool,
    semantics: SemanticContractResolution,
    semantics_by_arity: Arc<RwLock<HashMap<usize, SemanticContractResolution>>>,
}

#[derive(Clone)]
struct AliasEntry {
    target: RegistryKey,
    owner: Option<(RegistryKey, u64)>,
}

struct RegistryState {
    registrations: HashMap<RegistryKey, RegistryEntry>,
    aliases: HashMap<RegistryKey, AliasEntry>,
    semantic_epoch: u64,
    semantic_changes: VecDeque<(u64, Vec<RegistryKey>)>,
}

impl Default for RegistryState {
    fn default() -> Self {
        Self {
            registrations: HashMap::new(),
            aliases: HashMap::new(),
            semantic_epoch: 1,
            semantic_changes: VecDeque::new(),
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum SemanticConformanceIssue {
    CapabilityPanicked,
    DependencyContractPanicked,
    SemanticContractPanicked,
    ArityMetadataPanicked,
    VariadicMetadataPanicked,
    AliasMetadataPanicked,
    ArgumentSchemaPanicked,
    AritySchemaMismatch,
    DynamicDependencyMismatch,
    ShortCircuitMismatch,
    ReferenceResultMismatch,
    LocalEnvironmentMismatch,
    SpillResultMismatch,
    PrecisionContractMismatch,
    PrecisionContractInvalid,
}

#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct SemanticContractResolution {
    pub contract: Option<FunctionSemanticContract>,
    pub generation: u64,
    pub trusted_builtin: bool,
    pub issues: Vec<SemanticConformanceIssue>,
}

impl SemanticContractResolution {
    pub fn conforms(&self) -> bool {
        self.contract.is_some() && self.issues.is_empty()
    }
}

#[derive(Clone)]
#[non_exhaustive]
pub struct ResolvedFunction {
    pub namespace: String,
    pub canonical_name: String,
    pub function: Arc<dyn Function>,
    pub semantics: SemanticContractResolution,
}

static REGISTRY: Lazy<RwLock<RegistryState>> = Lazy::new(|| RwLock::new(RegistryState::default()));
static NEXT_GENERATION: AtomicU64 = AtomicU64::new(1);

/// Lock-free mirror of `RegistryState::semantic_epoch`.
///
/// Published only by `publish_semantic_change`, which takes the global
/// `REGISTRY` write guard by type, so any holder of a registry read guard (for
/// example a `SemanticEpochReadGuard`) observes exactly the guarded epoch here.
/// Without a guard it is a monotonic, possibly stale lower bound. State-only log
/// advancement on a local `RegistryState` (`advance_semantic_log`) never
/// touches it.
static SEMANTIC_EPOCH_MIRROR: AtomicU64 = AtomicU64::new(1);

/// Displacement count at which the `load_builtins` shortcut is permanently
/// disabled; the counter never wraps.
const BUILTIN_DISPLACEMENTS_EXHAUSTED: u64 = u64::MAX;

/// Bookkeeping for the `load_builtins` shortcut.
struct BuiltinLoadState {
    /// Registrations that replaced a trusted entry with a non-builtin function.
    /// Monotonic, bumped under the `REGISTRY` write lock; sticky at
    /// `BUILTIN_DISPLACEMENTS_EXHAUSTED`.
    displacements: AtomicU64,
    /// One plus the largest `displacements` value read at the start of a pass
    /// that ran to completion; 0 until one has.
    complete_since: AtomicU64,
}

impl BuiltinLoadState {
    const fn new(displacements: u64) -> Self {
        Self {
            displacements: AtomicU64::new(displacements),
            complete_since: AtomicU64::new(0),
        }
    }

    fn record_displacement(&self) {
        // `fetch_update` is deprecated in newer Rust and its replacement
        // `try_update` is missing from the Pyodide wheel's Rust 1.93, so use
        // a compare-exchange loop that builds on both.
        let mut count = self.displacements.load(Ordering::Acquire);
        loop {
            let next = count.saturating_add(1);
            match self.displacements.compare_exchange_weak(
                count,
                next,
                Ordering::AcqRel,
                Ordering::Acquire,
            ) {
                Ok(_) => break,
                Err(observed) => count = observed,
            }
        }
    }

    fn loaded(&self) -> bool {
        let complete_since = self.complete_since.load(Ordering::Acquire);
        let displacements = self.displacements.load(Ordering::Acquire);
        complete_since != 0
            && displacements != BUILTIN_DISPLACEMENTS_EXHAUSTED
            && displacements + 1 == complete_since
    }

    fn begin_pass(&self) -> u64 {
        self.displacements.load(Ordering::Acquire)
    }

    fn finish_pass(&self, token: u64) {
        // Every completed pass's token is sound: a displacement after its start
        // leaves the counter above it, so `loaded` stays false until a later
        // pass completes. Keeping the maximum stops a slow pass that started
        // before a displacement from hiding a newer pass's completion. An
        // exhausted token publishes nothing.
        if token != BUILTIN_DISPLACEMENTS_EXHAUSTED {
            self.complete_since.fetch_max(token + 1, Ordering::AcqRel);
        }
    }
}

static BUILTIN_LOAD: BuiltinLoadState = BuiltinLoadState::new(0);

/// Per-thread registration call counts, used as untimed cost evidence by
/// tests and probes. Not part of the product API.
#[cfg(any(test, feature = "test-support"))]
#[doc(hidden)]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct RegistrationCallCounts {
    /// Calls to `builtins::load_builtins`.
    pub load_builtins: u64,
    /// Calls to the internal `register` (builtin or user).
    pub register: u64,
    /// Semantic metadata inspections (`inspect_semantics*`).
    pub inspect_semantics: u64,
}

#[cfg(any(test, feature = "test-support"))]
thread_local! {
    static CALL_COUNTS: std::cell::Cell<RegistrationCallCounts> =
        const { std::cell::Cell::new(RegistrationCallCounts { load_builtins: 0, register: 0, inspect_semantics: 0 }) };
}

/// Registration call counts observed on the current thread.
#[cfg(any(test, feature = "test-support"))]
#[doc(hidden)]
pub fn registration_call_counts() -> RegistrationCallCounts {
    CALL_COUNTS.with(std::cell::Cell::get)
}

/// Bump one per-thread call counter; compiles to nothing without test support.
macro_rules! count_call {
    ($field:ident) => {
        #[cfg(any(test, feature = "test-support"))]
        CALL_COUNTS.with(|counts| {
            let mut current = counts.get();
            current.$field += 1;
            counts.set(current);
        });
    };
}

pub(crate) fn count_load_builtins_call() {
    count_call!(load_builtins);
}

#[inline]
fn norm<S: AsRef<str>>(s: S) -> String {
    s.as_ref().to_uppercase()
}

/// Current registry semantic epoch without taking the registry lock.
///
/// Equal to `SemanticEpochReadGuard::epoch` while such a guard is held; safe to
/// call under one (unlike `semantic_epoch`, which may queue behind a writer).
pub(crate) fn semantic_epoch_lock_free() -> u64 {
    SEMANTIC_EPOCH_MIRROR.load(Ordering::Acquire)
}

/// Whether every builtin is currently registered as a trusted builtin, so a
/// `load_builtins` pass would change nothing.
///
/// True once a pass has completed and no registration has displaced a trusted
/// builtin since that pass started. Registration never removes entries and a
/// trusted builtin entry can only be replaced by a displacing registration, so
/// every key the completed pass made trusted is still trusted.
///
/// This is a linearizable shortcut, not a guarantee that outlives the call: a
/// concurrent displacement may land right after it returns.
pub(crate) fn builtins_loaded() -> bool {
    BUILTIN_LOAD.loaded()
}

/// Start a full `load_builtins` pass; pass the token to `finish_builtin_load_pass`
/// only after every builtin registration returned.
pub(crate) fn begin_builtin_load_pass() -> u64 {
    BUILTIN_LOAD.begin_pass()
}

pub(crate) fn finish_builtin_load_pass(token: u64) {
    BUILTIN_LOAD.finish_pass(token);
}

pub fn semantic_epoch() -> u64 {
    REGISTRY
        .read()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
        .semantic_epoch
}

pub(crate) struct SemanticEpochReadGuard(std::sync::RwLockReadGuard<'static, RegistryState>);

impl SemanticEpochReadGuard {
    pub(crate) fn epoch(&self) -> u64 {
        self.0.semantic_epoch
    }

    pub(crate) fn semantic_changes_affect_requests_since(
        &self,
        epoch: u64,
        requests: impl IntoIterator<Item = (String, String, usize)>,
    ) -> bool {
        semantic_changes_affect_requests_in_state(&self.0, epoch, requests)
    }
}

pub(crate) fn semantic_epoch_read_guard() -> SemanticEpochReadGuard {
    SemanticEpochReadGuard(
        REGISTRY
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner()),
    )
}

pub(crate) struct SemanticChanges {
    pub(crate) epoch: u64,
    pub(crate) complete: bool,
    pub(crate) keys: Vec<(String, String)>,
}

/// Record a semantic change on the global registry and publish the epoch mirror.
///
/// Takes the global write guard by type so the mirror is only ever published
/// while `REGISTRY` is write-locked.
fn publish_semantic_change(
    state: &mut std::sync::RwLockWriteGuard<'static, RegistryState>,
    keys: impl IntoIterator<Item = RegistryKey>,
) {
    let epoch = advance_semantic_log(state, keys);
    let previous = SEMANTIC_EPOCH_MIRROR.fetch_max(epoch, Ordering::AcqRel);
    debug_assert!(previous <= epoch, "semantic epoch mirror moved backward");
}

/// Advance a registry state's epoch and change log without publishing the
/// global mirror; returns the new epoch.
fn advance_semantic_log(
    state: &mut RegistryState,
    keys: impl IntoIterator<Item = RegistryKey>,
) -> u64 {
    state.semantic_epoch = state.semantic_epoch.saturating_add(1);
    let epoch = state.semantic_epoch;
    state
        .semantic_changes
        .push_back((epoch, keys.into_iter().collect()));
    if state.semantic_changes.len() > 1_024 {
        state.semantic_changes.pop_front();
    }
    epoch
}

pub(crate) fn semantic_changes_since(epoch: u64) -> SemanticChanges {
    let state = REGISTRY
        .read()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    semantic_changes_since_in_state(&state, epoch)
}

fn semantic_changes_since_in_state(state: &RegistryState, epoch: u64) -> SemanticChanges {
    let complete = state
        .semantic_changes
        .front()
        .is_none_or(|(oldest, _)| epoch.saturating_add(1) >= *oldest);
    let keys = state
        .semantic_changes
        .iter()
        .filter(|(changed_epoch, _)| *changed_epoch > epoch)
        .flat_map(|(_, keys)| keys.iter().cloned())
        .collect();
    SemanticChanges {
        epoch: state.semantic_epoch,
        complete,
        keys,
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum RegistrationError {
    NameMetadataPanicked,
    NamespaceMetadataPanicked,
}

pub fn try_register_function(function: Arc<dyn Function>) -> Result<(), RegistrationError> {
    register(function, false)
}

pub fn register_function(function: Arc<dyn Function>) {
    let _ = try_register_function(function);
}
pub(crate) fn register_builtin(function: Arc<dyn Function>) {
    register(function, true).expect("builtin name and namespace metadata must not panic");
}

fn register(function: Arc<dyn Function>, trusted_builtin: bool) -> Result<(), RegistrationError> {
    count_call!(register);
    let namespace = catch_unwind(AssertUnwindSafe(|| function.namespace()))
        .map_err(|_| RegistrationError::NamespaceMetadataPanicked)?;
    let name = catch_unwind(AssertUnwindSafe(|| function.name()))
        .map_err(|_| RegistrationError::NameMetadataPanicked)?;
    let key = (norm(namespace), norm(name));
    #[cfg(test)]
    if trusted_builtin {
        tests::record_builtin_key(&key);
    }
    // Re-registering an already-registered trusted builtin is a no-op (see the
    // same check under the write lock below). Decide that under a read lock
    // before allocating a generation or inspecting metadata; the write-lock
    // check stays authoritative for registrations that race this one.
    if trusted_builtin
        && REGISTRY
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .registrations
            .get(&key)
            .is_some_and(|entry| entry.trusted_builtin)
    {
        return Ok(());
    }
    let generation = NEXT_GENERATION.fetch_add(1, Ordering::Relaxed);
    let aliases = catch_unwind(AssertUnwindSafe(|| function.aliases().to_vec()));
    let min_args = catch_unwind(AssertUnwindSafe(|| function.min_args()));
    let initial_arity = min_args.as_ref().copied().unwrap_or(0);
    let mut semantics = match min_args {
        Ok(arity) => inspect_semantics(&function, trusted_builtin, generation, arity),
        Err(_) => failed_resolution(
            generation,
            trusted_builtin,
            SemanticConformanceIssue::ArityMetadataPanicked,
        ),
    };
    let aliases = match aliases {
        Ok(aliases) => aliases,
        Err(_) => {
            semantics
                .issues
                .push(SemanticConformanceIssue::AliasMetadataPanicked);
            semantics.contract = None;
            Vec::new()
        }
    };

    let mut state = REGISTRY
        .write()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    if trusted_builtin
        && state
            .registrations
            .get(&key)
            .is_some_and(|entry| entry.trusted_builtin)
    {
        return Ok(());
    }
    let previous = state
        .registrations
        .get(&key)
        .map(|entry| (entry.generation, entry.trusted_builtin));
    if !trusted_builtin && previous.is_some_and(|(_, previous_trusted)| previous_trusted) {
        BUILTIN_LOAD.record_displacement();
    }
    let mut changed_spellings = Vec::new();
    if let Some((previous_generation, _)) = previous {
        changed_spellings.extend(
            state
                .aliases
                .iter()
                .filter(|(_, alias)| {
                    alias.owner.as_ref() == Some(&(key.clone(), previous_generation))
                })
                .map(|(alias_key, _)| alias_key.clone()),
        );
        state
            .aliases
            .retain(|_, alias| alias.owner.as_ref() != Some(&(key.clone(), previous_generation)));
    }
    state.registrations.insert(
        key.clone(),
        RegistryEntry {
            function: Arc::clone(&function),
            generation,
            trusted_builtin,
            semantics: semantics.clone(),
            semantics_by_arity: Arc::new(RwLock::new(HashMap::from([(initial_arity, semantics)]))),
        },
    );
    for alias in aliases {
        if !alias.eq_ignore_ascii_case(&key.1) {
            let alias_key = (key.0.clone(), norm(alias));
            changed_spellings.push(alias_key.clone());
            state.aliases.insert(
                alias_key,
                AliasEntry {
                    target: key.clone(),
                    owner: Some((key.clone(), generation)),
                },
            );
        }
    }
    changed_spellings.push(key);
    publish_semantic_change(&mut state, changed_spellings);
    Ok(())
}

fn failed_resolution(
    generation: u64,
    trusted_builtin: bool,
    issue: SemanticConformanceIssue,
) -> SemanticContractResolution {
    SemanticContractResolution {
        contract: None,
        generation,
        trusted_builtin,
        issues: vec![issue],
    }
}

fn inspect_semantics(
    function: &Arc<dyn Function>,
    trusted_builtin: bool,
    generation: u64,
    arity: usize,
) -> SemanticContractResolution {
    inspect_semantics_with_identity_metadata(function, trusted_builtin, generation, arity).0
}

fn inspect_semantics_with_identity_metadata(
    function: &Arc<dyn Function>,
    trusted_builtin: bool,
    generation: u64,
    arity: usize,
) -> (SemanticContractResolution, Option<(FnCaps, Vec<bool>)>) {
    count_call!(inspect_semantics);
    let mut issues = Vec::new();
    let inspected_caps = inspected(
        &mut issues,
        SemanticConformanceIssue::CapabilityPanicked,
        || function.caps(),
    );
    let caps = inspected_caps.unwrap_or_else(FnCaps::empty);
    let precision = inspected(
        &mut issues,
        SemanticConformanceIssue::DependencyContractPanicked,
        || function.dependency_contract(arity),
    )
    .flatten();
    let explicit = inspected(
        &mut issues,
        SemanticConformanceIssue::SemanticContractPanicked,
        || function.semantic_contract(arity),
    )
    .flatten();
    let schema = inspected(
        &mut issues,
        SemanticConformanceIssue::ArgumentSchemaPanicked,
        || function.arg_schema(),
    );
    let min_args = inspected(
        &mut issues,
        SemanticConformanceIssue::ArityMetadataPanicked,
        || function.min_args(),
    );
    let variadic = inspected(
        &mut issues,
        SemanticConformanceIssue::VariadicMetadataPanicked,
        || function.variadic(),
    );
    if let (Some(schema), Some(min_args), Some(variadic)) = (schema, min_args, variadic)
        && !schema_allows_arity(schema, min_args, variadic, arity, !trusted_builtin)
    {
        issues.push(SemanticConformanceIssue::AritySchemaMismatch);
    }
    let contract =
        explicit.or_else(|| trusted_builtin.then(|| trusted_contract_from_caps(caps, precision)));
    if let Some(contract) = contract {
        if contract.precision != precision {
            issues.push(SemanticConformanceIssue::PrecisionContractMismatch);
        }
        if !precision_is_valid(contract, arity) {
            issues.push(SemanticConformanceIssue::PrecisionContractInvalid);
        }
        check_capability(
            &mut issues,
            caps.contains(FnCaps::DYNAMIC_DEPENDENCY),
            contract.dependency == FunctionDependencySemantics::Dynamic,
            SemanticConformanceIssue::DynamicDependencyMismatch,
        );
        check_capability(
            &mut issues,
            caps.contains(FnCaps::SHORT_CIRCUIT),
            contract.evaluation == FunctionEvaluationSemantics::ShortCircuit,
            SemanticConformanceIssue::ShortCircuitMismatch,
        );
        check_capability(
            &mut issues,
            caps.contains(FnCaps::RETURNS_REFERENCE),
            contract.result.may_return_reference(),
            SemanticConformanceIssue::ReferenceResultMismatch,
        );
        check_capability(
            &mut issues,
            caps.contains(FnCaps::LOCAL_ENVIRONMENT),
            contract.environment == FunctionEnvironmentSemantics::LocalBindings,
            SemanticConformanceIssue::LocalEnvironmentMismatch,
        );
        check_capability(
            &mut issues,
            caps.contains(FnCaps::MAY_SPILL),
            contract.result.may_spill(),
            SemanticConformanceIssue::SpillResultMismatch,
        );
    }
    let identity_metadata = inspected_caps.zip(schema).map(|(caps, schema)| {
        let repeating = schema.iter().find(|argument| argument.repeating.is_some());
        let argument_by_ref = (0..arity)
            .map(|index| {
                schema
                    .get(index)
                    .or(repeating)
                    .is_some_and(|argument| argument.by_ref)
            })
            .collect();
        (caps, argument_by_ref)
    });
    (
        SemanticContractResolution {
            contract: issues.is_empty().then_some(contract).flatten(),
            generation,
            trusted_builtin,
            issues,
        },
        identity_metadata,
    )
}

fn precision_is_valid(contract: FunctionSemanticContract, arity: usize) -> bool {
    use crate::function_contract::{
        CriteriaValueRange, FunctionArgumentDependencyContract as Arguments,
        FunctionArgumentDependencyRole as Role,
    };
    let Some(precision) = contract.precision else {
        return true;
    };
    if !precision.arity.allows(arity)
        || contract.dependency != FunctionDependencySemantics::RecursiveSyntacticArgs
    {
        return false;
    }
    match precision.arguments {
        Arguments::AllArgs(role) | Arguments::Variadic(role) => {
            !matches!(role, Role::IgnoredLiteral | Role::Unsupported)
        }
        Arguments::CriteriaPairs(criteria) => {
            let value_valid = match criteria.value_range {
                CriteriaValueRange::None => true,
                CriteriaValueRange::Fixed(index) => index < arity,
                CriteriaValueRange::Optional {
                    provided_index,
                    fallback_criteria_range_index,
                } => provided_index <= arity && fallback_criteria_range_index < arity,
            };
            let pair_end = match criteria.value_range {
                CriteriaValueRange::Fixed(index) if index >= criteria.first_criteria_pair => index,
                CriteriaValueRange::Optional { provided_index, .. }
                    if provided_index >= criteria.first_criteria_pair =>
                {
                    provided_index
                }
                _ => arity,
            };
            value_valid
                && criteria.first_criteria_pair < pair_end
                && (pair_end - criteria.first_criteria_pair).is_multiple_of(2)
        }
        Arguments::LocalBindingPairs => {
            contract.environment == FunctionEnvironmentSemantics::LocalBindings
                && arity >= 3
                && !arity.is_multiple_of(2)
        }
        Arguments::LambdaParameters => {
            contract.environment == FunctionEnvironmentSemantics::LocalBindings && arity >= 1
        }
    }
}

fn schema_allows_arity(
    schema: &[crate::args::ArgSchema],
    min_args: usize,
    variadic: bool,
    arity: usize,
    strict_required_count: bool,
) -> bool {
    if schema.is_empty() {
        return min_args == 0 && arity == 0;
    }

    let mut optional_seen = false;
    let mut required_count = 0usize;
    let mut repeating = None;
    for (index, argument) in schema.iter().enumerate() {
        if argument.required {
            if optional_seen {
                return false;
            }
            required_count += 1;
        } else {
            optional_seen = true;
        }
        if let Some(width) = argument.repeating {
            if width == 0 || repeating.is_some() || index + 1 != schema.len() {
                return false;
            }
            repeating = Some(width);
        }
    }
    let represented_minimum = min_args.min(schema.len());
    if (strict_required_count && required_count != min_args)
        || (!strict_required_count && required_count < represented_minimum)
        || schema
            .iter()
            .take(represented_minimum)
            .any(|argument| !argument.required)
        || (!variadic && schema.len() > 1 && min_args > schema.len())
        || arity < min_args
    {
        return false;
    }
    if let Some(width) = repeating {
        if width > schema.len() {
            return false;
        }
        let fixed_prefix = schema.len() - width;
        return arity >= schema.len() && (arity - fixed_prefix).is_multiple_of(width);
    }
    if variadic {
        return true;
    }
    arity <= schema.len().max(min_args)
}

fn inspected<T>(
    issues: &mut Vec<SemanticConformanceIssue>,
    issue: SemanticConformanceIssue,
    inspect: impl FnOnce() -> T,
) -> Option<T> {
    match catch_unwind(AssertUnwindSafe(inspect)) {
        Ok(value) => Some(value),
        Err(_) => {
            issues.push(issue);
            None
        }
    }
}

fn check_capability(
    issues: &mut Vec<SemanticConformanceIssue>,
    capability: bool,
    semantic: bool,
    issue: SemanticConformanceIssue,
) {
    if capability != semantic {
        issues.push(issue);
    }
}

fn trusted_contract_from_caps(
    caps: FnCaps,
    precision: Option<crate::function_contract::FunctionDependencyContract>,
) -> FunctionSemanticContract {
    let mut contract = FunctionSemanticContract::trusted_builtin_default(precision);
    if caps.contains(FnCaps::DYNAMIC_DEPENDENCY) {
        contract.dependency = FunctionDependencySemantics::Dynamic;
    }
    if caps.contains(FnCaps::SHORT_CIRCUIT) {
        contract.evaluation = FunctionEvaluationSemantics::ShortCircuit;
    }
    contract.result = FunctionResultSemantics::from_capabilities(
        caps.contains(FnCaps::RETURNS_REFERENCE),
        caps.contains(FnCaps::MAY_SPILL),
    );
    if caps.contains(FnCaps::LOCAL_ENVIRONMENT) {
        contract.environment = FunctionEnvironmentSemantics::LocalBindings;
    }
    contract
}

const EXCEL_PREFIXES: &[&str] = &["_XLFN.", "_XLL.", "_XLWS."];

fn resolve_registered(
    state: &RegistryState,
    key: &RegistryKey,
) -> Option<(RegistryKey, RegistryEntry)> {
    if let Some(entry) = state.registrations.get(key) {
        return Some((key.clone(), entry.clone()));
    }
    let alias = state.aliases.get(key)?;
    state
        .registrations
        .get(&alias.target)
        .map(|entry| (alias.target.clone(), entry.clone()))
}

#[cfg(test)]
thread_local! {
    static RESOLUTION_WRITES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}

fn resolve_entry(ns: &str, name: &str) -> Option<(RegistryKey, RegistryEntry)> {
    #[cfg(test)]
    RESOLUTION_WRITES.with(|c| c.set(c.get() + 1));
    let ns = norm(ns);
    let normalized_name = norm(name);
    let key = (ns.clone(), normalized_name.clone());
    let mut state = REGISTRY
        .write()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    if let Some(entry) = resolve_registered(&state, &key) {
        return Some(entry);
    }
    let mut candidate = normalized_name.as_str();
    loop {
        let mut stripped_any = false;
        for prefix in EXCEL_PREFIXES {
            if let Some(rest) = candidate.strip_prefix(prefix) {
                candidate = rest;
                stripped_any = true;
                let stripped_key = (ns.clone(), candidate.to_string());
                if let Some((canonical, entry)) = resolve_registered(&state, &stripped_key) {
                    state.aliases.insert(
                        key.clone(),
                        AliasEntry {
                            target: canonical.clone(),
                            owner: Some((canonical.clone(), entry.generation)),
                        },
                    );
                    return Some((canonical, entry));
                }
                break;
            }
        }
        if !stripped_any {
            break;
        }
    }
    None
}

fn resolve_key_read_only(
    state: &RegistryState,
    key: &RegistryKey,
) -> Option<(RegistryKey, RegistryEntry)> {
    if let Some(entry) = resolve_registered(state, key) {
        return Some(entry);
    }
    let mut candidate = key.1.as_str();
    loop {
        let rest = EXCEL_PREFIXES
            .iter()
            .find_map(|prefix| candidate.strip_prefix(prefix))?;
        candidate = rest;
        let stripped = (key.0.clone(), candidate.to_string());
        if let Some(entry) = resolve_registered(state, &stripped) {
            return Some(entry);
        }
    }
}

fn resolve_entry_read_only(ns: &str, name: &str) -> Option<(RegistryKey, RegistryEntry)> {
    let state = REGISTRY
        .read()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    resolve_key_read_only(&state, &(norm(ns), norm(name)))
}

pub fn get(ns: &str, name: &str) -> Option<Arc<dyn Function>> {
    let key = (norm(ns), norm(name));
    {
        let state = REGISTRY
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let entry = state.registrations.get(&key).or_else(|| {
            let alias = state.aliases.get(&key)?;
            state.registrations.get(&alias.target)
        });
        if let Some(entry) = entry {
            return Some(Arc::clone(&entry.function));
        }
        if !EXCEL_PREFIXES
            .iter()
            .any(|prefix| key.1.starts_with(prefix))
        {
            return None;
        }
    }
    // Prefix misses may publish an alias. Recheck under the write lock, since a
    // registration or alias owner could have changed after releasing the read lock.
    resolve_entry(&key.0, &key.1).map(|(_, entry)| entry.function)
}

/// Read-only registry lookup for planning providers. Unlike [`get`], this does
/// not populate the global Excel-prefix alias cache.
#[doc(hidden)]
pub fn get_for_planning(ns: &str, name: &str) -> Option<Arc<dyn Function>> {
    resolve_entry_read_only(ns, name).map(|(_, entry)| entry.function)
}

pub(crate) struct GlobalRegistryFunctionProvider;

impl crate::traits::FunctionProvider for GlobalRegistryFunctionProvider {
    fn planning_semantic_revision(&self) -> Option<u64> {
        Some(0)
    }

    fn get_function(&self, ns: &str, name: &str) -> Option<Arc<dyn Function>> {
        get(ns, name)
    }

    fn get_function_for_planning(&self, ns: &str, name: &str) -> Option<Arc<dyn Function>> {
        get_for_planning(ns, name)
    }
}

#[derive(Clone)]
struct PlanningRegistration {
    canonical: RegistryKey,
    function: Arc<dyn Function>,
    generation: u64,
    trusted_builtin: bool,
}

/// Immutable registry view used by one formula-planning operation.
///
/// Function metadata is inspected before publication of the snapshot. Resolution
/// afterward is lock-free and never populates the global prefix-alias cache.
#[derive(Clone)]
pub(crate) struct RegistryPlanningSnapshot {
    epoch: u64,
    provider_revision: Option<u64>,
    requests: Arc<Vec<(String, String, usize)>>,
    functions: Arc<HashMap<RegistryKey, Arc<dyn Function>>>,
    capabilities: Arc<HashMap<RegistryKey, FnCaps>>,
    identities: Arc<HashMap<(String, String, usize), FunctionSemanticIdentity>>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum PlanningSnapshotError {
    RegistryChangedDuringCapture,
    ProviderRevisionUnavailable,
}

impl RegistryPlanningSnapshot {
    const CAPTURE_ATTEMPTS: usize = 16;

    pub(crate) fn capture_for_requests(
        runtime_provider: &dyn crate::traits::FunctionProvider,
        requests: impl IntoIterator<Item = (String, String, usize)>,
    ) -> Result<Self, PlanningSnapshotError> {
        crate::builtins::load_builtins();
        let mut requests: Vec<_> = requests.into_iter().collect();
        requests.sort();
        requests.dedup();
        Self::capture_with_hook(runtime_provider, &requests, Self::CAPTURE_ATTEMPTS, |_| {})
    }

    fn capture_with_hook(
        runtime_provider: &dyn crate::traits::FunctionProvider,
        requests: &[(String, String, usize)],
        attempts: usize,
        mut after_registry_copy: impl FnMut(usize),
    ) -> Result<Self, PlanningSnapshotError> {
        for attempt in 0..attempts {
            let provider_revision = if requests.is_empty() {
                None
            } else {
                Some(
                    runtime_provider
                        .planning_semantic_revision()
                        .ok_or(PlanningSnapshotError::ProviderRevisionUnavailable)?,
                )
            };
            let start_epoch = REGISTRY
                .read()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .semantic_epoch;
            let runtime_functions: HashMap<_, _> = requests
                .iter()
                .filter_map(|(namespace, name, _)| {
                    runtime_provider
                        .get_function_for_planning(namespace, name)
                        .map(|function| ((norm(namespace), norm(name)), function))
                })
                .collect();
            let (epoch, registrations) = {
                let state = REGISTRY
                    .read()
                    .unwrap_or_else(|poisoned| poisoned.into_inner());
                let registrations = requests
                    .iter()
                    .filter_map(|(namespace, name, arity)| {
                        let request_key = (norm(namespace), norm(name));
                        resolve_key_read_only(&state, &request_key).map(|(canonical, entry)| {
                            (
                                (request_key.0, request_key.1, *arity),
                                PlanningRegistration {
                                    canonical,
                                    function: entry.function,
                                    generation: entry.generation,
                                    trusted_builtin: entry.trusted_builtin,
                                },
                            )
                        })
                    })
                    .collect::<HashMap<_, _>>();
                (state.semantic_epoch, registrations)
            };

            after_registry_copy(attempt);

            let mut capabilities = HashMap::new();
            let mut identities = HashMap::new();
            for (namespace, name, arity) in requests {
                let request_key = (norm(namespace), norm(name));
                let Some(runtime) = runtime_functions.get(&request_key) else {
                    continue;
                };
                let Some(registration) =
                    registrations.get(&(request_key.0.clone(), request_key.1.clone(), *arity))
                else {
                    continue;
                };
                if !Arc::ptr_eq(runtime, &registration.function) {
                    continue;
                }
                let (semantics, identity_metadata) = inspect_semantics_with_identity_metadata(
                    &registration.function,
                    registration.trusted_builtin,
                    registration.generation,
                    *arity,
                );
                let Some(contract) = semantics.contract else {
                    continue;
                };
                let Some((caps, argument_by_ref)) = identity_metadata else {
                    continue;
                };
                capabilities.insert(request_key.clone(), caps);
                identities.insert(
                    (request_key.0, request_key.1, *arity),
                    FunctionSemanticIdentity {
                        namespace: registration.canonical.0.clone(),
                        canonical_name: registration.canonical.1.clone(),
                        generation: registration.generation,
                        caps,
                        contract,
                        argument_by_ref,
                    },
                );
            }

            for (key, function) in &runtime_functions {
                if !capabilities.contains_key(key)
                    && let Ok(caps) = catch_unwind(AssertUnwindSafe(|| function.caps()))
                {
                    capabilities.insert(key.clone(), caps);
                }
            }

            let provider_unchanged = provider_revision.is_none_or(|revision| {
                runtime_provider.planning_semantic_revision() == Some(revision)
            });
            // Validate the capture against the functions this snapshot actually
            // requested, not against the global registry epoch.
            //
            // Registering any function anywhere bumps `semantic_epoch`, so a
            // global equality check treated a completely unrelated registration
            // -- a different namespace, a different name -- as "the registry
            // changed underneath us". Under concurrent registration that
            // produced spurious `RegistryChangedDuringCapture` results and, in
            // authoritative mode, spurious FormulaPlane family fallbacks whose
            // reported reason blamed the provider.
            //
            // `semantic_changes_affect_requests_in_state` consults the change
            // log across the whole capture window [start_epoch, now] and only
            // reports a conflict when a *requested* key changed. It stays
            // conservative when the bounded change log has been truncated past
            // `start_epoch`, so precision never costs correctness.
            let requests_unchanged = {
                let state = REGISTRY
                    .read()
                    .unwrap_or_else(|poisoned| poisoned.into_inner());
                !semantic_changes_affect_requests_in_state(
                    &state,
                    start_epoch,
                    requests.iter().cloned(),
                )
            };
            let unchanged = requests_unchanged && provider_unchanged;
            if unchanged {
                return Ok(Self {
                    epoch,
                    provider_revision,
                    requests: Arc::new(requests.to_vec()),
                    functions: Arc::new(runtime_functions),
                    capabilities: Arc::new(capabilities),
                    identities: Arc::new(identities),
                });
            }
        }
        Err(PlanningSnapshotError::RegistryChangedDuringCapture)
    }

    pub(crate) fn epoch(&self) -> u64 {
        self.epoch
    }

    pub(crate) fn provider_revision(&self) -> Option<u64> {
        self.provider_revision
    }

    pub(crate) fn semantic_changes_affect_requests_since(&self, epoch: u64) -> bool {
        semantic_changes_affect_requests_since(epoch, self.requests.iter().cloned())
    }

    pub(crate) fn semantic_changes_affect_requests_since_guarded(
        &self,
        guard: &SemanticEpochReadGuard,
        epoch: u64,
    ) -> bool {
        guard.semantic_changes_affect_requests_since(epoch, self.requests.iter().cloned())
    }
}

pub(crate) fn semantic_changes_affect_requests_since(
    epoch: u64,
    requests: impl IntoIterator<Item = (String, String, usize)>,
) -> bool {
    let state = REGISTRY
        .read()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    semantic_changes_affect_requests_in_state(&state, epoch, requests)
}

fn semantic_changes_affect_requests_in_state(
    state: &RegistryState,
    epoch: u64,
    requests: impl IntoIterator<Item = (String, String, usize)>,
) -> bool {
    let changes = semantic_changes_since_in_state(state, epoch);
    if changes.epoch == epoch {
        return false;
    }
    if !changes.complete {
        return true;
    }
    let requests = requests
        .into_iter()
        .flat_map(|(namespace, name, _)| {
            let namespace = norm(namespace);
            let normalized = norm(name);
            let mut spellings = vec![(namespace.clone(), normalized.clone())];
            let mut stripped = normalized.as_str();
            while let Some(rest) = EXCEL_PREFIXES
                .iter()
                .find_map(|prefix| stripped.strip_prefix(prefix))
            {
                stripped = rest;
                spellings.push((namespace.clone(), stripped.to_string()));
            }
            spellings
        })
        .collect::<std::collections::BTreeSet<_>>();
    changes.keys.into_iter().any(|key| requests.contains(&key))
}

impl crate::traits::FunctionProvider for RegistryPlanningSnapshot {
    fn planning_semantic_revision(&self) -> Option<u64> {
        Some(self.provider_revision.unwrap_or(0))
    }

    fn get_function(&self, ns: &str, name: &str) -> Option<Arc<dyn Function>> {
        self.functions.get(&(norm(ns), norm(name))).cloned()
    }

    fn get_function_for_planning(&self, ns: &str, name: &str) -> Option<Arc<dyn Function>> {
        self.get_function(ns, name)
    }

    fn function_capabilities(&self, ns: &str, name: &str) -> Option<FnCaps> {
        self.capabilities.get(&(norm(ns), norm(name))).copied()
    }

    fn function_semantic_identity(
        &self,
        ns: &str,
        name: &str,
        arity: usize,
    ) -> Option<FunctionSemanticIdentity> {
        self.identities.get(&(norm(ns), norm(name), arity)).cloned()
    }
}

pub fn resolve(ns: &str, name: &str) -> Option<ResolvedFunction> {
    resolve_entry(ns, name).map(to_resolved)
}

pub fn resolve_with_epoch(ns: &str, name: &str) -> Option<(u64, ResolvedFunction)> {
    let key = (norm(ns), norm(name));
    let state = REGISTRY
        .read()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    resolve_registered(&state, &key).map(|entry| (state.semantic_epoch, to_resolved(entry)))
}
pub fn resolve_for_arity(ns: &str, name: &str, arity: usize) -> Option<ResolvedFunction> {
    resolve_entry(ns, name).map(|((namespace, canonical_name), entry)| {
        let semantics = {
            let cached = entry
                .semantics_by_arity
                .read()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .get(&arity)
                .cloned();
            cached.unwrap_or_else(|| {
                let inspected = inspect_semantics(
                    &entry.function,
                    entry.trusted_builtin,
                    entry.generation,
                    arity,
                );
                entry
                    .semantics_by_arity
                    .write()
                    .unwrap_or_else(|poisoned| poisoned.into_inner())
                    .entry(arity)
                    .or_insert_with(|| inspected.clone())
                    .clone()
            })
        };
        ResolvedFunction {
            semantics,
            namespace,
            canonical_name,
            function: entry.function,
        }
    })
}

pub(crate) fn resolve_semantic_identity<P: crate::traits::FunctionProvider + ?Sized>(
    provider: &P,
    ns: &str,
    name: &str,
    arity: usize,
) -> Option<FunctionSemanticIdentity> {
    let runtime = provider.get_function(ns, name)?;
    let resolved = resolve_for_arity(ns, name, arity)?;
    if !Arc::ptr_eq(&runtime, &resolved.function) {
        return None;
    }
    let contract = resolved.semantics.contract?;
    let argument_by_ref = catch_unwind(AssertUnwindSafe(|| {
        let schema = runtime.arg_schema();
        let repeating = schema.iter().find(|argument| argument.repeating.is_some());
        (0..arity)
            .map(|index| {
                schema
                    .get(index)
                    .or(repeating)
                    .is_some_and(|argument| argument.by_ref)
            })
            .collect()
    }))
    .ok()?;
    Some(FunctionSemanticIdentity {
        namespace: resolved.namespace,
        canonical_name: resolved.canonical_name,
        generation: resolved.semantics.generation,
        caps: runtime.caps(),
        contract,
        argument_by_ref,
    })
}

fn to_resolved(
    ((namespace, canonical_name), entry): (RegistryKey, RegistryEntry),
) -> ResolvedFunction {
    ResolvedFunction {
        namespace,
        canonical_name,
        function: entry.function,
        semantics: entry.semantics,
    }
}

pub fn register_alias(ns: &str, alias: &str, target_ns: &str, target_name: &str) {
    let mut state = REGISTRY
        .write()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    let alias_key = (norm(ns), norm(alias));
    let target = (norm(target_ns), norm(target_name));
    let old_target = state
        .aliases
        .get(&alias_key)
        .map(|entry| entry.target.clone());
    if old_target.as_ref() == Some(&target) {
        return;
    }
    state.aliases.insert(
        alias_key.clone(),
        AliasEntry {
            target: target.clone(),
            owner: None,
        },
    );
    // Retargeting changes resolution through the alias spelling, not resolution
    // of either directly named target.
    publish_semantic_change(&mut state, [alias_key]);
}

pub fn snapshot_registered() -> Vec<(String, String, Arc<dyn Function>)> {
    let state = REGISTRY
        .read()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    state
        .registrations
        .iter()
        .map(|((ns, name), entry)| (ns.clone(), name.clone(), Arc::clone(&entry.function)))
        .collect()
}
pub fn snapshot_semantics() -> Vec<ResolvedFunction> {
    let state = REGISTRY
        .read()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    state
        .registrations
        .iter()
        .map(|((namespace, canonical_name), entry)| ResolvedFunction {
            namespace: namespace.clone(),
            canonical_name: canonical_name.clone(),
            function: Arc::clone(&entry.function),
            semantics: entry.semantics.clone(),
        })
        .collect()
}

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use crate::traits::FunctionProvider;

    struct TestFn {
        ns: &'static str,
        name: &'static str,
        aliases: &'static [&'static str],
    }

    impl Function for TestFn {
        fn name(&self) -> &'static str {
            self.name
        }
        fn namespace(&self) -> &'static str {
            self.ns
        }
        fn aliases(&self) -> &'static [&'static str] {
            self.aliases
        }
        fn eval<'a, 'b, 'c>(
            &self,
            _args: &'c [crate::traits::ArgumentHandle<'a, 'b>],
            _ctx: &dyn crate::traits::FunctionContext<'b>,
        ) -> Result<crate::traits::CalcValue<'b>, formualizer_common::ExcelError> {
            Ok(crate::traits::CalcValue::Scalar(
                formualizer_common::LiteralValue::Number(1.0),
            ))
        }
    }

    struct PlanningFn {
        ns: &'static str,
        name: &'static str,
        aliases: &'static [&'static str],
        caps: FnCaps,
    }

    impl Function for PlanningFn {
        fn name(&self) -> &'static str {
            self.name
        }
        fn namespace(&self) -> &'static str {
            self.ns
        }
        fn aliases(&self) -> &'static [&'static str] {
            self.aliases
        }
        fn caps(&self) -> FnCaps {
            self.caps
        }
        fn min_args(&self) -> usize {
            1
        }
        fn variadic(&self) -> bool {
            true
        }
        fn arg_schema(&self) -> &'static [crate::args::ArgSchema] {
            static SCHEMA: std::sync::LazyLock<Vec<crate::args::ArgSchema>> =
                std::sync::LazyLock::new(|| {
                    let mut argument = crate::args::ArgSchema::any();
                    argument.repeating = Some(1);
                    vec![argument]
                });
            &SCHEMA
        }
        fn eval<'a, 'b, 'c>(
            &self,
            _args: &'c [crate::traits::ArgumentHandle<'a, 'b>],
            _ctx: &dyn crate::traits::FunctionContext<'b>,
        ) -> Result<crate::traits::CalcValue<'b>, formualizer_common::ExcelError> {
            unreachable!()
        }
    }

    fn planning_fn(
        ns: &'static str,
        name: &'static str,
        aliases: &'static [&'static str],
        caps: FnCaps,
    ) -> Arc<dyn Function> {
        Arc::new(PlanningFn {
            ns,
            name,
            aliases,
            caps,
        })
    }

    #[test]
    fn runtime_hits_only_clone_current_function_without_resolution_writes() {
        let ns = "__RUNTIME_READ_FAST_PATH__";
        let first = planning_fn(ns, "TARGET", &["ALIAS"], FnCaps::empty());
        register_function(first.clone());
        assert!(Arc::ptr_eq(&get(ns, "_xlfn._xlws.alias").unwrap(), &first));
        RESOLUTION_WRITES.with(|c| c.set(0));
        for name in ["target", "ALIAS", "_XLFN._XLWS.ALIAS"] {
            assert!(Arc::ptr_eq(&get(ns, name).unwrap(), &first));
        }
        assert!(get(ns, "MISSING").is_none());
        assert_eq!(RESOLUTION_WRITES.with(|c| c.get()), 0);
        let second = planning_fn(ns, "TARGET", &["ALIAS"], FnCaps::empty());
        register_function(second.clone());
        for name in ["target", "ALIAS", "_XLFN._XLWS.ALIAS"] {
            assert!(Arc::ptr_eq(&get(ns, name).unwrap(), &second));
        }
        std::thread::scope(|scope| {
            for _ in 0..8 {
                let second = &second;
                scope.spawn(move || {
                    RESOLUTION_WRITES.with(|c| c.set(0));
                    for _ in 0..1000 {
                        assert!(Arc::ptr_eq(&get(ns, "TARGET").unwrap(), second));
                        assert!(Arc::ptr_eq(&get(ns, "_XLFN._XLWS.ALIAS").unwrap(), second));
                    }
                    assert_eq!(RESOLUTION_WRITES.with(|c| c.get()), 0);
                });
            }
        });
    }

    #[test]
    fn planning_snapshot_resolves_direct_alias_namespace_and_prefix_without_cache_mutation() {
        let ns = "__PLANNING_PARITY__";
        register_builtin(planning_fn(ns, "TARGET", &["ALIAS"], FnCaps::empty()));
        let requests = [
            (ns.to_string(), "TARGET".to_string(), 1),
            (ns.to_string(), "alias".to_string(), 1),
            (ns.to_string(), "_xlfn._xlws.alias".to_string(), 1),
        ];
        let prefixed_key = (ns.to_string(), "_XLFN._XLWS.ALIAS".to_string());
        assert!(!REGISTRY.read().unwrap().aliases.contains_key(&prefixed_key));
        let snapshot = RegistryPlanningSnapshot::capture_for_requests(
            &GlobalRegistryFunctionProvider,
            requests,
        )
        .unwrap();
        assert!(!REGISTRY.read().unwrap().aliases.contains_key(&prefixed_key));

        let direct = snapshot
            .function_semantic_identity(ns, "TARGET", 1)
            .unwrap();
        for spelling in ["alias", "_xlfn._xlws.alias"] {
            let resolved = snapshot
                .function_semantic_identity(ns, spelling, 1)
                .unwrap();
            assert_eq!(resolved.namespace, ns);
            assert_eq!(resolved.canonical_name, "TARGET");
            assert_eq!(resolved.generation, direct.generation);
            assert!(Arc::ptr_eq(
                &snapshot.get_function(ns, spelling).unwrap(),
                &snapshot.get_function(ns, "TARGET").unwrap(),
            ));
        }
    }

    // FORM-000138 test coordination.
    //
    // `BUILTIN_DISPLACEMENT_LOCK` is held by every test that displaces a
    // builtin in the global (empty) namespace (only the FORM-000138 tests do)
    // and by every test that asserts builtin completeness or no-op call counts.
    // Tests that merely call `load_builtins`, and tests that displace their own
    // trusted fixtures in private namespaces, are not coordinated. Assertions
    // below therefore only rely on facts those tests cannot invalidate: the
    // monotonic displacement counter, the change log, per-thread call counts,
    // and "after `load_builtins` returns, every builtin key is trusted"
    // (which only a global-namespace displacement could break).
    pub(crate) static BUILTIN_DISPLACEMENT_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());

    pub(crate) fn lock_builtin_displacement() -> std::sync::MutexGuard<'static, ()> {
        BUILTIN_DISPLACEMENT_LOCK
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    /// Displacements of any trusted entry so far. Uncoordinated tests displace
    /// their own trusted fixtures in private namespaces, which conservatively
    /// forces one extra full builtin pass; call-count assertions allow for that.
    pub(crate) fn builtin_displacements() -> u64 {
        BUILTIN_LOAD.displacements.load(Ordering::Acquire)
    }

    thread_local! {
        static RECORDED_BUILTIN_KEYS: std::cell::RefCell<Option<std::collections::BTreeSet<RegistryKey>>> =
            const { std::cell::RefCell::new(None) };
    }

    pub(super) fn record_builtin_key(key: &RegistryKey) {
        RECORDED_BUILTIN_KEYS.with(|keys| {
            if let Some(keys) = keys.borrow_mut().as_mut() {
                keys.insert(key.clone());
            }
        });
    }

    /// Every key a full builtin pass registers, observed by running one on this
    /// thread (idempotent: already-trusted keys take the no-op fast path).
    fn expected_builtin_keys() -> std::collections::BTreeSet<RegistryKey> {
        RECORDED_BUILTIN_KEYS.with(|keys| *keys.borrow_mut() = Some(Default::default()));
        crate::builtins::register_all_builtins();
        let keys = RECORDED_BUILTIN_KEYS
            .with(|keys| keys.borrow_mut().take())
            .unwrap();
        assert!(keys.len() > 400, "{}", keys.len());
        assert!(keys.contains(&(String::new(), "SUM".to_string())));
        keys
    }

    /// Load until the shortcut reports loaded, then check that every expected
    /// builtin key is registered and trusted in one registry snapshot.
    fn assert_builtins_complete(expected: &std::collections::BTreeSet<RegistryKey>) {
        while !builtins_loaded() {
            crate::builtins::load_builtins();
        }
        let state = REGISTRY
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        for key in expected {
            let entry = state
                .registrations
                .get(key)
                .unwrap_or_else(|| panic!("builtin {key:?} missing"));
            assert!(entry.trusted_builtin, "builtin {key:?} not trusted");
        }
    }

    fn builtin_entry(name: &str) -> (Arc<dyn Function>, u64, bool) {
        let state = REGISTRY
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let entry = &state.registrations[&(String::new(), norm(name))];
        (
            Arc::clone(&entry.function),
            entry.generation,
            entry.trusted_builtin,
        )
    }

    // FORM-000138: shortcut bookkeeping over a local instance, so every
    // interleaving below is deterministic.
    #[test]
    fn builtin_load_state_interleavings() {
        // First load.
        let state = BuiltinLoadState::new(0);
        assert!(!state.loaded());
        let pass = state.begin_pass();
        state.finish_pass(pass);
        assert!(state.loaded());

        // Displacement after a completed pass reopens until the next pass.
        state.record_displacement();
        assert!(!state.loaded());
        let pass = state.begin_pass();
        state.finish_pass(pass);
        assert!(state.loaded());

        // Displacement during a pass: that pass does not count.
        let pass = state.begin_pass();
        state.record_displacement();
        state.finish_pass(pass);
        assert!(!state.loaded());

        // Overlapping passes: a stale pass finishing late does not hide a newer
        // completion, and a newer completion is not hidden by it.
        let stale = state.begin_pass();
        state.record_displacement();
        let fresh = state.begin_pass();
        state.finish_pass(fresh);
        assert!(state.loaded());
        state.finish_pass(stale);
        assert!(state.loaded());

        // A pass that never finishes (panicked) publishes nothing.
        state.record_displacement();
        let _abandoned = state.begin_pass();
        assert!(!state.loaded());
    }

    // FORM-000138: the displacement counter never wraps; exhaustion permanently
    // disables the shortcut.
    #[test]
    fn builtin_load_state_exhaustion_is_sticky() {
        let state = BuiltinLoadState::new(BUILTIN_DISPLACEMENTS_EXHAUSTED - 2);
        let pass = state.begin_pass();
        state.finish_pass(pass);
        assert!(state.loaded());
        state.record_displacement();
        assert!(!state.loaded());
        let pass = state.begin_pass();
        assert_eq!(pass, BUILTIN_DISPLACEMENTS_EXHAUSTED - 1);
        state.finish_pass(pass);
        assert!(state.loaded());

        state.record_displacement();
        assert_eq!(
            state.displacements.load(Ordering::Acquire),
            BUILTIN_DISPLACEMENTS_EXHAUSTED
        );
        assert!(!state.loaded());
        for _ in 0..3 {
            let pass = state.begin_pass();
            state.finish_pass(pass);
            assert!(!state.loaded());
            state.record_displacement();
            assert_eq!(
                state.displacements.load(Ordering::Acquire),
                BUILTIN_DISPLACEMENTS_EXHAUSTED
            );
        }
    }

    #[test]
    fn builtin_load_state_displacements_are_atomic_and_saturating() {
        for (initial, expected) in [(0, 8_000), (u64::MAX - 10, BUILTIN_DISPLACEMENTS_EXHAUSTED)] {
            let state = BuiltinLoadState::new(initial);
            std::thread::scope(|scope| {
                for _ in 0..8 {
                    let state = &state;
                    scope.spawn(move || {
                        for _ in 0..1_000 {
                            state.record_displacement();
                        }
                    });
                }
            });
            assert_eq!(state.displacements.load(Ordering::Acquire), expected);
        }
    }

    // FORM-000138: once loaded, `load_builtins` neither registers nor inspects
    // anything, and the registered builtins are complete and unchanged.
    #[test]
    fn repeated_builtin_loading_is_a_counted_no_op() {
        let _serial = lock_builtin_displacement();
        let expected = expected_builtin_keys();
        assert_builtins_complete(&expected);
        let (function, generation, _) = builtin_entry("SUM");
        let before = registration_call_counts();
        let displacements = builtin_displacements();
        for _ in 0..100 {
            crate::builtins::load_builtins();
        }
        let after = registration_call_counts();
        assert_eq!(after.load_builtins - before.load_builtins, 100);
        if builtin_displacements() == displacements {
            assert_eq!(after.register, before.register);
        }
        assert_eq!(after.inspect_semantics, before.inspect_semantics);
        let (current, current_generation, current_trusted) = builtin_entry("SUM");
        assert!(Arc::ptr_eq(&function, &current));
        assert_eq!(current_generation, generation);
        assert!(current_trusted);

        // A direct builtin re-registration takes the read-lock fast path: no
        // metadata inspection and the entry is untouched.
        let before = registration_call_counts();
        register_builtin(Arc::clone(&function));
        let after = registration_call_counts();
        assert_eq!(after.register - before.register, 1);
        assert_eq!(after.inspect_semantics, before.inspect_semantics);
        assert_eq!(builtin_entry("SUM").1, generation);
        assert_builtins_complete(&expected);
    }

    // FORM-000138: repeated planning snapshots (one per ordered-fallback
    // proposal in production) register nothing after the first load. Metadata
    // inspection is request-scoped: none for an empty request set, one per
    // requested function otherwise, never a builtin-set-sized pass.
    #[test]
    fn repeated_planning_snapshots_do_not_reregister_builtins() {
        let _serial = lock_builtin_displacement();
        let expected = expected_builtin_keys();
        assert_builtins_complete(&expected);
        let displacements = builtin_displacements();

        let before = registration_call_counts();
        for _ in 0..100 {
            RegistryPlanningSnapshot::capture_for_requests(
                &GlobalRegistryFunctionProvider,
                std::iter::empty(),
            )
            .unwrap();
        }
        let after = registration_call_counts();
        assert_eq!(after.load_builtins - before.load_builtins, 100);
        if builtin_displacements() == displacements {
            assert_eq!(after.register, before.register);
            assert_eq!(after.inspect_semantics, before.inspect_semantics);
        }

        let before = registration_call_counts();
        for _ in 0..100 {
            let snapshot = RegistryPlanningSnapshot::capture_for_requests(
                &GlobalRegistryFunctionProvider,
                [(String::new(), "SUM".to_string(), 2)],
            )
            .unwrap();
            assert!(snapshot.function_semantic_identity("", "SUM", 2).is_some());
        }
        let after = registration_call_counts();
        if builtin_displacements() == displacements {
            assert_eq!(after.register, before.register);
            assert_eq!(after.inspect_semantics - before.inspect_semantics, 100);
        }
    }

    // FORM-000138: a user registration that displaces a trusted builtin still
    // takes effect (counted displacement, published change), and once
    // `load_builtins` returns the builtin is trusted again, as before the
    // shortcut existed. Uncoordinated loaders may restore it earlier, so trust
    // is only asserted after our own load. Uses the same function object so
    // concurrent tests evaluating IMCOSH are unaffected.
    #[test]
    fn displaced_builtin_is_restored_by_next_load() {
        let _serial = lock_builtin_displacement();
        let expected = expected_builtin_keys();
        assert_builtins_complete(&expected);
        let (builtin, _, _) = builtin_entry("IMCOSH");

        let epoch = semantic_epoch();
        let displacements = builtin_displacements();
        register_function(Arc::clone(&builtin));
        assert!(builtin_displacements() > displacements);
        assert!(semantic_changes_affect_requests_since(
            epoch,
            [(String::new(), "IMCOSH".to_string(), 1)]
        ));

        crate::builtins::load_builtins();
        // The pass installs a fresh builtin object, so compare trust, not identity.
        assert!(builtin_entry("IMCOSH").2);
        assert_builtins_complete(&expected);

        // A same-name registration in another namespace is not a displacement.
        let displacements = builtin_displacements();
        let before = registration_call_counts();
        register_function(planning_fn(
            "__DISPLACE_OTHER__",
            "IMCOSH",
            &[],
            FnCaps::empty(),
        ));
        crate::builtins::load_builtins();
        if builtin_displacements() == displacements {
            assert_eq!(registration_call_counts().register, before.register + 1);
        }
    }

    // FORM-000138: registrations and builtin overrides racing repeated
    // `load_builtins` calls keep today's semantics: every user registration
    // lands and publishes, and loading afterwards leaves every builtin
    // registered and trusted.
    #[test]
    fn concurrent_loading_and_registration_preserve_semantics() {
        let _serial = lock_builtin_displacement();
        let expected = expected_builtin_keys();
        assert_builtins_complete(&expected);
        let (builtin, _, _) = builtin_entry("IMCOSH");
        let epoch = semantic_epoch();
        let barrier = Arc::new(std::sync::Barrier::new(4));
        let mut workers = Vec::new();
        for _ in 0..2 {
            let barrier = Arc::clone(&barrier);
            workers.push(std::thread::spawn(move || {
                barrier.wait();
                for _ in 0..200 {
                    crate::builtins::load_builtins();
                }
            }));
        }
        {
            let barrier = Arc::clone(&barrier);
            workers.push(std::thread::spawn(move || {
                barrier.wait();
                for index in 0..50 {
                    let name: &'static str = Box::leak(format!("F{index}").into_boxed_str());
                    register_function(planning_fn(
                        "__CONCURRENT_LOAD__",
                        name,
                        &[],
                        FnCaps::empty(),
                    ));
                }
            }));
        }
        {
            let barrier = Arc::clone(&barrier);
            let builtin = Arc::clone(&builtin);
            workers.push(std::thread::spawn(move || {
                barrier.wait();
                for _ in 0..20 {
                    register_function(Arc::clone(&builtin));
                    std::thread::yield_now();
                }
            }));
        }
        for worker in workers {
            worker.join().unwrap();
        }
        for index in 0..50 {
            assert!(get("__CONCURRENT_LOAD__", &format!("F{index}")).is_some());
        }
        assert!(semantic_epoch() >= epoch + 70);
        assert_builtins_complete(&expected);
        assert!(builtin_entry("IMCOSH").2);
    }

    // FORM-000138: the mirror equals the guarded epoch while a guard is held,
    // including while a writer is (possibly) waiting, and local-state log
    // advancement never touches it. No `load_builtins` here.
    #[test]
    fn lock_free_epoch_matches_guard() {
        let ns = "__LOCK_FREE_EPOCH__";
        register_function(planning_fn(ns, "TARGET", &[], FnCaps::empty()));
        let guard = semantic_epoch_read_guard();
        assert_eq!(semantic_epoch_lock_free(), guard.epoch());

        let mut local = RegistryState::default();
        for index in 0..=1_024 {
            advance_semantic_log(&mut local, [(String::new(), format!("LOCAL_{index}"))]);
        }
        assert_eq!(semantic_epoch_lock_free(), guard.epoch());

        let (started_tx, started_rx) = std::sync::mpsc::sync_channel(0);
        let writer = std::thread::spawn(move || {
            started_tx.send(()).unwrap();
            register_function(planning_fn(ns, "TARGET", &[], FnCaps::empty()));
        });
        started_rx.recv().unwrap();
        std::thread::yield_now();
        assert_eq!(semantic_epoch_lock_free(), guard.epoch());
        let guarded = guard.epoch();
        drop(guard);
        writer.join().unwrap();
        let mirrored = semantic_epoch_lock_free();
        assert!(mirrored > guarded);
        assert!(mirrored <= semantic_epoch());
    }

    // FORM-000138: the mirror never decreases under concurrent registration
    // and concurrent local-state log advancement.
    #[test]
    fn lock_free_epoch_never_decreases() {
        let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let observer = {
            let stop = Arc::clone(&stop);
            std::thread::spawn(move || {
                let mut last = semantic_epoch_lock_free();
                let mut samples = 0u64;
                while !stop.load(Ordering::Acquire) || samples < 1_000 {
                    let current = semantic_epoch_lock_free();
                    assert!(
                        current >= last,
                        "mirror moved backward: {last} -> {current}"
                    );
                    last = current;
                    samples += 1;
                }
                last
            })
        };
        let local = std::thread::spawn(|| {
            let mut state = RegistryState::default();
            for index in 0..5_000 {
                advance_semantic_log(&mut state, [(String::new(), format!("LOCAL_{index}"))]);
            }
        });
        let before = semantic_epoch_lock_free();
        for index in 0..200 {
            let name: &'static str = Box::leak(format!("M{index}").into_boxed_str());
            register_function(planning_fn(
                "__MIRROR_MONOTONIC__",
                name,
                &[],
                FnCaps::empty(),
            ));
        }
        local.join().unwrap();
        stop.store(true, Ordering::Release);
        let last = observer.join().unwrap();
        assert!(semantic_epoch_lock_free() >= before + 200);
        assert!(semantic_epoch_lock_free() >= last);
    }

    #[test]
    fn guarded_request_change_check_does_not_relock_behind_queued_writer() {
        let ns = "__GUARDED_REQUEST_CHANGE__";
        register_function(planning_fn(ns, "TARGET", &[], FnCaps::empty()));
        let snapshot = RegistryPlanningSnapshot::capture_for_requests(
            &GlobalRegistryFunctionProvider,
            [(ns.to_string(), "TARGET".to_string(), 1)],
        )
        .unwrap();
        register_function(planning_fn(ns, "TARGET", &[], FnCaps::empty()));

        let guard = semantic_epoch_read_guard();
        let (queued_tx, queued_rx) = std::sync::mpsc::sync_channel(0);
        let writer = std::thread::spawn(move || {
            assert!(REGISTRY.try_write().is_err());
            queued_tx.send(()).unwrap();
            let mut state = REGISTRY
                .write()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            publish_semantic_change(&mut state, [(ns.to_string(), "QUEUED_WRITER".to_string())]);
        });
        queued_rx.recv().unwrap();
        std::thread::yield_now();

        assert!(snapshot.semantic_changes_affect_requests_since_guarded(&guard, snapshot.epoch(),));
        drop(guard);
        writer.join().unwrap();
    }

    #[test]
    fn workbook_planning_fallback_does_not_populate_prefix_alias_cache() {
        let ns = "__PLANNING_WORKBOOK_PREFIX__";
        register_builtin(planning_fn(ns, "TARGET", &["ALIAS"], FnCaps::empty()));
        let prefixed_key = (ns.to_string(), "_XLFN.ALIAS".to_string());
        assert!(!REGISTRY.read().unwrap().aliases.contains_key(&prefixed_key));

        let snapshot = RegistryPlanningSnapshot::capture_for_requests(
            &crate::test_workbook::TestWorkbook::default(),
            [(ns.to_string(), "_xlfn.alias".to_string(), 1)],
        )
        .unwrap();

        assert!(
            snapshot
                .function_semantic_identity(ns, "_xlfn.alias", 1)
                .is_some()
        );
        assert!(!REGISTRY.read().unwrap().aliases.contains_key(&prefixed_key));
    }

    #[test]
    fn planning_snapshot_is_immutable_across_replacement() {
        let ns = "__PLANNING_IMMUTABLE__";
        register_builtin(planning_fn(ns, "TARGET", &["OLD_ALIAS"], FnCaps::empty()));
        let requests = [
            (ns.to_string(), "TARGET".to_string(), 1),
            (ns.to_string(), "OLD_ALIAS".to_string(), 1),
        ];
        let snapshot = RegistryPlanningSnapshot::capture_with_hook(
            &GlobalRegistryFunctionProvider,
            &requests,
            10_000,
            |_| {},
        )
        .unwrap();
        let old_function = snapshot.get_function(ns, "TARGET").unwrap();
        let old_identity = snapshot
            .function_semantic_identity(ns, "TARGET", 1)
            .unwrap();

        register_function(planning_fn(ns, "TARGET", &[], FnCaps::MAY_SPILL));
        let current = resolve_for_arity(ns, "TARGET", 1).unwrap();
        assert!(current.semantics.generation > old_identity.generation);
        assert!(!Arc::ptr_eq(&old_function, &current.function));
        assert_eq!(
            snapshot
                .function_semantic_identity(ns, "TARGET", 1)
                .unwrap(),
            old_identity
        );
        assert!(Arc::ptr_eq(
            &old_function,
            &snapshot.get_function(ns, "TARGET").unwrap(),
        ));
        assert!(get(ns, "OLD_ALIAS").is_none());
        assert_eq!(
            snapshot
                .function_semantic_identity(ns, "OLD_ALIAS", 1)
                .unwrap(),
            old_identity
        );
        assert!(Arc::ptr_eq(
            &old_function,
            &snapshot.get_function(ns, "OLD_ALIAS").unwrap(),
        ));
    }

    #[test]
    fn planning_snapshot_requires_explicit_side_effect_free_provider_opt_in() {
        struct RuntimeOnlyProvider(Arc<dyn Function>);
        impl FunctionProvider for RuntimeOnlyProvider {
            fn get_function(&self, _ns: &str, _name: &str) -> Option<Arc<dyn Function>> {
                Some(Arc::clone(&self.0))
            }
        }

        let ns = "__PLANNING_FAIL_CLOSED__";
        register_builtin(planning_fn(ns, "TARGET", &[], FnCaps::empty()));
        let result = RegistryPlanningSnapshot::capture_for_requests(
            &RuntimeOnlyProvider(planning_fn(ns, "TARGET", &[], FnCaps::empty())),
            [(ns.to_string(), "TARGET".to_string(), 1)],
        );
        assert_eq!(
            result.err(),
            Some(PlanningSnapshotError::ProviderRevisionUnavailable)
        );
    }

    #[test]
    fn planning_snapshot_preserves_runtime_override_without_global_semantics() {
        struct OverrideProvider(Arc<dyn Function>);
        impl FunctionProvider for OverrideProvider {
            fn planning_semantic_revision(&self) -> Option<u64> {
                Some(0)
            }
            fn get_function(&self, _ns: &str, _name: &str) -> Option<Arc<dyn Function>> {
                Some(Arc::clone(&self.0))
            }
            fn get_function_for_planning(
                &self,
                _ns: &str,
                _name: &str,
            ) -> Option<Arc<dyn Function>> {
                Some(Arc::clone(&self.0))
            }
        }

        let ns = "__PLANNING_OVERRIDE__";
        register_builtin(planning_fn(ns, "TARGET", &[], FnCaps::empty()));
        let global = get(ns, "TARGET").unwrap();
        let runtime = planning_fn(ns, "TARGET", &[], FnCaps::MAY_SPILL);
        let provider = OverrideProvider(Arc::clone(&runtime));
        let snapshot = RegistryPlanningSnapshot::capture_for_requests(
            &provider,
            [(ns.to_string(), "TARGET".to_string(), 1)],
        )
        .unwrap();

        let captured = snapshot.get_function(ns, "TARGET").unwrap();
        assert!(Arc::ptr_eq(&captured, &runtime));
        assert!(!Arc::ptr_eq(&captured, &global));
        assert!(
            snapshot
                .function_semantic_identity(ns, "TARGET", 1)
                .is_none()
        );
        assert_eq!(snapshot.functions.len(), 1);
        assert_eq!(
            snapshot.function_capabilities(ns, "TARGET"),
            Some(FnCaps::MAY_SPILL)
        );
        assert_eq!(snapshot.capabilities.len(), 1);
        assert!(snapshot.identities.is_empty());
        assert!(snapshot.get_function(ns, "UNREQUESTED").is_none());
    }

    #[test]
    fn planning_snapshot_retries_provider_revision_flip_to_runtime_override() {
        struct FlippingProvider {
            function: Arc<RwLock<Arc<dyn Function>>>,
            revision: Arc<AtomicU64>,
        }
        impl FunctionProvider for FlippingProvider {
            fn planning_semantic_revision(&self) -> Option<u64> {
                Some(self.revision.load(Ordering::Acquire))
            }
            fn get_function(&self, _ns: &str, _name: &str) -> Option<Arc<dyn Function>> {
                Some(Arc::clone(&self.function.read().unwrap()))
            }
            fn get_function_for_planning(&self, ns: &str, name: &str) -> Option<Arc<dyn Function>> {
                self.get_function(ns, name)
            }
        }

        let ns = "__PLANNING_PROVIDER_FLIP__";
        register_builtin(planning_fn(ns, "TARGET", &[], FnCaps::empty()));
        let global = get(ns, "TARGET").unwrap();
        let override_function = planning_fn(ns, "TARGET", &[], FnCaps::MAY_SPILL);
        let function = Arc::new(RwLock::new(global));
        let revision = Arc::new(AtomicU64::new(0));
        let provider = FlippingProvider {
            function: Arc::clone(&function),
            revision: Arc::clone(&revision),
        };
        let requests = [(ns.to_string(), "TARGET".to_string(), 1)];
        let snapshot =
            RegistryPlanningSnapshot::capture_with_hook(&provider, &requests, 2, |attempt| {
                if attempt == 0 {
                    *function.write().unwrap() = Arc::clone(&override_function);
                    revision.fetch_add(1, Ordering::AcqRel);
                }
            })
            .unwrap();

        assert_eq!(snapshot.provider_revision(), Some(1));
        assert!(Arc::ptr_eq(
            &snapshot.get_function(ns, "TARGET").unwrap(),
            &override_function
        ));
        assert!(
            snapshot
                .function_semantic_identity(ns, "TARGET", 1)
                .is_none()
        );
    }

    #[test]
    fn planning_snapshot_capture_tolerates_unrelated_concurrent_registrations() {
        // Regression test for the flake in the provider-revision test family.
        //
        // A capture used to be validated against the global `semantic_epoch`,
        // so registering *any* function on another thread invalidated it. In
        // the test suite that surfaced as an intermittent failure that moved
        // between tests depending on which happened to be running in parallel;
        // in production it would surface as spurious FormulaPlane fallbacks in
        // any application that registers custom functions while evaluating.
        //
        // The churn below never touches the requested key, so a correct capture
        // must succeed even with the tightest useful attempt budget.
        // The capture hook fires between the registry copy and the validation
        // step -- exactly where a concurrent registration would land -- so this
        // reproduces the race deterministically rather than relying on thread
        // interleaving.
        let ns = "__PLANNING_UNRELATED_CHURN__";
        register_builtin(planning_fn(ns, "TARGET", &[], FnCaps::empty()));
        let requests = [(ns.to_string(), "TARGET".to_string(), 1)];

        let mut unrelated_registrations = 0usize;
        let snapshot = RegistryPlanningSnapshot::capture_with_hook(
            &GlobalRegistryFunctionProvider,
            &requests,
            2,
            |_| {
                // `register_function` is untrusted, so unlike a repeated builtin
                // registration it always publishes a semantic change. Firing on
                // every attempt means a global-epoch check can never converge.
                register_function(planning_fn(
                    "__PLANNING_UNRELATED_CHURN_OTHER__",
                    "OTHER",
                    &[],
                    FnCaps::empty(),
                ));
                unrelated_registrations += 1;
            },
        )
        .expect("unrelated registrations must not invalidate a planning snapshot capture");

        assert!(unrelated_registrations > 0, "hook must have registered");
        assert!(Arc::ptr_eq(
            &snapshot.get_function(ns, "TARGET").unwrap(),
            &get(ns, "TARGET").unwrap(),
        ));
    }

    #[test]
    fn planning_snapshot_capture_retries_and_fails_deterministically() {
        let ns = "__PLANNING_RACE__";
        register_builtin(planning_fn(ns, "TARGET", &[], FnCaps::empty()));
        let requests = [(ns.to_string(), "TARGET".to_string(), 1)];
        let before = semantic_epoch();
        let retried = RegistryPlanningSnapshot::capture_with_hook(
            &GlobalRegistryFunctionProvider,
            &requests,
            100,
            |attempt| {
                if attempt == 0 {
                    register_function(planning_fn(ns, "TARGET", &[], FnCaps::empty()));
                }
            },
        )
        .unwrap();
        assert!(retried.epoch() > before);

        let failed = RegistryPlanningSnapshot::capture_with_hook(
            &GlobalRegistryFunctionProvider,
            &requests,
            2,
            |_| register_function(planning_fn(ns, "TARGET", &[], FnCaps::empty())),
        );
        assert_eq!(
            failed.err(),
            Some(PlanningSnapshotError::RegistryChangedDuringCapture)
        );
    }

    #[test]
    fn planning_snapshot_nested_function_authority_matches_global_registry() {
        let ns = "";
        register_builtin(planning_fn(ns, "__PLAN_OUTER__", &[], FnCaps::empty()));
        register_builtin(planning_fn(ns, "__PLAN_INNER__", &[], FnCaps::empty()));
        let requests = [
            (String::new(), "__PLAN_OUTER__".to_string(), 1),
            (String::new(), "_xlfn.__PLAN_INNER__".to_string(), 1),
        ];
        let snapshot = RegistryPlanningSnapshot::capture_for_requests(
            &GlobalRegistryFunctionProvider,
            requests,
        )
        .unwrap();
        let ast =
            formualizer_parse::parser::parse("=__PLAN_OUTER__(_xlfn.__PLAN_INNER__(A1))").unwrap();
        let frozen = crate::engine::template::canonical::canonicalize_template_with_provider(
            &ast,
            2,
            2,
            Some(&snapshot),
        );
        let global = crate::engine::template::canonical::canonicalize_template_with_provider(
            &ast,
            2,
            2,
            Some(&GlobalRegistryFunctionProvider),
        );
        assert_eq!(frozen, global);
        assert!(frozen.labels.is_authority_supported());
    }

    #[test]
    fn parallel_snapshot_capture_and_prefix_resolution_does_not_deadlock() {
        let ns = "__PLANNING_PARALLEL__";
        register_builtin(planning_fn(ns, "TARGET", &["ALIAS"], FnCaps::empty()));
        let (send, receive) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            let mut workers = Vec::new();
            for worker in 0..4 {
                workers.push(std::thread::spawn(move || {
                    for iteration in 0..100 {
                        if worker == 0 && iteration % 10 == 0 {
                            register_function(planning_fn(
                                ns,
                                "TARGET",
                                &["ALIAS"],
                                FnCaps::empty(),
                            ));
                        }
                        let _ = RegistryPlanningSnapshot::capture_for_requests(
                            &GlobalRegistryFunctionProvider,
                            [(ns.to_string(), "_xlfn.alias".to_string(), 1)],
                        );
                        let _ = get(ns, "_xlfn.alias");
                    }
                }));
            }
            for worker in workers {
                worker.join().unwrap();
            }
            send.send(()).unwrap();
        });
        receive
            .recv_timeout(std::time::Duration::from_secs(10))
            .expect("parallel registry planning timed out (possible lock inversion)");
    }

    #[test]
    fn resolves_prefixes_aliases_and_direct_registration() {
        let ns = "__REG_PREFIX__";
        register_function(Arc::new(TestFn {
            ns,
            name: "FILTER",
            aliases: &["LEGACY"],
        }));
        assert_eq!(get(ns, "_xlfn._xlws.legacy").unwrap().name(), "FILTER");
        register_function(Arc::new(TestFn {
            ns,
            name: "_XLFN.FILTER",
            aliases: &[],
        }));
        assert_eq!(get(ns, "_xlfn.filter").unwrap().name(), "_XLFN.FILTER");
    }

    #[test]
    fn trusted_replacement_records_removed_owned_alias_spelling() {
        let namespace = "__REG_STALE_ALIAS__";
        register_builtin(Arc::new(TestFn {
            ns: namespace,
            name: "TARGET",
            aliases: &["STALE_OWNED_ALIAS"],
        }));
        let before = semantic_epoch();
        register_function(Arc::new(TestFn {
            ns: namespace,
            name: "TARGET",
            aliases: &["NEW_OWNED_ALIAS"],
        }));
        let changes = semantic_changes_since(before);
        assert!(
            changes
                .keys
                .contains(&(namespace.to_string(), "STALE_OWNED_ALIAS".to_string()))
        );
        assert!(
            changes
                .keys
                .contains(&(namespace.to_string(), "NEW_OWNED_ALIAS".to_string()))
        );
    }

    #[test]
    fn alias_requests_keep_the_spelling_used_by_the_formula() {
        let ns = "__ALIAS_REQUEST_SPELLING__";
        register_builtin(planning_fn(ns, "TARGET", &[], FnCaps::empty()));
        register_alias(ns, "FORMULA_ALIAS", ns, "TARGET");
        let request = (ns.to_string(), "FORMULA_ALIAS".to_string(), 1);
        let snapshot = RegistryPlanningSnapshot::capture_for_requests(
            &GlobalRegistryFunctionProvider,
            [request.clone()],
        )
        .unwrap();

        assert_eq!(snapshot.requests.as_ref(), &[request]);
        assert_eq!(
            snapshot
                .function_semantic_identity(ns, "FORMULA_ALIAS", 1)
                .unwrap()
                .canonical_name,
            "TARGET"
        );
    }

    #[test]
    fn alias_mutations_affect_alias_requests_but_not_direct_target_requests() {
        let ns = "__ALIAS_CHANGE_SCOPE__";
        register_builtin(planning_fn(ns, "OLD_TARGET", &[], FnCaps::empty()));
        register_builtin(planning_fn(ns, "NEW_TARGET", &[], FnCaps::empty()));

        let add_epoch = semantic_epoch();
        register_alias(ns, "ADDED_ALIAS", ns, "OLD_TARGET");
        assert!(semantic_changes_affect_requests_since(
            add_epoch,
            [(ns.to_string(), "ADDED_ALIAS".to_string(), 1)]
        ));
        assert!(!semantic_changes_affect_requests_since(
            add_epoch,
            [(ns.to_string(), "OLD_TARGET".to_string(), 1)]
        ));

        register_alias(ns, "RETARGETED_ALIAS", ns, "OLD_TARGET");
        let retarget_epoch = semantic_epoch();
        register_alias(ns, "RETARGETED_ALIAS", ns, "NEW_TARGET");
        assert!(semantic_changes_affect_requests_since(
            retarget_epoch,
            [(ns.to_string(), "RETARGETED_ALIAS".to_string(), 1)]
        ));
        assert!(semantic_changes_affect_requests_since(
            retarget_epoch,
            [(ns.to_string(), "_xlfn.RETARGETED_ALIAS".to_string(), 1)]
        ));
        for target in ["OLD_TARGET", "NEW_TARGET"] {
            assert!(!semantic_changes_affect_requests_since(
                retarget_epoch,
                [(ns.to_string(), target.to_string(), 1)]
            ));
        }

        register_alias(ns, "REMOVED_ALIAS", ns, "OLD_TARGET");
        let remove_epoch = semantic_epoch();
        {
            let mut state = REGISTRY
                .write()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            let alias_key = (ns.to_string(), "REMOVED_ALIAS".to_string());
            assert!(state.aliases.remove(&alias_key).is_some());
            publish_semantic_change(&mut state, [alias_key]);
        }
        assert!(semantic_changes_affect_requests_since(
            remove_epoch,
            [(ns.to_string(), "REMOVED_ALIAS".to_string(), 1)]
        ));
        assert!(!semantic_changes_affect_requests_since(
            remove_epoch,
            [(ns.to_string(), "OLD_TARGET".to_string(), 1)]
        ));
    }

    #[test]
    fn request_change_check_stays_conservative_after_log_truncation() {
        let mut state = RegistryState::default();
        let before = state.semantic_epoch;
        for index in 0..=1_024 {
            advance_semantic_log(&mut state, [(String::new(), format!("UNRELATED_{index}"))]);
        }

        assert!(semantic_changes_affect_requests_in_state(
            &state,
            before,
            [(String::new(), "TARGET".to_string(), 1)]
        ));
    }

    #[test]
    fn replacement_advances_semantic_generation() {
        let ns = "__REG_GENERATION__";
        register_function(Arc::new(TestFn {
            ns,
            name: "F",
            aliases: &[],
        }));
        let first = resolve(ns, "F").unwrap().semantics.generation;
        let epoch = semantic_epoch();
        register_function(Arc::new(TestFn {
            ns,
            name: "F",
            aliases: &[],
        }));
        let second = resolve(ns, "F").unwrap().semantics.generation;
        assert!(second > first);
        let changes = semantic_changes_since(epoch);
        assert!(changes.epoch > epoch);
        assert!(changes.keys.contains(&(ns.to_string(), "F".to_string())));
    }

    struct PanickingSchemaFn;

    impl Function for PanickingSchemaFn {
        fn name(&self) -> &'static str {
            "PANICKING_SCHEMA"
        }
        fn semantic_contract(&self, _arity: usize) -> Option<FunctionSemanticContract> {
            Some(FunctionSemanticContract::trusted_builtin_default(None))
        }
        fn arg_schema(&self) -> &'static [crate::args::ArgSchema] {
            panic!("bad schema")
        }
        fn eval<'a, 'b, 'c>(
            &self,
            _args: &'c [crate::traits::ArgumentHandle<'a, 'b>],
            _ctx: &dyn crate::traits::FunctionContext<'b>,
        ) -> Result<crate::traits::CalcValue<'b>, formualizer_common::ExcelError> {
            unreachable!()
        }
    }

    #[test]
    fn schema_panic_as_sole_defect_is_non_panicking_and_fails_closed() {
        register_function(Arc::new(PanickingSchemaFn));
        let semantics = resolve("", "PANICKING_SCHEMA").unwrap().semantics;
        assert!(semantics.contract.is_none());
        assert!(
            semantics
                .issues
                .contains(&SemanticConformanceIssue::ArgumentSchemaPanicked)
        );
        assert_eq!(
            semantics.issues,
            vec![SemanticConformanceIssue::ArgumentSchemaPanicked]
        );
        assert!(!semantics.conforms());
    }

    #[test]
    fn every_registered_builtin_has_a_conforming_semantic_contract() {
        crate::builtins::load_builtins();
        let builtins: Vec<_> = snapshot_semantics()
            .into_iter()
            .filter(|entry| entry.semantics.trusted_builtin)
            .collect();
        assert!(builtins.len() > 100);
        let rejected: Vec<_> = builtins
            .iter()
            .filter(|entry| !entry.semantics.conforms())
            .map(|entry| {
                (
                    &entry.namespace,
                    &entry.canonical_name,
                    &entry.semantics.issues,
                )
            })
            .collect();
        assert!(rejected.is_empty(), "non-conforming builtins: {rejected:?}");
    }

    #[test]
    fn semantic_contract_is_context_and_arity_aware() {
        crate::builtins::lookup::register_builtins();
        let row_without_arg = resolve_for_arity("", "ROW", 0).unwrap();
        let row_with_arg = resolve_for_arity("", "ROW", 1).unwrap();
        assert_eq!(
            row_without_arg.semantics.contract.unwrap().context,
            crate::function_contract::FunctionContextDependence::PlacementDependent
        );
        assert_eq!(
            row_with_arg.semantics.contract.unwrap().context,
            crate::function_contract::FunctionContextDependence::None
        );
    }

    #[test]
    fn semantic_identity_encodes_effective_by_reference_roles_for_call_arity() {
        crate::builtins::load_builtins();
        let provider = GlobalRegistryFunctionProvider;
        let sum = resolve_semantic_identity(&provider, "", "SUM", 3).unwrap();
        assert_eq!(sum.argument_by_ref, vec![false, false, false]);

        let row = resolve_semantic_identity(&provider, "", "ROW", 1).unwrap();
        assert_eq!(row.argument_by_ref, vec![true]);
    }

    struct ExplicitSafeCustomFn;

    impl Function for ExplicitSafeCustomFn {
        fn name(&self) -> &'static str {
            "EXPLICIT_SAFE_CUSTOM"
        }
        fn semantic_contract(&self, _arity: usize) -> Option<FunctionSemanticContract> {
            Some(FunctionSemanticContract::trusted_builtin_default(None))
        }
        fn eval<'a, 'b, 'c>(
            &self,
            _args: &'c [crate::traits::ArgumentHandle<'a, 'b>],
            _ctx: &dyn crate::traits::FunctionContext<'b>,
        ) -> Result<crate::traits::CalcValue<'b>, formualizer_common::ExcelError> {
            unreachable!()
        }
    }

    struct MismatchedPrecisionCustomFn;

    impl Function for MismatchedPrecisionCustomFn {
        fn name(&self) -> &'static str {
            "MISMATCHED_PRECISION_CUSTOM"
        }
        fn dependency_contract(
            &self,
            arity: usize,
        ) -> Option<crate::function_contract::FunctionDependencyContract> {
            crate::function_contract::FunctionDependencyContract::static_scalar_all_args(arity)
        }
        fn semantic_contract(&self, _arity: usize) -> Option<FunctionSemanticContract> {
            Some(FunctionSemanticContract::trusted_builtin_default(None))
        }
        fn min_args(&self) -> usize {
            1
        }
        fn arg_schema(&self) -> &'static [crate::args::ArgSchema] {
            static SCHEMA: std::sync::LazyLock<Vec<crate::args::ArgSchema>> =
                std::sync::LazyLock::new(|| vec![crate::args::ArgSchema::any()]);
            &SCHEMA
        }
        fn eval<'a, 'b, 'c>(
            &self,
            _args: &'c [crate::traits::ArgumentHandle<'a, 'b>],
            _ctx: &dyn crate::traits::FunctionContext<'b>,
        ) -> Result<crate::traits::CalcValue<'b>, formualizer_common::ExcelError> {
            unreachable!()
        }
    }

    #[test]
    fn explicit_precision_must_equal_dependency_contract() {
        register_function(Arc::new(MismatchedPrecisionCustomFn));
        let semantics = resolve_for_arity("", "MISMATCHED_PRECISION_CUSTOM", 1)
            .unwrap()
            .semantics;
        assert!(semantics.contract.is_none());
        assert!(
            semantics
                .issues
                .contains(&SemanticConformanceIssue::PrecisionContractMismatch)
        );
    }

    #[test]
    fn explicit_custom_semantics_can_conform_without_becoming_trusted() {
        register_function(Arc::new(ExplicitSafeCustomFn));
        let semantics = resolve_for_arity("", "EXPLICIT_SAFE_CUSTOM", 0)
            .unwrap()
            .semantics;
        assert!(!semantics.trusted_builtin);
        assert!(semantics.conforms());
    }

    #[test]
    fn concurrent_replacements_leave_only_final_owned_alias() {
        let ns = "__REG_CONCURRENT__";
        register_function(Arc::new(TestFn {
            ns,
            name: "TARGET",
            aliases: &["INITIAL"],
        }));
        let mut workers = Vec::new();
        for alias in ["A", "B", "C", "D"] {
            workers.push(std::thread::spawn(move || {
                for _ in 0..100 {
                    let aliases: &'static [&'static str] = Box::leak(Box::new([alias]));
                    register_function(Arc::new(TestFn {
                        ns,
                        name: "TARGET",
                        aliases,
                    }));
                    assert_eq!(get(ns, "TARGET").unwrap().name(), "TARGET");
                }
            }));
        }
        for worker in workers {
            worker.join().unwrap();
        }
        register_function(Arc::new(TestFn {
            ns,
            name: "TARGET",
            aliases: &["FINAL"],
        }));
        for stale in ["INITIAL", "A", "B", "C", "D"] {
            assert!(get(ns, stale).is_none());
        }
        assert!(get(ns, "FINAL").is_some());
    }

    #[test]
    fn independent_exception_inventory_matches_builtin_caps_and_context() {
        crate::builtins::load_builtins();
        for name in [
            "RAND",
            "RANDBETWEEN",
            "RANDARRAY",
            "TODAY",
            "NOW",
            "OFFSET",
            "INDIRECT",
        ] {
            assert!(
                get("", name).unwrap().caps().contains(FnCaps::VOLATILE),
                "{name}"
            );
        }
        for name in ["OFFSET", "INDIRECT"] {
            assert!(
                get("", name)
                    .unwrap()
                    .caps()
                    .contains(FnCaps::DYNAMIC_DEPENDENCY),
                "{name}"
            );
        }
        for name in ["INDEX", "OFFSET", "INDIRECT", "CHOOSE"] {
            assert!(
                get("", name)
                    .unwrap()
                    .caps()
                    .contains(FnCaps::RETURNS_REFERENCE),
                "{name}"
            );
        }
        for name in ["LET", "LAMBDA"] {
            assert!(
                get("", name)
                    .unwrap()
                    .caps()
                    .contains(FnCaps::LOCAL_ENVIRONMENT),
                "{name}"
            );
        }
        for name in [
            "IF",
            "IFERROR",
            "IFNA",
            "IFS",
            "SWITCH",
            "CHOOSE",
            "FILTER",
            "UNIQUE",
            "SEQUENCE",
            "TRANSPOSE",
            "TAKE",
            "DROP",
            "SORT",
            "SORTBY",
            "RANDARRAY",
            "HSTACK",
            "VSTACK",
            "TOCOL",
            "TOROW",
            "CHOOSECOLS",
            "CHOOSEROWS",
            "FREQUENCY",
            "LINEST",
            "TREND",
            "GROWTH",
            "LOGEST",
            "MODE.MULT",
            "TEXTSPLIT",
        ] {
            assert!(
                get("", name).unwrap().caps().contains(FnCaps::MAY_SPILL),
                "{name}"
            );
        }
        const SHORT_CIRCUIT: &[&str] = &[
            "IF", "IFERROR", "IFNA", "IFS", "SWITCH", "CHOOSE", "LET", "LAMBDA", "AND", "OR",
        ];
        let observed_short_circuit: std::collections::BTreeSet<_> = snapshot_registered()
            .into_iter()
            .filter(|(namespace, _, function)| {
                namespace.is_empty() && function.caps().contains(FnCaps::SHORT_CIRCUIT)
            })
            .map(|(_, name, _)| name)
            .collect();
        let expected_short_circuit: std::collections::BTreeSet<_> = SHORT_CIRCUIT
            .iter()
            .map(|name| (*name).to_string())
            .collect();
        assert_eq!(observed_short_circuit, expected_short_circuit);
        for name in SHORT_CIRCUIT {
            let contract = resolve_for_arity("", name, get("", name).unwrap().min_args())
                .unwrap()
                .semantics
                .contract
                .unwrap();
            assert_eq!(
                contract.evaluation,
                FunctionEvaluationSemantics::ShortCircuit,
                "{name}"
            );
        }
        assert_eq!(
            resolve_for_arity("", "CHOOSE", 2)
                .unwrap()
                .semantics
                .contract
                .unwrap()
                .result,
            FunctionResultSemantics::MayReturnReferenceAndSpill
        );
        for name in ["ROW", "COLUMN"] {
            let contract = resolve_for_arity("", name, 0)
                .unwrap()
                .semantics
                .contract
                .unwrap();
            assert_eq!(
                contract.context,
                crate::function_contract::FunctionContextDependence::PlacementDependent,
                "{name}"
            );
            assert_eq!(
                resolve_for_arity("", name, 1)
                    .unwrap()
                    .semantics
                    .contract
                    .unwrap()
                    .context,
                crate::function_contract::FunctionContextDependence::None,
                "{name} with argument"
            );
        }
        for name in ["CELL", "ISFORMULA", "FORMULATEXT", "SHEET", "SHEETS"] {
            let contract = resolve_for_arity("", name, get("", name).unwrap().min_args())
                .unwrap()
                .semantics
                .contract
                .unwrap();
            assert_eq!(
                contract.context,
                crate::function_contract::FunctionContextDependence::WorkbookMetadata,
                "{name}"
            );
        }
    }

    struct NamePanicFn;
    impl Function for NamePanicFn {
        fn name(&self) -> &'static str {
            panic!("name")
        }
        fn eval<'a, 'b, 'c>(
            &self,
            _args: &'c [crate::traits::ArgumentHandle<'a, 'b>],
            _ctx: &dyn crate::traits::FunctionContext<'b>,
        ) -> Result<crate::traits::CalcValue<'b>, formualizer_common::ExcelError> {
            unreachable!()
        }
    }
    struct NamespacePanicFn;
    impl Function for NamespacePanicFn {
        fn name(&self) -> &'static str {
            "NS_PANIC"
        }
        fn namespace(&self) -> &'static str {
            panic!("namespace")
        }
        fn eval<'a, 'b, 'c>(
            &self,
            _args: &'c [crate::traits::ArgumentHandle<'a, 'b>],
            _ctx: &dyn crate::traits::FunctionContext<'b>,
        ) -> Result<crate::traits::CalcValue<'b>, formualizer_common::ExcelError> {
            unreachable!()
        }
    }

    #[test]
    fn canonical_metadata_panics_decline_registration_without_unwinding() {
        assert_eq!(
            try_register_function(Arc::new(NamePanicFn)),
            Err(RegistrationError::NameMetadataPanicked)
        );
        assert_eq!(
            try_register_function(Arc::new(NamespacePanicFn)),
            Err(RegistrationError::NamespaceMetadataPanicked)
        );
        register_function(Arc::new(NamePanicFn));
        register_function(Arc::new(NamespacePanicFn));
        assert!(get("", "NS_PANIC").is_none());
    }

    #[derive(Clone, Copy)]
    enum BadSchemaKind {
        TooLarge,
        Repeating,
        MinDisagreement,
        RequiredCount,
        TooManyRequired,
        RepeatWidth,
    }
    struct BadSchemaFn {
        name: &'static str,
        kind: BadSchemaKind,
    }
    impl Function for BadSchemaFn {
        fn name(&self) -> &'static str {
            self.name
        }
        fn min_args(&self) -> usize {
            if matches!(
                self.kind,
                BadSchemaKind::MinDisagreement
                    | BadSchemaKind::RequiredCount
                    | BadSchemaKind::RepeatWidth
            ) {
                2
            } else {
                1
            }
        }
        fn variadic(&self) -> bool {
            matches!(self.kind, BadSchemaKind::RepeatWidth)
        }
        fn semantic_contract(&self, _arity: usize) -> Option<FunctionSemanticContract> {
            Some(FunctionSemanticContract::trusted_builtin_default(None))
        }
        fn arg_schema(&self) -> &'static [crate::args::ArgSchema] {
            static ONE: std::sync::LazyLock<Vec<crate::args::ArgSchema>> =
                std::sync::LazyLock::new(|| vec![crate::args::ArgSchema::any()]);
            static BAD_REPEAT: std::sync::LazyLock<Vec<crate::args::ArgSchema>> =
                std::sync::LazyLock::new(|| {
                    let mut arg = crate::args::ArgSchema::any();
                    arg.repeating = Some(0);
                    vec![arg]
                });
            static REQUIRED_COUNT: std::sync::LazyLock<Vec<crate::args::ArgSchema>> =
                std::sync::LazyLock::new(|| {
                    let mut optional = crate::args::ArgSchema::any();
                    optional.required = false;
                    vec![crate::args::ArgSchema::any(), optional]
                });
            static REPEAT_WIDTH: std::sync::LazyLock<Vec<crate::args::ArgSchema>> =
                std::sync::LazyLock::new(|| {
                    let first = crate::args::ArgSchema::any();
                    let mut second = crate::args::ArgSchema::any();
                    second.repeating = Some(2);
                    vec![first, second]
                });
            match self.kind {
                BadSchemaKind::TooLarge => &ONE,
                BadSchemaKind::Repeating => &BAD_REPEAT,
                BadSchemaKind::MinDisagreement => &[],
                BadSchemaKind::RequiredCount => &REQUIRED_COUNT,
                BadSchemaKind::TooManyRequired => &REPEAT_WIDTH,
                BadSchemaKind::RepeatWidth => &REPEAT_WIDTH,
            }
        }
        fn eval<'a, 'b, 'c>(
            &self,
            _args: &'c [crate::traits::ArgumentHandle<'a, 'b>],
            _ctx: &dyn crate::traits::FunctionContext<'b>,
        ) -> Result<crate::traits::CalcValue<'b>, formualizer_common::ExcelError> {
            unreachable!()
        }
    }

    #[test]
    fn malformed_arity_and_schema_contracts_fail_closed() {
        for (name, kind, arity) in [
            ("TOO_LARGE", BadSchemaKind::TooLarge, 2),
            ("BAD_REPEAT", BadSchemaKind::Repeating, 1),
            ("MIN_DISAGREEMENT", BadSchemaKind::MinDisagreement, 2),
            ("REQUIRED_COUNT", BadSchemaKind::RequiredCount, 2),
            ("TOO_MANY_REQUIRED", BadSchemaKind::TooManyRequired, 1),
            ("REPEAT_WIDTH", BadSchemaKind::RepeatWidth, 3),
        ] {
            register_function(Arc::new(BadSchemaFn { name, kind }));
            let semantics = resolve_for_arity("", name, arity).unwrap().semantics;
            assert!(semantics.contract.is_none(), "{name}");
            assert!(
                semantics
                    .issues
                    .contains(&SemanticConformanceIssue::AritySchemaMismatch),
                "{name}: {:?}",
                semantics.issues
            );
        }
    }

    #[test]
    fn valid_optional_and_width_n_repeating_schemas_conform() {
        let required = crate::args::ArgSchema::any();
        let mut optional = crate::args::ArgSchema::any();
        optional.required = false;
        assert!(schema_allows_arity(
            &[required.clone(), optional],
            1,
            false,
            2,
            true
        ));

        let mut repeat_end = crate::args::ArgSchema::any();
        repeat_end.repeating = Some(2);
        let repeating = [required, repeat_end];
        assert!(schema_allows_arity(&repeating, 2, true, 4, true));
        assert!(!schema_allows_arity(&repeating, 2, true, 3, true));
    }

    #[test]
    fn replacement_readers_observe_generation_and_epoch_atomically() {
        let ns = "__REG_SNAPSHOT_RACE__";
        register_builtin(Arc::new(TestFn {
            ns,
            name: "TARGET",
            aliases: &[],
        }));
        let (initial_epoch, initial) = resolve_with_epoch(ns, "TARGET").unwrap();
        let initial_generation = initial.semantics.generation;
        let barrier = Arc::new(std::sync::Barrier::new(5));
        let mut readers = Vec::new();
        for _ in 0..4 {
            let barrier = Arc::clone(&barrier);
            readers.push(std::thread::spawn(move || {
                barrier.wait();
                for _ in 0..1_000 {
                    let (epoch, resolved) = resolve_with_epoch(ns, "TARGET").unwrap();
                    if resolved.semantics.generation != initial_generation {
                        assert!(epoch > initial_epoch);
                    }
                }
            }));
        }
        barrier.wait();
        register_function(Arc::new(TestFn {
            ns,
            name: "TARGET",
            aliases: &[],
        }));
        for reader in readers {
            reader.join().unwrap();
        }
        let (epoch, resolved) = resolve_with_epoch(ns, "TARGET").unwrap();
        assert!(epoch > initial_epoch);
        assert!(resolved.semantics.generation > initial_generation);
    }
}