dialcache/

policy.rs

//! Static operation policy, sparse runtime overlays and per-invocation resolution.

use serde_json::{Map, Value};

use crate::limits::{DEFAULT_REMOTE_READ_TIMEOUT_MS, MAX_CACHE_TTL_SEC, MAX_DEADLINE_MS};
use crate::observe::DisabledReason;

/// Per-use-case shadow validation policy.
#[derive(Debug, Clone, PartialEq, Default)]
pub struct ShadowPolicy {
    /// Independent stable cohort percentage in `[0, 100]`. Omitted and zero disable shadow work.
    pub ramp: Option<f64>,
    /// Shorthand for the whole [`mismatch_logging`](Self::mismatch_logging)
    /// group: `true` stands for `value` on with [`MismatchDiff::Values`],
    /// `false` for both leaves off. Leaves set in `mismatch_logging` override
    /// what it expands to, before runtime overlays merge leaf by leaf.
    pub log_mismatches: Option<bool>,
    /// Content of the warning emitted per confirmed mismatch. Omitted leaves
    /// default off.
    pub mismatch_logging: Option<MismatchLogging>,
}

/// Content of the one warning emitted per confirmed shadow mismatch. Both
/// leaves default off, and the warning is emitted when either is on.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct MismatchLogging {
    /// Log the JSON of both compared values. Defaults to false.
    pub value: Option<bool>,
    /// How much of the structural difference to log. Defaults to
    /// [`MismatchDiff::Off`].
    pub diff: Option<MismatchDiff>,
}

/// How much of the structural difference a confirmed-mismatch warning logs.
/// The levels are ordered: each logs everything the previous one does.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Default)]
pub enum MismatchDiff {
    /// No difference fields.
    #[default]
    Off,
    /// Where the values differ: `mismatchPaths` and `mismatchPathCount`.
    Paths,
    /// The paths plus what each side holds there: `diffJson`.
    Values,
}

impl MismatchDiff {
    /// The level as the JSON configuration spells it: `off`, `paths` or `values`.
    pub fn as_str(self) -> &'static str {
        match self {
            MismatchDiff::Off => "off",
            MismatchDiff::Paths => "paths",
            MismatchDiff::Values => "values",
        }
    }

    fn from_json(value: &Value) -> Option<MismatchDiff> {
        match value.as_str()? {
            "off" => Some(MismatchDiff::Off),
            "paths" => Some(MismatchDiff::Paths),
            "values" => Some(MismatchDiff::Values),
            _ => None,
        }
    }
}

impl std::fmt::Display for MismatchDiff {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

/// Static caching policy of one operation. Omitted leaves use library
/// defaults or are overridden by a runtime overlay.
#[derive(Debug, Clone, PartialEq, Default)]
pub struct Policy {
    /// Memoize values for the outermost enabled scope. Defaults to false.
    pub request_local: Option<bool>,
    /// Share one execution across concurrent same-key callers. Defaults to true.
    pub coalesce: Option<bool>,
    /// Process-local TTL in whole seconds, `1..=MAX_CACHE_TTL_SEC`. Omitted disables the layer.
    pub local_ttl_sec: Option<u64>,
    /// Remote TTL in whole seconds. Omitted disables the layer.
    pub remote_ttl_sec: Option<u64>,
    /// Local serving cohort percentage; a configured TTL implies 100.
    pub local_ramp: Option<f64>,
    /// Remote serving cohort percentage; a configured TTL implies 100.
    pub remote_ramp: Option<f64>,
    /// Exclusive stale-recovery age ceiling in seconds; must exceed the remote TTL. Zero disables.
    pub stale_on_error_max_age_sec: Option<u64>,
    /// Remote read budget in milliseconds, `1..=MAX_DEADLINE_MS`.
    pub remote_read_timeout_ms: Option<u64>,
    /// Shadow validation cohort and warning policy. Omitted means no shadow work.
    pub shadow: Option<ShadowPolicy>,
}

impl Policy {
    /// Both serving layers enabled with the same TTL and full ramps.
    pub fn enabled(ttl_sec: u64) -> Self {
        Policy {
            local_ttl_sec: Some(ttl_sec),
            remote_ttl_sec: Some(ttl_sec),
            local_ramp: Some(100.0),
            remote_ramp: Some(100.0),
            ..Policy::default()
        }
    }

    /// The explicit kill switch: request memoization, recovery and shadow
    /// work off, both serving layers ramped to zero. As an overlay it disables
    /// every inherited path instead of relying on omission.
    pub fn disabled() -> Self {
        Policy {
            request_local: Some(false),
            stale_on_error_max_age_sec: Some(0),
            local_ramp: Some(0.0),
            remote_ramp: Some(0.0),
            shadow: Some(ShadowPolicy {
                ramp: Some(0.0),
                log_mismatches: None,
                mismatch_logging: Some(MismatchLogging {
                    value: Some(false),
                    diff: Some(MismatchDiff::Off),
                }),
            }),
            ..Policy::default()
        }
    }

    /// Set whether values are memoized for the outermost enabled scope.
    pub fn request_local(mut self, enabled: bool) -> Self {
        self.request_local = Some(enabled);
        self
    }

    /// Set whether concurrent same-key callers share one execution.
    pub fn coalesce(mut self, enabled: bool) -> Self {
        self.coalesce = Some(enabled);
        self
    }

    /// Set the process-local TTL in whole seconds, enabling the layer with
    /// an implied full ramp.
    pub fn local_ttl_sec(mut self, ttl_sec: u64) -> Self {
        self.local_ttl_sec = Some(ttl_sec);
        self
    }

    /// Set the remote TTL in whole seconds, enabling the layer with an
    /// implied full ramp.
    pub fn remote_ttl_sec(mut self, ttl_sec: u64) -> Self {
        self.remote_ttl_sec = Some(ttl_sec);
        self
    }

    /// Set the process-local serving cohort percentage, `0.0..=100.0`.
    pub fn local_ramp(mut self, ramp: f64) -> Self {
        self.local_ramp = Some(ramp);
        self
    }

    /// Set the remote serving cohort percentage, `0.0..=100.0`.
    pub fn remote_ramp(mut self, ramp: f64) -> Self {
        self.remote_ramp = Some(ramp);
        self
    }

    /// Set the stale-recovery age ceiling in seconds. Zero disables
    /// recovery; a positive value must exceed the remote TTL.
    pub fn stale_on_error_max_age_sec(mut self, seconds: u64) -> Self {
        self.stale_on_error_max_age_sec = Some(seconds);
        self
    }

    /// Set the remote read budget in milliseconds, overriding the instance default.
    pub fn remote_read_timeout_ms(mut self, ms: u64) -> Self {
        self.remote_read_timeout_ms = Some(ms);
        self
    }

    /// Set the shadow validation policy.
    pub fn shadow(mut self, shadow: ShadowPolicy) -> Self {
        self.shadow = Some(shadow);
        self
    }

    /// Parse the JSON-shaped TypeScript configuration
    /// (`ttlSec`, `ramp`, `shadow`, `requestLocal`, `coalesce`,
    /// `staleOnErrorMaxAgeSec`, `remoteReadTimeoutMs`). `null` or absent means an empty policy.
    ///
    /// A present leaf holding `null` is an invalid supplied value, not an
    /// omission: JSON has no way to spell JavaScript's `undefined`, so every
    /// member of an object counts as present.
    pub fn from_json(value: &Value) -> Result<Policy, PolicyError> {
        let mut policy = Policy::default();
        if value.is_null() {
            return Ok(policy);
        }
        let config = policy_object(value, "defaultConfig")?;
        reject_shadow_ramp(config)?;
        if let Some(layers) = config.get("ttlSec") {
            let layers = policy_object(layers, "ttlSec")?;
            for (layer, slot) in [
                ("local", &mut policy.local_ttl_sec),
                ("remote", &mut policy.remote_ttl_sec),
            ] {
                if let Some(leaf) = layers.get(layer) {
                    let seconds = ttl_sec(leaf)
                        .ok_or_else(|| PolicyError(format!("invalid static ttlSec.{layer}")))?;
                    *slot = Some(seconds);
                }
            }
        }
        if let Some(layers) = config.get("ramp") {
            let layers = policy_object(layers, "ramp")?;
            for (layer, slot) in [
                ("local", &mut policy.local_ramp),
                ("remote", &mut policy.remote_ramp),
            ] {
                if let Some(leaf) = layers.get(layer) {
                    let ramp = finite_range(leaf, 0.0, 100.0, false)
                        .ok_or_else(|| PolicyError(format!("invalid static ramp.{layer}")))?;
                    *slot = Some(ramp);
                }
            }
        }
        for (field, slot) in [
            ("requestLocal", &mut policy.request_local),
            ("coalesce", &mut policy.coalesce),
        ] {
            if let Some(leaf) = config.get(field) {
                let flag = leaf
                    .as_bool()
                    .ok_or_else(|| PolicyError(format!("{field} must be boolean")))?;
                *slot = Some(flag);
            }
        }
        if let Some(leaf) = config.get("staleOnErrorMaxAgeSec") {
            let seconds = finite_range(leaf, 0.0, MAX_CACHE_TTL_SEC as f64, true)
                .ok_or_else(|| PolicyError("invalid static staleOnErrorMaxAgeSec".to_string()))?;
            policy.stale_on_error_max_age_sec = Some(seconds as u64);
        }
        if let Some(leaf) = config.get("remoteReadTimeoutMs") {
            let ms = finite_range(leaf, 1.0, MAX_DEADLINE_MS as f64, true)
                .ok_or_else(|| PolicyError("invalid remoteReadTimeoutMs".to_string()))?;
            policy.remote_read_timeout_ms = Some(ms as u64);
        }
        if let Some(shadow) = config.get("shadow") {
            let shadow = policy_object(shadow, "shadow")?;
            let mut parsed = ShadowPolicy::default();
            if let Some(leaf) = shadow.get("ramp") {
                let ramp = finite_range(leaf, 0.0, 100.0, false)
                    .ok_or_else(|| PolicyError("invalid static shadow.ramp".to_string()))?;
                parsed.ramp = Some(ramp);
            }
            if let Some(leaf) = shadow.get("logMismatches") {
                let flag = leaf.as_bool().ok_or_else(|| {
                    PolicyError("shadow.logMismatches must be boolean".to_string())
                })?;
                parsed.log_mismatches = Some(flag);
            }
            if let Some(group) = shadow.get("mismatchLogging") {
                let group = policy_object(group, "shadow mismatchLogging config")?;
                let mut leaves = MismatchLogging::default();
                if let Some(leaf) = group.get("value") {
                    let flag = leaf.as_bool().ok_or_else(|| {
                        PolicyError("shadow.mismatchLogging.value must be boolean".to_string())
                    })?;
                    leaves.value = Some(flag);
                }
                if let Some(leaf) = group.get("diff") {
                    let level = MismatchDiff::from_json(leaf).ok_or_else(|| {
                        PolicyError("invalid static shadow.mismatchLogging.diff".to_string())
                    })?;
                    leaves.diff = Some(level);
                }
                parsed.mismatch_logging = Some(leaves);
            }
            policy.shadow = Some(parsed);
        }
        policy.validate()?;
        Ok(policy)
    }

    /// The JSON-shaped form of this policy, with omitted leaves absent.
    ///
    /// `ttlSec` and `ramp` are always objects (possibly empty). All other
    /// leaves, including `requestLocal` and `coalesce`, appear only when set,
    /// so converting this policy to a runtime overlay preserves inheritance.
    pub fn to_json(&self) -> Value {
        let mut ttl = Map::new();
        if let Some(seconds) = self.local_ttl_sec {
            ttl.insert("local".to_string(), Value::from(seconds));
        }
        if let Some(seconds) = self.remote_ttl_sec {
            ttl.insert("remote".to_string(), Value::from(seconds));
        }
        let mut ramp = Map::new();
        if let Some(value) = self.local_ramp {
            ramp.insert("local".to_string(), ramp_json(value));
        }
        if let Some(value) = self.remote_ramp {
            ramp.insert("remote".to_string(), ramp_json(value));
        }
        let mut object = Map::new();
        object.insert("ttlSec".to_string(), Value::Object(ttl));
        object.insert("ramp".to_string(), Value::Object(ramp));
        if let Some(enabled) = self.request_local {
            object.insert("requestLocal".to_string(), Value::Bool(enabled));
        }
        if let Some(enabled) = self.coalesce {
            object.insert("coalesce".to_string(), Value::Bool(enabled));
        }
        if let Some(seconds) = self.stale_on_error_max_age_sec {
            object.insert("staleOnErrorMaxAgeSec".to_string(), Value::from(seconds));
        }
        if let Some(ms) = self.remote_read_timeout_ms {
            object.insert("remoteReadTimeoutMs".to_string(), Value::from(ms));
        }
        if let Some(shadow) = &self.shadow {
            let mut leaves = Map::new();
            if let Some(value) = shadow.ramp {
                leaves.insert("ramp".to_string(), ramp_json(value));
            }
            if let Some(flag) = shadow.log_mismatches {
                leaves.insert("logMismatches".to_string(), Value::Bool(flag));
            }
            if let Some(group) = shadow.mismatch_logging {
                let mut written = Map::new();
                if let Some(flag) = group.value {
                    written.insert("value".to_string(), Value::Bool(flag));
                }
                if let Some(level) = group.diff {
                    written.insert("diff".to_string(), Value::from(level.as_str()));
                }
                leaves.insert("mismatchLogging".to_string(), Value::Object(written));
            }
            object.insert("shadow".to_string(), Value::Object(leaves));
        }
        Value::Object(object)
    }

    /// Reject statically invalid leaves.
    pub fn validate(&self) -> Result<(), PolicyError> {
        for ttl in [self.local_ttl_sec, self.remote_ttl_sec]
            .into_iter()
            .flatten()
        {
            if !(1..=MAX_CACHE_TTL_SEC).contains(&ttl) {
                return Err(PolicyError(
                    "static TTL must be whole seconds within 365 days".to_string(),
                ));
            }
        }
        for ramp in [self.local_ramp, self.remote_ramp].into_iter().flatten() {
            if !ramp_in_domain(ramp) {
                return Err(PolicyError(
                    "static ramp must be between zero and 100".to_string(),
                ));
            }
        }
        if let Some(age) = self.stale_on_error_max_age_sec {
            let exceeds_remote = match self.remote_ttl_sec {
                Some(remote) if remote > 0 => age > remote,
                _ => false,
            };
            if age > MAX_CACHE_TTL_SEC || (age > 0 && !exceeds_remote) {
                return Err(PolicyError(
                    "static recovery age must exceed a positive remote TTL".to_string(),
                ));
            }
        }
        if let Some(ms) = self.remote_read_timeout_ms {
            if !(1..=MAX_DEADLINE_MS).contains(&ms) {
                return Err(PolicyError("invalid remote read deadline".to_string()));
            }
        }
        if let Some(ramp) = self.shadow.as_ref().and_then(|shadow| shadow.ramp) {
            if !ramp_in_domain(ramp) {
                return Err(PolicyError(
                    "static shadow ramp must be between zero and 100".to_string(),
                ));
            }
        }
        Ok(())
    }
}

/// A sparse runtime overlay returned by a [`PolicyProvider`](crate::PolicyProvider).
///
/// Each present leaf replaces the operation leaf; omitted leaves inherit.
/// Leaves may hold invalid values on purpose: their consequences are defined
/// by the portable contract (an invalid TTL or ramp disables only that
/// layer, an invalid flag or read deadline bypasses caching for the call).
#[derive(Debug, Clone, PartialEq)]
pub struct RuntimePolicy(pub Value);

impl RuntimePolicy {
    /// Wrap a JSON overlay as received from a configuration source. Nothing
    /// is validated here; resolution judges every leaf per call.
    pub fn from_json(value: Value) -> Self {
        RuntimePolicy(value)
    }
}

impl From<Policy> for RuntimePolicy {
    fn from(policy: Policy) -> Self {
        RuntimePolicy(policy.to_json())
    }
}

impl From<Value> for RuntimePolicy {
    fn from(value: Value) -> Self {
        RuntimePolicy(value)
    }
}

/// Instance defaults consulted during resolution.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PolicyDefaults {
    /// Instance remote read budget in milliseconds, used when neither the
    /// operation nor the overlay sets one. Zero selects the library default
    /// of 50 ms.
    pub remote_read_timeout_ms: u64,
}

impl Default for PolicyDefaults {
    fn default() -> Self {
        PolicyDefaults {
            remote_read_timeout_ms: DEFAULT_REMOTE_READ_TIMEOUT_MS,
        }
    }
}

/// Why a whole invocation's policy could not be resolved.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
#[error("{0}")]
pub struct PolicyError(pub String);

/// One serving layer after resolution.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct ResolvedLayer {
    /// Whether the layer serves this key: configured and admitted by its ramp.
    pub enabled: bool,
    /// Why the layer is disabled, when it is.
    pub reason: Option<DisabledReason>,
    /// A valid TTL and ramp remain available when the ramp excluded this key.
    pub configured: bool,
    /// The effective TTL in milliseconds; zero when not configured.
    pub ttl_ms: u64,
    /// The effective cohort percentage; zero when not configured.
    pub ramp: f64,
}

impl ResolvedLayer {
    fn disabled(reason: DisabledReason) -> Self {
        ResolvedLayer {
            enabled: false,
            reason: Some(reason),
            configured: false,
            ttl_ms: 0,
            ramp: 0.0,
        }
    }
}

/// Shadow policy after resolution.
#[derive(Debug, Clone, Copy, PartialEq, Default)]
pub struct ResolvedShadow {
    /// Cohort selection only; admission still needs an eligible path, hook and capacity.
    pub enabled: bool,
    /// The shadow cohort percentage; zero when omitted or invalid.
    pub ramp: f64,
    /// Whether a confirmed-mismatch warning carries the JSON of both values:
    /// the effective `mismatchLogging.value` leaf after the shorthand
    /// expanded, `false` when omitted or invalid.
    pub mismatch_value: bool,
    /// How much of the difference a confirmed-mismatch warning carries: the
    /// effective `mismatchLogging.diff` leaf, [`MismatchDiff::Off`] when
    /// omitted or invalid.
    pub mismatch_diff: MismatchDiff,
    /// The shadow ramp leaf was present but invalid: shadow work is off and a
    /// `config_resolution` error is emitted on the remote layer whenever a
    /// shadow job would otherwise have been scheduled (a remote hit or a
    /// ramped-down remote).
    pub config_error: bool,
    /// A mismatch-logging leaf or the `logMismatches` shorthand was invalid
    /// and acts off. One `config_resolution` covers all of them, recorded only
    /// if a job is admitted.
    pub logging_config_error: bool,
}

impl ResolvedShadow {
    /// Whether a confirmed mismatch emits a warning: `value` is on or `diff`
    /// is above [`MismatchDiff::Off`].
    pub fn logs_mismatches(&self) -> bool {
        self.mismatch_value || self.mismatch_diff != MismatchDiff::Off
    }
}

/// The captured policy of one enabled invocation.
#[derive(Debug, Clone, PartialEq)]
pub struct ResolvedPolicy {
    /// Whether the call memoizes in the outermost enabled scope.
    pub request_local: bool,
    /// Whether the call joins or leads a single flight for its key.
    pub coalesce: bool,
    /// The process-local layer.
    pub local: ResolvedLayer,
    /// The remote layer.
    pub remote: ResolvedLayer,
    /// The effective remote read budget in milliseconds: overlay, then
    /// operation, then instance default.
    pub remote_read_timeout_ms: u64,
    /// Zero when recovery is off.
    pub stale_on_error_max_age_ms: u64,
    /// The recovery age leaf was present but invalid (not whole seconds, or
    /// not above the remote TTL): recovery is off and a `config_resolution`
    /// error is emitted on the remote layer.
    pub stale_on_error_config_error: bool,
    /// The shadow validation policy.
    pub shadow: ResolvedShadow,
}

/// Merge a static policy with a sparse overlay once, for one logical key.
///
/// A malformed container, boolean or read deadline fails the whole
/// invocation; TTL or ramp errors disable only that layer; an invalid
/// recovery or shadow option preserves ordinary serving.
pub fn resolve_policy(
    base: &Policy,
    overlay: Option<&RuntimePolicy>,
    logical_key: &str,
    defaults: PolicyDefaults,
) -> Result<ResolvedPolicy, PolicyError> {
    let mut remote_read_timeout_ms = defaults.remote_read_timeout_ms;
    if remote_read_timeout_ms == 0 {
        remote_read_timeout_ms = DEFAULT_REMOTE_READ_TIMEOUT_MS;
    }
    if remote_read_timeout_ms > MAX_DEADLINE_MS {
        return Err(PolicyError("invalid instance read deadline".to_string()));
    }
    base.validate()?;
    let mut merged = match base.to_json() {
        Value::Object(object) => object,
        _ => unreachable!("Policy::to_json always produces an object"),
    };
    // Defaults belong to resolution, not the sparse policy representation.
    // Present overlay values (including invalid nulls) still replace them.
    merged.entry("requestLocal").or_insert(Value::Bool(false));
    merged.entry("coalesce").or_insert(Value::Bool(true));
    if let Some(RuntimePolicy(overlay)) = overlay {
        if !overlay.is_null() {
            merge_overlay(&mut merged, overlay)?;
        }
    }

    let request_local = merged
        .get("requestLocal")
        .and_then(Value::as_bool)
        .ok_or_else(|| PolicyError("runtime requestLocal must be boolean".to_string()))?;
    let coalesce = merged
        .get("coalesce")
        .and_then(Value::as_bool)
        .ok_or_else(|| PolicyError("runtime coalesce must be boolean".to_string()))?;
    if let Some(leaf) = merged.get("remoteReadTimeoutMs") {
        let ms = finite_range(leaf, 1.0, MAX_DEADLINE_MS as f64, true)
            .ok_or_else(|| PolicyError("invalid runtime remoteReadTimeoutMs".to_string()))?;
        remote_read_timeout_ms = ms as u64;
    }

    let empty = Map::new();
    let ttls = merged
        .get("ttlSec")
        .and_then(Value::as_object)
        .unwrap_or(&empty);
    let ramps = merged
        .get("ramp")
        .and_then(Value::as_object)
        .unwrap_or(&empty);
    let local = resolve_layer(ttls, ramps, logical_key, "local");
    let remote = resolve_layer(ttls, ramps, logical_key, "remote");

    let mut stale_on_error_max_age_ms = 0;
    let mut stale_on_error_config_error = false;
    if let Some(age) = merged.get("staleOnErrorMaxAgeSec") {
        let numeric_zero = policy_number(age) == Some(0.0);
        if !remote.configured {
            stale_on_error_config_error =
                remote.reason == Some(DisabledReason::PolicyDisabled) && !numeric_zero;
        } else if !numeric_zero {
            match ttl_sec(age).map(|seconds| seconds * 1_000) {
                Some(ms) if ms > remote.ttl_ms => stale_on_error_max_age_ms = ms,
                _ => stale_on_error_config_error = true,
            }
        }
    }

    let mut shadow = ResolvedShadow::default();
    if let Some(leaves) = merged.get("shadow").and_then(Value::as_object) {
        if let Some(leaf) = leaves.get("ramp") {
            match finite_range(leaf, 0.0, 100.0, false) {
                Some(ramp) => {
                    shadow.ramp = ramp;
                    shadow.enabled = admits(ramp, logical_key, "shadow");
                }
                None => shadow.config_error = true,
            }
        }
        let logging = resolve_mismatch_logging(leaves);
        shadow.mismatch_value = logging.value;
        shadow.mismatch_diff = logging.diff;
        shadow.logging_config_error = logging.invalid;
    }

    Ok(ResolvedPolicy {
        request_local,
        coalesce,
        local,
        remote,
        remote_read_timeout_ms,
        stale_on_error_max_age_ms,
        stale_on_error_config_error,
        shadow,
    })
}

/// Apply a sparse runtime overlay to the static policy's JSON form. Every
/// present member counts, including `null`, so an explicit null leaf replaces
/// the operation leaf and is then judged as an invalid supplied value.
fn merge_overlay(merged: &mut Map<String, Value>, overlay: &Value) -> Result<(), PolicyError> {
    let overlay = policy_object(overlay, "runtime config")?;
    reject_shadow_ramp(overlay)?;
    for (name, leaves) in [
        ("ttlSec", &["local", "remote"][..]),
        ("ramp", &["local", "remote"][..]),
    ] {
        let Some(incoming) = overlay.get(name) else {
            continue;
        };
        let incoming = policy_object(incoming, name)?;
        let mut output = match merged.get(name) {
            Some(Value::Object(existing)) => existing.clone(),
            _ => Map::new(),
        };
        for leaf in leaves {
            if let Some(value) = incoming.get(*leaf) {
                output.insert((*leaf).to_string(), value.clone());
            }
        }
        merged.insert(name.to_string(), Value::Object(output));
    }
    if let Some(incoming) = overlay.get("shadow") {
        let incoming = policy_object(incoming, "shadow")?;
        let inherited = match merged.get("shadow") {
            Some(Value::Object(existing)) => existing.clone(),
            _ => Map::new(),
        };
        merged.insert(
            "shadow".to_string(),
            Value::Object(merge_shadow(&inherited, incoming)?),
        );
    }
    for name in [
        "requestLocal",
        "coalesce",
        "staleOnErrorMaxAgeSec",
        "remoteReadTimeoutMs",
    ] {
        if let Some(value) = overlay.get(name) {
            merged.insert(name.to_string(), value.clone());
        }
    }
    Ok(())
}

/// Merge one runtime `shadow` group over the inherited one. `ramp` merges as
/// a leaf. Each layer's `logMismatches` shorthand expands to the leaves it
/// stands for before the leaf-wise merge, so a runtime shorthand replaces both
/// inherited leaves while a leaf written beside it still wins within its own
/// layer. A non-object runtime `mismatchLogging` fails the merge.
fn merge_shadow(
    inherited: &Map<String, Value>,
    incoming: &Map<String, Value>,
) -> Result<Map<String, Value>, PolicyError> {
    let mut output = Map::new();
    if let Some(ramp) = incoming.get("ramp").or_else(|| inherited.get("ramp")) {
        output.insert("ramp".to_string(), ramp.clone());
    }
    if let Some(group) = incoming.get("mismatchLogging") {
        policy_object(group, "shadow mismatchLogging config")?;
    }
    let base = expand_mismatch_logging(inherited);
    let written = expand_mismatch_logging(incoming);
    // Only the runtime layer can carry a malformed shorthand: a static one is
    // a typed boolean. It stays raw, so resolution records the one
    // `config_resolution` it owes; its expanded leaves already won the merge.
    if written.invalid {
        if let Some(shorthand) = incoming.get("logMismatches") {
            output.insert("logMismatches".to_string(), shorthand.clone());
        }
    }
    let mut group = Map::new();
    if let Some(value) = written.value.or(base.value) {
        group.insert("value".to_string(), value);
    }
    if let Some(diff) = written.diff.or(base.diff) {
        group.insert("diff".to_string(), diff);
    }
    if !group.is_empty() {
        output.insert("mismatchLogging".to_string(), Value::Object(group));
    }
    Ok(output)
}

/// The mismatch-logging leaves one shadow group writes, before validation;
/// `None` is unwritten. `invalid` marks a malformed shorthand.
#[derive(Default)]
struct MismatchLeaves {
    value: Option<Value>,
    diff: Option<Value>,
    invalid: bool,
}

/// Expand one shadow group's `logMismatches` shorthand into the leaves it
/// stands for, `true` as value on with diff `values` and `false` as both off,
/// then let the leaves written in the same group's `mismatchLogging` override
/// them. A malformed shorthand expands to both leaves off and is marked
/// invalid.
fn expand_mismatch_logging(shadow: &Map<String, Value>) -> MismatchLeaves {
    let expanded = |value: bool, diff: MismatchDiff, invalid: bool| MismatchLeaves {
        value: Some(Value::Bool(value)),
        diff: Some(Value::from(diff.as_str())),
        invalid,
    };
    let mut leaves = match shadow.get("logMismatches") {
        None => MismatchLeaves::default(),
        Some(Value::Bool(true)) => expanded(true, MismatchDiff::Values, false),
        Some(Value::Bool(false)) => expanded(false, MismatchDiff::Off, false),
        Some(_) => expanded(false, MismatchDiff::Off, true),
    };
    if let Some(group) = shadow.get("mismatchLogging").and_then(Value::as_object) {
        if let Some(value) = group.get("value") {
            leaves.value = Some(value.clone());
        }
        if let Some(diff) = group.get("diff") {
            leaves.diff = Some(diff.clone());
        }
    }
    leaves
}

/// The effective mismatch-logging leaves of one merged shadow group.
struct MismatchLoggingResolution {
    value: bool,
    diff: MismatchDiff,
    invalid: bool,
}

/// Resolve each leaf independently: an omitted leaf is off, an invalid leaf
/// acts off, and `invalid` reports that the group owes one
/// `config_resolution`, however many of its inputs were invalid.
fn resolve_mismatch_logging(shadow: &Map<String, Value>) -> MismatchLoggingResolution {
    let leaves = expand_mismatch_logging(shadow);
    let value = match &leaves.value {
        None => Some(false),
        Some(leaf) => leaf.as_bool(),
    };
    let diff = match &leaves.diff {
        None => Some(MismatchDiff::Off),
        Some(leaf) => MismatchDiff::from_json(leaf),
    };
    MismatchLoggingResolution {
        value: value.unwrap_or(false),
        diff: diff.unwrap_or(MismatchDiff::Off),
        invalid: leaves.invalid || value.is_none() || diff.is_none(),
    }
}

fn resolve_layer(
    ttls: &Map<String, Value>,
    ramps: &Map<String, Value>,
    logical_key: &str,
    layer: &str,
) -> ResolvedLayer {
    let Some(ttl) = ttls.get(layer) else {
        return ResolvedLayer::disabled(DisabledReason::PolicyDisabled);
    };
    let Some(ttl_ms) = ttl_sec(ttl).map(|seconds| seconds * 1_000) else {
        return ResolvedLayer::disabled(DisabledReason::InvalidTtl);
    };
    let ramp = match ramps.get(layer) {
        None => 100.0,
        Some(leaf) => match finite_range(leaf, 0.0, 100.0, false) {
            Some(ramp) => ramp,
            None => return ResolvedLayer::disabled(DisabledReason::InvalidRamp),
        },
    };
    let enabled = admits(ramp, logical_key, layer);
    ResolvedLayer {
        enabled,
        reason: if enabled {
            None
        } else {
            Some(DisabledReason::RampedDown)
        },
        configured: true,
        ttl_ms,
        ramp,
    }
}

/// Cohort admission: a full ramp admits every key, a zero ramp none, and a
/// partial ramp admits keys whose stable sample is strictly below it.
fn admits(ramp: f64, logical_key: &str, discriminator: &str) -> bool {
    ramp >= 100.0 || (ramp > 0.0 && crate::identity::cohort(logical_key, discriminator) < ramp)
}

fn ramp_in_domain(ramp: f64) -> bool {
    ramp.is_finite() && (0.0..=100.0).contains(&ramp)
}

/// Spell a ramp the way JavaScript and Go's `encoding/json` do: an integral
/// value has no fraction (`100`, not `100.0`), so a static policy's JSON form
/// compares equal to the same policy parsed from text.
fn ramp_json(ramp: f64) -> Value {
    if ramp_in_domain(ramp) && ramp.trunc() == ramp {
        Value::from(ramp as u64)
    } else {
        Value::from(ramp)
    }
}

/// The numeric domain: JSON numbers only. Strings and booleans never count.
fn policy_number(value: &Value) -> Option<f64> {
    value.as_number().and_then(serde_json::Number::as_f64)
}

fn finite_range(value: &Value, min: f64, max: f64, integer: bool) -> Option<f64> {
    let n = policy_number(value)?;
    (n.is_finite() && n >= min && n <= max && (!integer || n.trunc() == n)).then_some(n)
}

/// A TTL-domain leaf in whole seconds.
fn ttl_sec(value: &Value) -> Option<u64> {
    finite_range(value, 1.0, MAX_CACHE_TTL_SEC as f64, true).map(|n| n as u64)
}

fn policy_object<'a>(value: &'a Value, name: &str) -> Result<&'a Map<String, Value>, PolicyError> {
    value
        .as_object()
        .ok_or_else(|| PolicyError(format!("DialCache {name} must be an object")))
}

fn reject_shadow_ramp(object: &Map<String, Value>) -> Result<(), PolicyError> {
    if object.contains_key("shadowRamp") {
        return Err(PolicyError(
            "shadowRamp was replaced by shadow.ramp".to_string(),
        ));
    }
    Ok(())
}

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

    const KEY: &str = "policy:item:one#lookup";
    const MAX_TTL: u64 = MAX_CACHE_TTL_SEC;

    fn resolve(base: &Policy, overlay: Option<Value>) -> Result<ResolvedPolicy, PolicyError> {
        let overlay = overlay.map(RuntimePolicy);
        resolve_policy(base, overlay.as_ref(), KEY, PolicyDefaults::default())
    }

    fn resolved(base: &Policy, overlay: Option<Value>) -> ResolvedPolicy {
        resolve(base, overlay).expect("policy resolves")
    }

    // Port of TestPolicySparseResolutionAndSnapshots.
    #[test]
    fn sparse_resolution_and_inheritance() {
        let base = Policy::from_json(&json!({
            "requestLocal": true, "coalesce": false,
            "ttlSec": {"local": 1.0, "remote": 2.0},
            "staleOnErrorMaxAgeSec": 5.0, "remoteReadTimeoutMs": 30.0,
            "shadow": {"ramp": 100.0, "logMismatches": true},
        }))
        .unwrap();
        let overlay = RuntimePolicy(json!({
            "ttlSec": {"remote": 4.0},
            "ramp": {"local": 0.0},
            "shadow": {"logMismatches": false},
        }));
        let r = resolve_policy(
            &base,
            Some(&overlay),
            KEY,
            PolicyDefaults {
                remote_read_timeout_ms: 20,
            },
        )
        .unwrap();
        assert!(r.request_local);
        assert!(!r.coalesce);
        assert!(!r.local.enabled);
        assert_eq!(r.local.reason, Some(DisabledReason::RampedDown));
        assert!(r.local.configured);
        assert_eq!(r.local.ttl_ms, 1_000);
        assert_eq!(r.local.ramp, 0.0);
        assert!(r.remote.enabled);
        assert_eq!(r.remote.reason, None);
        assert_eq!(r.remote.ttl_ms, 4_000);
        assert_eq!(r.remote.ramp, 100.0);
        assert_eq!(r.stale_on_error_max_age_ms, 5_000);
        assert!(!r.stale_on_error_config_error);
        assert_eq!(r.remote_read_timeout_ms, 30);
        assert!(r.shadow.enabled);
        assert_eq!(r.shadow.ramp, 100.0);
        assert!(!r.shadow.logs_mismatches());
        assert!(!r.shadow.config_error);
        assert!(!r.shadow.logging_config_error);

        // A clone is an independent snapshot of every leaf.
        let snapshot = base.clone();
        let mut mutated = base.clone();
        mutated.remote_read_timeout_ms = Some(99);
        mutated.shadow.as_mut().unwrap().ramp = Some(0.0);
        assert_eq!(snapshot.remote_read_timeout_ms, Some(30));
        assert_eq!(snapshot.shadow.as_ref().unwrap().ramp, Some(100.0));

        // A null provider reply inherits the whole operation policy.
        for overlay in [None, Some(Value::Null)] {
            let inherit = resolve_policy(
                &snapshot,
                overlay.map(RuntimePolicy).as_ref(),
                KEY,
                PolicyDefaults::default(),
            )
            .unwrap();
            assert!(inherit.local.enabled);
            assert!(inherit.remote.enabled);
            assert_eq!(inherit.remote_read_timeout_ms, 30);
            assert!(inherit.shadow.enabled);
            assert!(inherit.shadow.mismatch_value);
            assert_eq!(inherit.shadow.mismatch_diff, MismatchDiff::Values);
        }
    }

    // Port of TestPolicyStaticValidation.
    #[test]
    fn static_validation_rejects_malformed_configs() {
        let invalid = [
            json!(true),
            json!([]),
            json!("policy"),
            json!(1),
            json!({"shadowRamp": 1}),
            json!({"ttlSec": null}),
            json!({"ramp": null}),
            json!({"shadow": null}),
            json!({"ttlSec": []}),
            json!({"ttlSec": 60}),
            json!({"shadow": true}),
            json!({"requestLocal": null}),
            json!({"coalesce": null}),
            json!({"coalesce": "false"}),
            json!({"requestLocal": 1}),
            json!({"ttlSec": {"local": null}}),
            json!({"ttlSec": {"local": 0}}),
            json!({"ttlSec": {"local": -1}}),
            json!({"ttlSec": {"local": 1.5}}),
            json!({"ttlSec": {"local": "60"}}),
            json!({"ttlSec": {"remote": 31_536_001}}),
            json!({"ramp": {"remote": null}}),
            json!({"ramp": {"remote": "nan"}}),
            json!({"ramp": {"remote": -0.5}}),
            json!({"ramp": {"local": 100.5}}),
            json!({"ramp": {"local": true}}),
            json!({"remoteReadTimeoutMs": null}),
            json!({"remoteReadTimeoutMs": 0}),
            json!({"remoteReadTimeoutMs": 1.5}),
            json!({"remoteReadTimeoutMs": 2_147_483_648u64}),
            json!({"staleOnErrorMaxAgeSec": null}),
            json!({"staleOnErrorMaxAgeSec": -1}),
            json!({"staleOnErrorMaxAgeSec": 2.5}),
            json!({"staleOnErrorMaxAgeSec": 2}),
            json!({"ttlSec": {"remote": 2}, "staleOnErrorMaxAgeSec": 2}),
            json!({"ttlSec": {"remote": 2}, "staleOnErrorMaxAgeSec": 1}),
            json!({"ttlSec": {"local": 2}, "staleOnErrorMaxAgeSec": 3}),
            json!({"ttlSec": {"remote": 2}, "staleOnErrorMaxAgeSec": 31_536_001}),
            json!({"shadow": {"ramp": null}}),
            json!({"shadow": {"ramp": 101}}),
            json!({"shadow": {"ramp": "50"}}),
            json!({"shadow": {"logMismatches": null}}),
            json!({"shadow": {"logMismatches": 1}}),
            json!({"shadow": {"mismatchLogging": null}}),
            json!({"shadow": {"mismatchLogging": []}}),
            json!({"shadow": {"mismatchLogging": true}}),
            json!({"shadow": {"mismatchLogging": {"value": null}}}),
            json!({"shadow": {"mismatchLogging": {"value": "true"}}}),
            json!({"shadow": {"mismatchLogging": {"diff": null}}}),
            json!({"shadow": {"mismatchLogging": {"diff": "full"}}}),
            json!({"shadow": {"mismatchLogging": {"diff": true}}}),
            json!({"shadow": {"mismatchLogging": {"diff": "Paths"}}}),
        ];
        for config in invalid {
            assert!(
                Policy::from_json(&config).is_err(),
                "accepted invalid static config: {config}"
            );
        }

        let accepted = Policy::from_json(&json!({
            "ttlSec": {"remote": 31_536_000},
            "remoteReadTimeoutMs": 2_147_483_647u64,
        }))
        .unwrap();
        assert_eq!(accepted.remote_ttl_sec, Some(MAX_TTL));
        assert_eq!(accepted.remote_read_timeout_ms, Some(MAX_DEADLINE_MS));
        assert_eq!(accepted.local_ttl_sec, None);

        // Zero recovery age is valid without a remote TTL; a positive one needs to exceed it.
        assert!(Policy::from_json(&json!({"staleOnErrorMaxAgeSec": 0})).is_ok());
        assert!(
            Policy::from_json(&json!({"ttlSec": {"remote": 2}, "staleOnErrorMaxAgeSec": 3}))
                .is_ok()
        );
        // Integer-valued floats and unsigned integers share one numeric domain.
        let floats =
            Policy::from_json(&json!({"ttlSec": {"local": 60.0}, "ramp": {"local": 50}})).unwrap();
        assert_eq!(floats.local_ttl_sec, Some(60));
        assert_eq!(floats.local_ramp, Some(50.0));
        // Ramp boundaries are inclusive.
        assert!(Policy::from_json(
            &json!({"ramp": {"local": 0, "remote": 100}, "shadow": {"ramp": 0}})
        )
        .is_ok());

        // Direct validation mirrors the same domains without the parser.
        assert!(Policy::default().local_ttl_sec(0).validate().is_err());
        assert!(Policy::default()
            .local_ttl_sec(MAX_TTL + 1)
            .validate()
            .is_err());
        assert!(Policy::default().remote_ttl_sec(MAX_TTL).validate().is_ok());
        assert!(Policy::default().local_ramp(f64::NAN).validate().is_err());
        assert!(Policy::default()
            .remote_ramp(f64::INFINITY)
            .validate()
            .is_err());
        assert!(Policy::default().remote_ramp(100.0001).validate().is_err());
        assert!(Policy::default().local_ramp(-0.1).validate().is_err());
        assert!(Policy::default()
            .remote_read_timeout_ms(0)
            .validate()
            .is_err());
        assert!(Policy::default()
            .remote_read_timeout_ms(MAX_DEADLINE_MS + 1)
            .validate()
            .is_err());
        assert!(Policy::default()
            .remote_read_timeout_ms(MAX_DEADLINE_MS)
            .validate()
            .is_ok());
        assert!(Policy::default()
            .stale_on_error_max_age_sec(1)
            .validate()
            .is_err());
        assert!(Policy::default()
            .stale_on_error_max_age_sec(0)
            .validate()
            .is_ok());
        assert!(Policy::default()
            .remote_ttl_sec(2)
            .stale_on_error_max_age_sec(2)
            .validate()
            .is_err());
        assert!(Policy::default()
            .remote_ttl_sec(2)
            .stale_on_error_max_age_sec(3)
            .validate()
            .is_ok());
        assert!(Policy::default()
            .remote_ttl_sec(2)
            .stale_on_error_max_age_sec(MAX_TTL + 1)
            .validate()
            .is_err());
        let nan_shadow = ShadowPolicy {
            ramp: Some(f64::NAN),
            ..ShadowPolicy::default()
        };
        assert!(Policy::default().shadow(nan_shadow).validate().is_err());
        let high_shadow = ShadowPolicy {
            ramp: Some(100.5),
            ..ShadowPolicy::default()
        };
        assert!(Policy::default().shadow(high_shadow).validate().is_err());
        let flag_only = ShadowPolicy {
            log_mismatches: Some(true),
            mismatch_logging: Some(MismatchLogging {
                value: Some(false),
                diff: Some(MismatchDiff::Paths),
            }),
            ..ShadowPolicy::default()
        };
        assert!(Policy::default().shadow(flag_only).validate().is_ok());
        assert!(Policy::enabled(60).validate().is_ok());
        assert!(Policy::disabled().validate().is_ok());
    }

    #[test]
    fn from_json_treats_absent_and_null_wholes_as_empty() {
        assert_eq!(Policy::from_json(&Value::Null).unwrap(), Policy::default());
        assert_eq!(Policy::from_json(&json!({})).unwrap(), Policy::default());
        // Unknown members are ignored, as in the TypeScript constructor.
        assert_eq!(
            Policy::from_json(&json!({"unknown": 1})).unwrap(),
            Policy::default()
        );
        // Empty containers are valid and leave every leaf omitted.
        let empty = Policy::from_json(&json!({"ttlSec": {}, "ramp": {}, "shadow": {}})).unwrap();
        assert_eq!(empty.local_ttl_sec, None);
        assert_eq!(empty.shadow, Some(ShadowPolicy::default()));
    }

    #[test]
    fn from_json_round_trips_through_to_json() {
        let source = json!({
            "ttlSec": {"local": 5, "remote": 60},
            "ramp": {"local": 12.5, "remote": 100},
            "requestLocal": true,
            "coalesce": false,
            "staleOnErrorMaxAgeSec": 120,
            "remoteReadTimeoutMs": 25,
            "shadow": {
                "ramp": 3.5,
                "logMismatches": true,
                "mismatchLogging": {"value": false, "diff": "paths"},
            },
        });
        let policy = Policy::from_json(&source).unwrap();
        assert_eq!(
            policy,
            Policy {
                request_local: Some(true),
                coalesce: Some(false),
                local_ttl_sec: Some(5),
                remote_ttl_sec: Some(60),
                local_ramp: Some(12.5),
                remote_ramp: Some(100.0),
                stale_on_error_max_age_sec: Some(120),
                remote_read_timeout_ms: Some(25),
                shadow: Some(ShadowPolicy {
                    ramp: Some(3.5),
                    log_mismatches: Some(true),
                    mismatch_logging: Some(MismatchLogging {
                        value: Some(false),
                        diff: Some(MismatchDiff::Paths),
                    }),
                }),
            }
        );
        assert_eq!(policy.to_json(), source);
        assert_eq!(Policy::from_json(&policy.to_json()).unwrap(), policy);
    }

    #[test]
    fn to_json_matches_the_static_policy_map_shape() {
        // Empty policy: layer maps present but empty, all other leaves absent.
        assert_eq!(
            Policy::default().to_json(),
            json!({"ttlSec": {}, "ramp": {}})
        );
        // Only present leaves appear inside the layer maps; integral ramps spell without a fraction.
        assert_eq!(
            Policy::default()
                .local_ttl_sec(7)
                .remote_ramp(40.0)
                .to_json(),
            json!({"ttlSec": {"local": 7}, "ramp": {"remote": 40}})
        );
        assert_eq!(
            Policy::default().local_ramp(12.5).to_json(),
            json!({"ttlSec": {}, "ramp": {"local": 12.5}})
        );
        // An explicit flag value is emitted even when it equals the default.
        assert_eq!(
            Policy::default()
                .request_local(false)
                .coalesce(true)
                .to_json(),
            json!({"ttlSec": {}, "ramp": {}, "requestLocal": false, "coalesce": true})
        );
        // A present but empty shadow policy is an empty object.
        assert_eq!(
            Policy::default().shadow(ShadowPolicy::default()).to_json(),
            json!({"ttlSec": {}, "ramp": {}, "shadow": {}})
        );
        assert_eq!(
            Policy::default()
                .shadow(ShadowPolicy {
                    log_mismatches: Some(false),
                    ..ShadowPolicy::default()
                })
                .to_json(),
            json!({
                "ttlSec": {}, "ramp": {},
                "shadow": {"logMismatches": false},
            })
        );
        // Only the set mismatch-logging leaves appear in their group.
        assert_eq!(
            Policy::default()
                .shadow(ShadowPolicy {
                    mismatch_logging: Some(MismatchLogging {
                        value: None,
                        diff: Some(MismatchDiff::Values),
                    }),
                    ..ShadowPolicy::default()
                })
                .to_json(),
            json!({
                "ttlSec": {}, "ramp": {},
                "shadow": {"mismatchLogging": {"diff": "values"}},
            })
        );
        // Zero recovery age is a present leaf, not an omission.
        assert_eq!(
            Policy::default()
                .stale_on_error_max_age_sec(0)
                .remote_read_timeout_ms(75)
                .to_json(),
            json!({
                "ttlSec": {}, "ramp": {},
                "staleOnErrorMaxAgeSec": 0, "remoteReadTimeoutMs": 75,
            })
        );
        assert_eq!(
            Policy::enabled(60).to_json(),
            json!({
                "ttlSec": {"local": 60, "remote": 60},
                "ramp": {"local": 100, "remote": 100},
            })
        );
        assert_eq!(
            Policy::disabled().to_json(),
            json!({
                "ttlSec": {},
                "ramp": {"local": 0, "remote": 0},
                "requestLocal": false,
                "staleOnErrorMaxAgeSec": 0,
                "shadow": {"ramp": 0, "mismatchLogging": {"value": false, "diff": "off"}},
            })
        );
    }

    #[test]
    fn enabled_and_disabled_presets_resolve_as_documented() {
        let enabled = resolved(&Policy::enabled(60), None);
        assert!(!enabled.request_local);
        assert!(enabled.coalesce);
        assert!(enabled.local.enabled && enabled.remote.enabled);
        assert_eq!(
            (enabled.local.ttl_ms, enabled.remote.ttl_ms),
            (60_000, 60_000)
        );
        assert_eq!((enabled.local.ramp, enabled.remote.ramp), (100.0, 100.0));
        assert_eq!(
            enabled.remote_read_timeout_ms,
            DEFAULT_REMOTE_READ_TIMEOUT_MS
        );
        assert_eq!(enabled.stale_on_error_max_age_ms, 0);
        assert!(!enabled.stale_on_error_config_error);
        assert_eq!(enabled.shadow, ResolvedShadow::default());

        // Standing alone, the kill switch has no TTLs: both layers are simply not configured,
        // and its zero recovery age is not a configuration error.
        let alone = resolved(&Policy::disabled(), None);
        assert_eq!(alone.local.reason, Some(DisabledReason::PolicyDisabled));
        assert_eq!(alone.remote.reason, Some(DisabledReason::PolicyDisabled));
        assert!(!alone.stale_on_error_config_error);
        assert!(!alone.shadow.enabled && !alone.shadow.config_error);
        assert!(!alone.shadow.logs_mismatches() && !alone.shadow.logging_config_error);

        // As an overlay it ramps every inherited path to zero instead of relying on omission.
        let base = Policy::enabled(60)
            .request_local(true)
            .stale_on_error_max_age_sec(120)
            .shadow(ShadowPolicy {
                ramp: Some(100.0),
                log_mismatches: Some(true),
                mismatch_logging: Some(MismatchLogging {
                    value: Some(true),
                    diff: Some(MismatchDiff::Values),
                }),
            });
        let killed = resolved(&base, Some(RuntimePolicy::from(Policy::disabled()).0));
        assert!(!killed.request_local);
        assert!(killed.coalesce);
        for layer in [killed.local, killed.remote] {
            assert!(!layer.enabled);
            assert_eq!(layer.reason, Some(DisabledReason::RampedDown));
            assert!(layer.configured);
            assert_eq!(layer.ttl_ms, 60_000);
            assert_eq!(layer.ramp, 0.0);
        }
        assert_eq!(killed.stale_on_error_max_age_ms, 0);
        assert!(!killed.stale_on_error_config_error);
        assert_eq!(killed.shadow, ResolvedShadow::default());
    }

    // Port of TestRuntimePolicyFailureScopes.
    #[test]
    fn runtime_failure_scopes() {
        let base = Policy::default()
            .request_local(true)
            .local_ttl_sec(1)
            .remote_ttl_sec(2);
        for overlay in [
            json!(false),
            json!([]),
            json!("config"),
            json!(0),
            json!({"shadowRamp": 5}),
            json!({"ttlSec": null}),
            json!({"ttlSec": []}),
            json!({"ttlSec": 5}),
            json!({"ramp": null}),
            json!({"shadow": null}),
            json!({"shadow": "off"}),
            json!({"requestLocal": null}),
            json!({"requestLocal": "true"}),
            json!({"coalesce": 1}),
            json!({"coalesce": null}),
            json!({"remoteReadTimeoutMs": 0}),
            json!({"remoteReadTimeoutMs": null}),
            json!({"remoteReadTimeoutMs": 1.5}),
            json!({"remoteReadTimeoutMs": "30"}),
            json!({"remoteReadTimeoutMs": 2_147_483_648u64}),
        ] {
            assert!(
                resolve(&base, Some(overlay.clone())).is_err(),
                "accepted invalid invocation policy: {overlay}"
            );
        }

        struct Case {
            overlay: Value,
            local: Option<DisabledReason>,
            remote: Option<DisabledReason>,
            recovery_error: bool,
        }
        let invalid_ttl = Some(DisabledReason::InvalidTtl);
        let invalid_ramp = Some(DisabledReason::InvalidRamp);
        let cases = [
            Case {
                overlay: json!({"ttlSec": {"local": null}}),
                local: invalid_ttl,
                remote: None,
                recovery_error: false,
            },
            Case {
                overlay: json!({"ttlSec": {"local": "1"}}),
                local: invalid_ttl,
                remote: None,
                recovery_error: false,
            },
            Case {
                overlay: json!({"ttlSec": {"local": 0}}),
                local: invalid_ttl,
                remote: None,
                recovery_error: false,
            },
            Case {
                overlay: json!({"ttlSec": {"local": 1.5}}),
                local: invalid_ttl,
                remote: None,
                recovery_error: false,
            },
            Case {
                overlay: json!({"ttlSec": {"local": 31_536_001}}),
                local: invalid_ttl,
                remote: None,
                recovery_error: false,
            },
            Case {
                overlay: json!({"ramp": {"remote": true}}),
                local: None,
                remote: invalid_ramp,
                recovery_error: false,
            },
            Case {
                overlay: json!({"ramp": {"remote": null}}),
                local: None,
                remote: invalid_ramp,
                recovery_error: false,
            },
            Case {
                overlay: json!({"ramp": {"remote": 100.1}}),
                local: None,
                remote: invalid_ramp,
                recovery_error: false,
            },
            Case {
                overlay: json!({"ramp": {"local": -1}}),
                local: invalid_ramp,
                remote: None,
                recovery_error: false,
            },
            Case {
                overlay: json!({"staleOnErrorMaxAgeSec": null}),
                local: None,
                remote: None,
                recovery_error: true,
            },
            Case {
                overlay: json!({"staleOnErrorMaxAgeSec": 1}),
                local: None,
                remote: None,
                recovery_error: true,
            },
            Case {
                overlay: json!({"staleOnErrorMaxAgeSec": 2}),
                local: None,
                remote: None,
                recovery_error: true,
            },
            Case {
                overlay: json!({"staleOnErrorMaxAgeSec": "3"}),
                local: None,
                remote: None,
                recovery_error: true,
            },
            Case {
                overlay: json!({"staleOnErrorMaxAgeSec": 2.5}),
                local: None,
                remote: None,
                recovery_error: true,
            },
            Case {
                overlay: json!({"staleOnErrorMaxAgeSec": 0}),
                local: None,
                remote: None,
                recovery_error: false,
            },
            Case {
                overlay: json!({"staleOnErrorMaxAgeSec": 0.0}),
                local: None,
                remote: None,
                recovery_error: false,
            },
            Case {
                overlay: json!({"staleOnErrorMaxAgeSec": 3}),
                local: None,
                remote: None,
                recovery_error: false,
            },
            Case {
                overlay: json!({"ttlSec": {"remote": -1}, "staleOnErrorMaxAgeSec": -1}),
                local: None,
                remote: invalid_ttl,
                recovery_error: false,
            },
            Case {
                overlay: json!({"ramp": {"remote": "bad"}, "staleOnErrorMaxAgeSec": 5}),
                local: None,
                remote: invalid_ramp,
                recovery_error: false,
            },
        ];
        for case in cases {
            let r = resolved(&base, Some(case.overlay.clone()));
            assert!(r.request_local, "{}", case.overlay);
            assert!(r.coalesce, "{}", case.overlay);
            assert_eq!(
                r.local.reason, case.local,
                "local reason for {}",
                case.overlay
            );
            assert_eq!(
                r.remote.reason, case.remote,
                "remote reason for {}",
                case.overlay
            );
            assert_eq!(r.local.enabled, case.local.is_none(), "{}", case.overlay);
            assert_eq!(r.remote.enabled, case.remote.is_none(), "{}", case.overlay);
            assert_eq!(
                r.stale_on_error_config_error, case.recovery_error,
                "recovery for {}",
                case.overlay
            );
            if case.recovery_error {
                assert_eq!(r.stale_on_error_max_age_ms, 0, "{}", case.overlay);
            }
        }
        // A valid runtime age above the remote TTL enables recovery.
        assert_eq!(
            resolved(&base, Some(json!({"staleOnErrorMaxAgeSec": 3}))).stale_on_error_max_age_ms,
            3_000
        );

        // Recovery configured without any remote TTL is a diagnostic error on a disabled layer.
        let missing = resolved(
            &Policy::default(),
            Some(json!({"staleOnErrorMaxAgeSec": 3})),
        );
        assert!(missing.stale_on_error_config_error);
        assert_eq!(missing.remote.reason, Some(DisabledReason::PolicyDisabled));
        assert!(!missing.remote.configured);
        assert_eq!(missing.stale_on_error_max_age_ms, 0);
        // ... unless it is numeric zero.
        let zero = resolved(
            &Policy::default(),
            Some(json!({"staleOnErrorMaxAgeSec": 0})),
        );
        assert!(!zero.stale_on_error_config_error);
        // A non-numeric leaf on a disabled remote layer is still an error.
        let text = resolved(
            &Policy::default(),
            Some(json!({"staleOnErrorMaxAgeSec": "0"})),
        );
        assert!(text.stale_on_error_config_error);
    }

    // Port of TestRecoveryRetentionAndShadowDiagnosticsRemainIndependent.
    #[test]
    fn recovery_retention_and_shadow_diagnostics_remain_independent() {
        let base = Policy::default()
            .remote_ttl_sec(1)
            .remote_ramp(0.0)
            .stale_on_error_max_age_sec(86_400);
        let r = resolved(
            &base,
            Some(json!({"shadow": {"ramp": 100, "logMismatches": "invalid"}})),
        );
        assert!(!r.remote.enabled);
        assert!(r.remote.configured);
        assert_eq!(r.remote.reason, Some(DisabledReason::RampedDown));
        assert_eq!(r.stale_on_error_max_age_ms, 86_400_000);
        assert!(!r.stale_on_error_config_error);
        assert!(r.shadow.enabled);
        assert_eq!(r.shadow.ramp, 100.0);
        assert!(!r.shadow.config_error);
        assert!(r.shadow.logging_config_error);
        assert!(!r.shadow.logs_mismatches());

        let r = resolved(&base, Some(json!({"shadow": {"ramp": null}})));
        assert!(r.shadow.config_error);
        assert!(!r.shadow.enabled);
        assert_eq!(r.shadow.ramp, 0.0);
        assert!(r.remote.configured);
        assert_eq!(r.stale_on_error_max_age_ms, 86_400_000);

        for invalid in [json!("50"), json!(-1), json!(100.5), json!(true)] {
            let r = resolved(&base, Some(json!({"shadow": {"ramp": invalid}})));
            assert!(r.shadow.config_error, "{invalid}");
            assert!(!r.shadow.enabled, "{invalid}");
        }
        // Zero shadow ramp is valid and silently off.
        let off = resolved(&base, Some(json!({"shadow": {"ramp": 0}})));
        assert!(!off.shadow.enabled && !off.shadow.config_error);
        // A null logging flag is malformed and disables warnings without touching the cohort.
        let logging = resolved(
            &base,
            Some(json!({"shadow": {"ramp": 100, "logMismatches": null}})),
        );
        assert!(
            logging.shadow.enabled
                && logging.shadow.logging_config_error
                && !logging.shadow.logs_mismatches()
        );
    }

    #[test]
    fn overlay_semantics_follow_the_portable_table() {
        let base = Policy::enabled(10)
            .request_local(true)
            .coalesce(false)
            .stale_on_error_max_age_sec(30)
            .remote_read_timeout_ms(40)
            .shadow(ShadowPolicy {
                ramp: Some(100.0),
                log_mismatches: Some(true),
                mismatch_logging: None,
            });

        // Omitted leaves inherit the operation leaf; an empty overlay changes nothing.
        let inherited = resolved(&base, None);
        assert_eq!(resolved(&base, Some(json!({}))), inherited);
        assert_eq!(
            resolved(&base, Some(json!({"ttlSec": {}, "ramp": {}, "shadow": {}}))),
            inherited
        );
        assert_eq!(resolved(&base, Some(Value::Null)), inherited);
        assert!(inherited.request_local && !inherited.coalesce);
        assert_eq!(inherited.remote_read_timeout_ms, 40);
        assert_eq!(inherited.stale_on_error_max_age_ms, 30_000);
        assert!(inherited.shadow.enabled && inherited.shadow.mismatch_value);
        assert_eq!(inherited.shadow.mismatch_diff, MismatchDiff::Values);

        // A present leaf replaces the operation leaf, sparsely.
        let patched = resolved(
            &base,
            Some(json!({"ttlSec": {"local": 3}, "ramp": {"remote": 0}})),
        );
        assert_eq!(patched.local.ttl_ms, 3_000);
        assert!(patched.local.enabled);
        assert_eq!(patched.remote.ttl_ms, 10_000);
        assert_eq!(patched.remote.reason, Some(DisabledReason::RampedDown));
        assert!(patched.remote.configured);
        // Recovery still measures against the merged remote TTL.
        assert_eq!(patched.stale_on_error_max_age_ms, 30_000);

        // An explicit null leaf is an invalid supplied value that disables only that layer.
        let nulled = resolved(&base, Some(json!({"ttlSec": {"remote": null}})));
        assert!(nulled.local.enabled);
        assert_eq!(nulled.remote.reason, Some(DisabledReason::InvalidTtl));
        assert!(!nulled.remote.configured);
        // The pre-existing recovery age on a layer disabled by an invalid TTL is not a config error.
        assert!(!nulled.stale_on_error_config_error);
        assert_eq!(nulled.stale_on_error_max_age_ms, 0);

        // An invalid flag fails the whole invocation.
        assert!(resolve(&base, Some(json!({"coalesce": "yes"}))).is_err());
        assert!(resolve(&base, Some(json!({"coalesce": null}))).is_err());
        assert!(resolve(&base, Some(json!({"requestLocal": 0}))).is_err());
        // ... as does an invalid runtime read deadline.
        assert!(resolve(&base, Some(json!({"remoteReadTimeoutMs": -5}))).is_err());
        // A valid runtime flag or deadline wins over the operation.
        let flipped = resolved(
            &base,
            Some(json!({"coalesce": true, "requestLocal": false, "remoteReadTimeoutMs": 7})),
        );
        assert!(flipped.coalesce && !flipped.request_local);
        assert_eq!(flipped.remote_read_timeout_ms, 7);

        // An invalid optional recovery age preserves remote serving with a diagnostic error.
        let recovery = resolved(&base, Some(json!({"staleOnErrorMaxAgeSec": 10})));
        assert!(recovery.remote.enabled);
        assert_eq!(recovery.stale_on_error_max_age_ms, 0);
        assert!(recovery.stale_on_error_config_error);
        let recovery = resolved(&base, Some(json!({"staleOnErrorMaxAgeSec": "30"})));
        assert!(recovery.remote.enabled && recovery.stale_on_error_config_error);
        // Zero disables inherited recovery silently.
        let off = resolved(&base, Some(json!({"staleOnErrorMaxAgeSec": 0})));
        assert!(off.remote.enabled);
        assert_eq!(off.stale_on_error_max_age_ms, 0);
        assert!(!off.stale_on_error_config_error);
        // Raising the remote TTL above the inherited age invalidates the inherited recovery.
        let raised = resolved(&base, Some(json!({"ttlSec": {"remote": 30}})));
        assert!(raised.remote.enabled && raised.stale_on_error_config_error);
        assert_eq!(raised.stale_on_error_max_age_ms, 0);

        // Shadow leaves merge independently of serving leaves.
        let shadow = resolved(&base, Some(json!({"shadow": {"logMismatches": false}})));
        assert!(shadow.shadow.enabled && !shadow.shadow.logs_mismatches());
        let shadow = resolved(&base, Some(json!({"shadow": {"ramp": 0}})));
        assert!(
            !shadow.shadow.enabled
                && shadow.shadow.logs_mismatches()
                && !shadow.shadow.config_error
        );
        assert!(shadow.local.enabled && shadow.remote.enabled);
    }

    #[test]
    fn runtime_leaves_can_configure_paths_the_operation_omitted() {
        // Runtime TTLs enable layers the operation left off, with an implied full ramp.
        let r = resolved(
            &Policy::default(),
            Some(json!({"ttlSec": {"local": 5, "remote": 6}})),
        );
        assert!(r.local.enabled && r.remote.enabled);
        assert_eq!((r.local.ramp, r.remote.ramp), (100.0, 100.0));
        assert_eq!((r.local.ttl_ms, r.remote.ttl_ms), (5_000, 6_000));
        // A ramp without a TTL leaves the layer policy-disabled.
        let r = resolved(&Policy::default(), Some(json!({"ramp": {"local": 100}})));
        assert_eq!(r.local.reason, Some(DisabledReason::PolicyDisabled));
        assert!(!r.local.configured);
        // Runtime shadow and recovery leaves attach to a runtime remote layer.
        let r = resolved(
            &Policy::default(),
            Some(
                json!({"ttlSec": {"remote": 6}, "staleOnErrorMaxAgeSec": 7, "shadow": {"ramp": 100}}),
            ),
        );
        assert_eq!(r.stale_on_error_max_age_ms, 7_000);
        assert!(r.shadow.enabled && !r.shadow.logs_mismatches());
        // A runtime static-policy overlay (via From<Policy>) behaves like its JSON form.
        let overlay = RuntimePolicy::from(Policy::default().local_ttl_sec(9));
        let r = resolve_policy(
            &Policy::default(),
            Some(&overlay),
            KEY,
            PolicyDefaults::default(),
        )
        .unwrap();
        assert!(r.local.enabled);
        assert_eq!(r.local.ttl_ms, 9_000);
        assert_eq!(r.remote.reason, Some(DisabledReason::PolicyDisabled));
    }

    #[test]
    fn typed_runtime_overlays_preserve_omitted_flags() {
        let base = Policy::default().request_local(true).coalesce(false);
        for overlay in [Policy::default(), Policy::default().local_ttl_sec(9)] {
            let r = resolve_policy(
                &base,
                Some(&RuntimePolicy::from(overlay)),
                KEY,
                PolicyDefaults::default(),
            )
            .unwrap();
            assert!(r.request_local, "an omitted flag must inherit the baseline");
            assert!(!r.coalesce, "an omitted flag must inherit the baseline");
        }

        let explicit = resolved(
            &base,
            Some(RuntimePolicy::from(Policy::default().request_local(false).coalesce(true)).0),
        );
        assert!(!explicit.request_local);
        assert!(explicit.coalesce);

        let disabled = resolved(&base, Some(RuntimePolicy::from(Policy::disabled()).0));
        assert!(!disabled.request_local);
        assert!(
            !disabled.coalesce,
            "the kill switch leaves coalescing unset"
        );
    }

    #[test]
    fn json_round_trip_preserves_omitted_and_explicit_flags() {
        for request_local in [None, Some(false), Some(true)] {
            for coalesce in [None, Some(false), Some(true)] {
                let policy = Policy {
                    request_local,
                    coalesce,
                    ..Policy::default()
                };
                assert_eq!(Policy::from_json(&policy.to_json()).unwrap(), policy);
            }
        }
    }

    #[test]
    fn instance_defaults_and_deadline_precedence() {
        let base = Policy::enabled(1);
        // Zero instance default means the library default.
        let r = resolve_policy(
            &base,
            None,
            KEY,
            PolicyDefaults {
                remote_read_timeout_ms: 0,
            },
        )
        .unwrap();
        assert_eq!(r.remote_read_timeout_ms, DEFAULT_REMOTE_READ_TIMEOUT_MS);
        // The instance default applies when neither operation nor runtime supplies a budget.
        let r = resolve_policy(
            &base,
            None,
            KEY,
            PolicyDefaults {
                remote_read_timeout_ms: 20,
            },
        )
        .unwrap();
        assert_eq!(r.remote_read_timeout_ms, 20);
        // Operation beats instance; runtime beats operation.
        let op = base.clone().remote_read_timeout_ms(30);
        let r = resolve_policy(
            &op,
            None,
            KEY,
            PolicyDefaults {
                remote_read_timeout_ms: 20,
            },
        )
        .unwrap();
        assert_eq!(r.remote_read_timeout_ms, 30);
        let overlay = RuntimePolicy(json!({"remoteReadTimeoutMs": 40}));
        let r = resolve_policy(
            &op,
            Some(&overlay),
            KEY,
            PolicyDefaults {
                remote_read_timeout_ms: 20,
            },
        )
        .unwrap();
        assert_eq!(r.remote_read_timeout_ms, 40);
        // An out-of-domain instance default fails resolution before anything else.
        let err = resolve_policy(
            &base,
            None,
            KEY,
            PolicyDefaults {
                remote_read_timeout_ms: MAX_DEADLINE_MS + 1,
            },
        );
        assert_eq!(
            err.unwrap_err(),
            PolicyError("invalid instance read deadline".to_string())
        );
        assert!(resolve_policy(
            &base,
            None,
            KEY,
            PolicyDefaults {
                remote_read_timeout_ms: MAX_DEADLINE_MS
            }
        )
        .is_ok());
        // A statically invalid base fails resolution even with a valid overlay.
        let invalid = Policy::default().local_ttl_sec(0);
        assert!(resolve(&invalid, Some(json!({"ttlSec": {"local": 5}}))).is_err());
    }

    #[test]
    fn error_messages_name_the_offending_container() {
        assert_eq!(
            Policy::from_json(&json!([])).unwrap_err(),
            PolicyError("DialCache defaultConfig must be an object".to_string())
        );
        assert_eq!(
            Policy::from_json(&json!({"ttlSec": null})).unwrap_err(),
            PolicyError("DialCache ttlSec must be an object".to_string())
        );
        assert_eq!(
            Policy::from_json(&json!({"shadowRamp": 1})).unwrap_err(),
            PolicyError("shadowRamp was replaced by shadow.ramp".to_string())
        );
        assert_eq!(
            resolve(&Policy::default(), Some(json!(1))).unwrap_err(),
            PolicyError("DialCache runtime config must be an object".to_string())
        );
        assert_eq!(
            resolve(&Policy::default(), Some(json!({"shadow": 1}))).unwrap_err(),
            PolicyError("DialCache shadow must be an object".to_string())
        );
        assert_eq!(
            resolve(&Policy::default(), Some(json!({"shadowRamp": null}))).unwrap_err(),
            PolicyError("shadowRamp was replaced by shadow.ramp".to_string())
        );
        assert_eq!(
            resolve(&Policy::default(), Some(json!({"coalesce": null}))).unwrap_err(),
            PolicyError("runtime coalesce must be boolean".to_string())
        );
        assert_eq!(
            resolve(&Policy::default(), Some(json!({"requestLocal": "x"}))).unwrap_err(),
            PolicyError("runtime requestLocal must be boolean".to_string())
        );
        assert_eq!(
            resolve(&Policy::default(), Some(json!({"remoteReadTimeoutMs": 0}))).unwrap_err(),
            PolicyError("invalid runtime remoteReadTimeoutMs".to_string())
        );
    }

    fn shadow_logging(base: &Policy, overlay: Option<Value>) -> (bool, MismatchDiff, bool) {
        let shadow = resolved(base, overlay).shadow;
        (
            shadow.mismatch_value,
            shadow.mismatch_diff,
            shadow.logging_config_error,
        )
    }

    fn shadow_base(log_mismatches: Option<bool>, group: Option<MismatchLogging>) -> Policy {
        Policy::default().shadow(ShadowPolicy {
            ramp: Some(100.0),
            log_mismatches,
            mismatch_logging: group,
        })
    }

    #[test]
    fn mismatch_logging_resolves_the_recorded_four_row_table() {
        use MismatchDiff::{Off, Paths, Values};
        let leaves = |value, diff| Some(MismatchLogging { value, diff });
        // { logMismatches: true, mismatchLogging: { value: false } }, in either layer.
        let row = shadow_base(Some(true), leaves(Some(false), None));
        assert_eq!(shadow_logging(&row, None), (false, Values, false));
        assert_eq!(
            shadow_logging(
                &shadow_base(None, None),
                Some(
                    json!({"shadow": {"logMismatches": true, "mismatchLogging": {"value": false}}})
                )
            ),
            (false, Values, false)
        );
        // { logMismatches: false, mismatchLogging: { diff: "paths" } }, in either layer.
        let row = shadow_base(Some(false), leaves(None, Some(Paths)));
        assert_eq!(shadow_logging(&row, None), (false, Paths, false));
        assert_eq!(
            shadow_logging(
                &shadow_base(None, None),
                Some(
                    json!({"shadow": {"logMismatches": false, "mismatchLogging": {"diff": "paths"}}})
                )
            ),
            (false, Paths, false)
        );
        // A runtime shorthand replaces both inherited leaves.
        let paths = shadow_base(None, leaves(None, Some(Paths)));
        assert_eq!(
            shadow_logging(&paths, Some(json!({"shadow": {"logMismatches": false}}))),
            (false, Off, false)
        );
        let value_off = shadow_base(None, leaves(Some(false), None));
        assert_eq!(
            shadow_logging(&value_off, Some(json!({"shadow": {"logMismatches": true}}))),
            (true, Values, false)
        );
    }

    #[test]
    fn mismatch_logging_leaves_merge_after_each_layer_expands() {
        use MismatchDiff::{Off, Paths, Values};
        let leaves = |value, diff| Some(MismatchLogging { value, diff });
        // An omitted runtime leaf inherits; a written one wins leaf by leaf.
        let base = shadow_base(None, leaves(Some(true), Some(Paths)));
        assert_eq!(
            shadow_logging(
                &base,
                Some(json!({"shadow": {"mismatchLogging": {"diff": "values"}}}))
            ),
            (true, Values, false)
        );
        assert_eq!(
            shadow_logging(
                &base,
                Some(json!({"shadow": {"mismatchLogging": {"value": false}}}))
            ),
            (false, Paths, false)
        );
        // A static shorthand is a baseline like any other pair of static leaves.
        let shorthand = shadow_base(Some(true), None);
        assert_eq!(
            shadow_logging(
                &shorthand,
                Some(json!({"shadow": {"mismatchLogging": {"diff": "paths"}}}))
            ),
            (true, Paths, false)
        );
        // An overlay without a shadow group, or with an empty one, keeps the static leaves.
        for overlay in [
            json!({}),
            json!({"shadow": {}}),
            json!({"shadow": {"ramp": 50}}),
        ] {
            assert_eq!(
                shadow_logging(&shorthand, Some(overlay)),
                (true, Values, false)
            );
        }
        // The disabled() overlay turns every inherited leaf off without a diagnostic.
        let all_on = shadow_base(Some(true), leaves(Some(true), Some(Values)));
        assert_eq!(
            shadow_logging(&all_on, Some(RuntimePolicy::from(Policy::disabled()).0)),
            (false, Off, false)
        );
    }

    #[test]
    fn a_malformed_runtime_shorthand_stays_raw_for_one_config_resolution() {
        let inherited = shadow_base(
            None,
            Some(MismatchLogging {
                value: Some(false),
                diff: Some(MismatchDiff::Values),
            }),
        );
        let written = json!({"logMismatches": "yes", "mismatchLogging": {"value": true}});
        let base = match inherited.to_json() {
            Value::Object(object) => object["shadow"].as_object().unwrap().clone(),
            _ => unreachable!(),
        };
        // The malformed shorthand expands to both leaves off over the static
        // leaves, the leaf written beside it still wins, and the raw value stays.
        let merged = merge_shadow(&base, written.as_object().unwrap()).unwrap();
        assert_eq!(
            Value::Object(merged),
            json!({
                "ramp": 100,
                "logMismatches": "yes",
                "mismatchLogging": {"value": true, "diff": "off"},
            })
        );
        assert_eq!(
            shadow_logging(&inherited, Some(json!({"shadow": written}))),
            (true, MismatchDiff::Off, true)
        );
        // Any number of invalid inputs owes the same single flag.
        assert_eq!(
            shadow_logging(
                &inherited,
                Some(json!({"shadow": {
                    "logMismatches": null,
                    "mismatchLogging": {"value": "yes", "diff": "full"},
                }}))
            ),
            (false, MismatchDiff::Off, true)
        );
    }

    #[test]
    fn mismatch_logging_resolves_each_leaf_independently() {
        use MismatchDiff::{Off, Paths, Values};
        let cases = [
            (json!({}), (false, Off, false)),
            (json!({"mismatchLogging": {}}), (false, Off, false)),
            (
                json!({"mismatchLogging": {"value": true, "diff": "paths"}}),
                (true, Paths, false),
            ),
            (
                json!({"mismatchLogging": {"value": "true", "diff": "paths"}}),
                (false, Paths, true),
            ),
            (
                json!({"mismatchLogging": {"value": null}}),
                (false, Off, true),
            ),
            (
                json!({"mismatchLogging": {"value": true, "diff": "full"}}),
                (true, Off, true),
            ),
            (json!({"mismatchLogging": {"diff": 1}}), (false, Off, true)),
            (
                json!({"mismatchLogging": {"diff": null}}),
                (false, Off, true),
            ),
            (
                json!({"mismatchLogging": {"diff": "toString"}}),
                (false, Off, true),
            ),
            (
                json!({"mismatchLogging": {"value": true, "diff": "constructor"}}),
                (true, Off, true),
            ),
            (
                json!({"mismatchLogging": {"diff": "Values"}}),
                (false, Off, true),
            ),
            (json!({"logMismatches": "yes"}), (false, Off, true)),
            (json!({"logMismatches": null}), (false, Off, true)),
            (json!({"logMismatches": 0}), (false, Off, true)),
            (
                json!({"logMismatches": "yes", "mismatchLogging": {"value": true, "diff": "values"}}),
                (true, Values, true),
            ),
            // Unknown members stay ignored, as on main.
            (
                json!({"mismatchLogging": {"diff": "paths", "unknown": 1}, "unknown": true}),
                (false, Paths, false),
            ),
        ];
        for (shadow, expected) in cases {
            assert_eq!(
                shadow_logging(&Policy::default(), Some(json!({"shadow": shadow.clone()}))),
                expected,
                "{shadow}"
            );
        }
    }

    #[test]
    fn a_non_object_runtime_mismatch_logging_group_fails_resolution() {
        for group in [
            json!(null),
            json!([]),
            json!(1),
            json!("paths"),
            json!(true),
        ] {
            for base in [Policy::default(), shadow_base(Some(true), None)] {
                assert_eq!(
                    resolve(
                        &base,
                        Some(json!({"shadow": {"logMismatches": true, "mismatchLogging": group.clone()}}))
                    )
                    .unwrap_err(),
                    PolicyError("DialCache shadow mismatchLogging config must be an object".to_string()),
                    "{group}"
                );
            }
        }
    }

    // Port of the TypeScript merge grid: the recorded rule is written out
    // without the code under test and checked over every combination.
    #[test]
    fn every_combination_of_shorthand_and_leaves_resolves_as_each_layer_expanded_first() {
        fn layer_leaves(
            shorthand: Option<&Value>,
            group: Option<&Value>,
        ) -> (Option<Value>, Option<Value>) {
            let (mut value, mut diff) = (None, None);
            if let Some(shorthand) = shorthand {
                let on = shorthand == &Value::Bool(true);
                value = Some(Value::Bool(on));
                diff = Some(Value::from(if on { "values" } else { "off" }));
            }
            if let Some(group) = group.and_then(Value::as_object) {
                if let Some(leaf) = group.get("value") {
                    value = Some(leaf.clone());
                }
                if let Some(leaf) = group.get("diff") {
                    diff = Some(leaf.clone());
                }
            }
            (value, diff)
        }
        let static_shorthands = [None, Some(true), Some(false)];
        let static_groups = [
            None,
            Some((None, None)),
            Some((Some(true), None)),
            Some((Some(false), None)),
            Some((None, Some(MismatchDiff::Paths))),
            Some((None, Some(MismatchDiff::Off))),
            Some((Some(true), Some(MismatchDiff::Values))),
        ];
        let runtime_shorthands = [
            None,
            Some(json!(true)),
            Some(json!(false)),
            Some(json!("yes")),
            Some(json!(null)),
            Some(json!(0)),
        ];
        let runtime_groups = [
            None,
            Some(json!({})),
            Some(json!({"value": true})),
            Some(json!({"value": false})),
            Some(json!({"diff": "paths"})),
            Some(json!({"diff": "values"})),
            Some(json!({"value": true, "diff": "off"})),
            Some(json!({"value": "yes"})),
            Some(json!({"value": null})),
            Some(json!({"diff": "full"})),
            Some(json!({"diff": "toString"})),
        ];
        let mut combinations = 0;
        for static_shorthand in static_shorthands {
            for static_group in static_groups {
                let base = shadow_base(
                    static_shorthand,
                    static_group.map(|(value, diff)| MismatchLogging { value, diff }),
                );
                let static_shadow = base.to_json()["shadow"].clone();
                let inherited = layer_leaves(
                    static_shadow.get("logMismatches"),
                    static_shadow.get("mismatchLogging"),
                );
                for runtime_shorthand in &runtime_shorthands {
                    for runtime_group in &runtime_groups {
                        let mut shadow = Map::new();
                        if let Some(shorthand) = runtime_shorthand {
                            shadow.insert("logMismatches".to_string(), shorthand.clone());
                        }
                        if let Some(group) = runtime_group {
                            shadow.insert("mismatchLogging".to_string(), group.clone());
                        }
                        let written =
                            layer_leaves(runtime_shorthand.as_ref(), runtime_group.as_ref());
                        let value = written.0.or(inherited.0.clone());
                        let diff = written.1.or(inherited.1.clone());
                        let valid_value = value.as_ref().is_none_or(Value::is_boolean);
                        let level = diff.as_ref().and_then(MismatchDiff::from_json);
                        let valid_diff = diff.is_none() || level.is_some();
                        let malformed = runtime_shorthand.as_ref().is_some_and(|s| !s.is_boolean());
                        let expected = (
                            valid_value && value == Some(Value::Bool(true)),
                            level.unwrap_or(MismatchDiff::Off),
                            malformed || !valid_value || !valid_diff,
                        );
                        let overlay = json!({"shadow": Value::Object(shadow)});
                        assert_eq!(
                            shadow_logging(&base, Some(overlay.clone())),
                            expected,
                            "static {static_shadow} runtime {overlay}"
                        );
                        combinations += 1;
                    }
                }
            }
        }
        assert_eq!(combinations, 3 * 7 * 6 * 11);
    }
}