optionchain_simulator 0.2.11

//! Session model for v2 rolling simulations.
//!
//! Separate from [`crate::session::model`] on purpose. `SimulationParameters`
//! and `Session` are public, IronCondor-facing, and persisted as serde JSON, so
//! adding rolling fields to them would be source-breaking even where serde
//! defaults kept old documents loading. ADR 0001 §12 therefore gives v2 its own
//! types, its own stored schema, and its own store key space, and freezes v1
//! exactly as it is.
//!
//! Two properties distinguish these types from their v1 counterparts:
//!
//! - **The effective inputs are resolved, not optional.** After conversion the
//!   seed, the simulated start and the step interval are concrete values, not
//!   `Option`s that some later code path has to default again. Whatever the
//!   request omitted is decided once, here, and echoed back so the run can be
//!   replayed.
//! - **The configuration is immutable.** There is no PATCH or PUT for a v2
//!   simulation: changing any parameter changes the tape, so it creates a new
//!   simulation instead of mutating one.

use crate::api::rest::limits::{MAX_CHAIN_SIZE, MAX_STEPS, strikes_per_chain};
use crate::api::rest::models::validate_walk_type;
use crate::api::rest::requests_v2::CreateSimulationRequest;
use crate::api::rest::validation::{
    bounded_decimal_field, decimal_field, positive_field, strictly_positive_field, symbol_field,
    time_frame_field,
};
use crate::domain::expiry::{CalendarVersion, ExpirationSchedule, tzdb_version};
use crate::domain::ladder::{MAX_PINNED_WIDTH, StrikeLadder, ensure_ladder_fits};
use crate::domain::simulator::DEFAULT_CHAIN_SIZE;
use crate::domain::spread::{MAX_PROPORTIONAL, MAX_TICK, MAX_WIDENING};
use crate::infrastructure::max_snapshot_contracts;
use crate::session::model::{SessionState, SimulationMethod};
use crate::utils::ChainError;
use chrono::{DateTime, NaiveTime, TimeDelta, Timelike, Utc};
use chrono_tz::Tz;
use optionstratlib::utils::TimeFrame;
use positive::Positive;
use rand::RngExt;
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize};
use std::fmt;
use std::str::FromStr;
use std::time::SystemTime;
use tracing::warn;
use uuid::Uuid;

/// Schema version stamped on every stored v2 session.
///
/// It rides with the document rather than being inferred from its shape, so a
/// future migration can tell an old document from a new one without guessing.
/// Bumping it is a semver event for the stored contract (ADR 0001 §12.2).
pub const SESSION_V2_SCHEMA_VERSION: u32 = 1;

/// Shortest simulated step interval, in seconds.
///
/// Crate-internal on purpose: issue #48 turns the v2 bounds into validated
/// `OCS_MAX_*` environment knobs following the `LazyLock` pattern in
/// `api::rest::limits`, and publishing them as `const u64` first would make
/// that conversion a breaking change to a released item.
pub(crate) const MIN_STEP_INTERVAL_SECONDS: u64 = 1;

/// Longest simulated step interval, in seconds — one 365-day year.
///
/// Crate-internal for the same reason as [`MIN_STEP_INTERVAL_SECONDS`].
pub(crate) const MAX_STEP_INTERVAL_SECONDS: u64 = 31_536_000;

/// Seconds in a 365-day year, used to derive an interval from a `Custom`
/// time frame expressed in periods per year.
const SECONDS_PER_YEAR: u64 = 31_536_000;

/// The calendar policy accepted today. Anything else is rejected so a stored
/// simulation can never be reinterpreted under a policy it was not created
/// with.
const SUPPORTED_CALENDAR: &str = "weekdays_v1";

/// Derives the simulated step interval from a stochastic-model time frame.
///
/// `time_frame` scales the model; `step_interval_seconds` drives the simulated
/// clock. They are allowed to differ, and this is only the default used when
/// the request omits the interval.
///
/// # Errors
///
/// Returns [`ChainError::Validation`] naming `step_interval_seconds` when the
/// frame has no usable second-length: `Microsecond` and `Millisecond` derive to
/// less than one second, and a small `Custom` periods-per-year derives to more
/// than a year. Both are rejected rather than silently clamped, because a
/// clamped interval would silently change the simulated clock.
fn derive_step_interval_seconds(time_frame: TimeFrame) -> Result<u64, ChainError> {
    let too_small = || ChainError::Validation {
        field: "step_interval_seconds".to_string(),
        reason: format!(
            "cannot be derived from time_frame {time_frame:?}: it is shorter than {MIN_STEP_INTERVAL_SECONDS} second; supply step_interval_seconds explicitly"
        ),
    };

    let seconds = match time_frame {
        TimeFrame::Microsecond | TimeFrame::Millisecond => return Err(too_small()),
        TimeFrame::Second => 1,
        TimeFrame::Minute => 60,
        TimeFrame::Hour => 3_600,
        TimeFrame::Day => 86_400,
        TimeFrame::Week => 604_800,
        TimeFrame::Month => 2_592_000,
        TimeFrame::Quarter => 7_776_000,
        TimeFrame::Year => SECONDS_PER_YEAR,
        TimeFrame::Custom(periods_per_year) => {
            let periods = periods_per_year.to_f64();
            if periods <= 0.0 || !periods.is_finite() {
                return Err(ChainError::Validation {
                    field: "time_frame".to_string(),
                    reason: format!(
                        "custom periods per year must be finite and positive, got {periods}"
                    ),
                });
            }
            let seconds = (SECONDS_PER_YEAR as f64 / periods).round();
            if !(MIN_STEP_INTERVAL_SECONDS as f64..=MAX_STEP_INTERVAL_SECONDS as f64)
                .contains(&seconds)
            {
                return Err(ChainError::Validation {
                    field: "step_interval_seconds".to_string(),
                    reason: format!(
                        "derived interval {seconds} s is outside [{MIN_STEP_INTERVAL_SECONDS}, {MAX_STEP_INTERVAL_SECONDS}]; supply step_interval_seconds explicitly"
                    ),
                });
            }
            seconds as u64
        }
    };

    Ok(seconds)
}

/// Rejects a `Positive` that is zero, naming the field.
///
/// `Positive` guarantees non-negative, not strictly positive, so the request
/// path's `strictly_positive_field` check has no type-level counterpart on a
/// value read back from the store.
fn reject_zero(field: &str, value: Positive) -> Result<(), ChainError> {
    if value == Positive::ZERO {
        return Err(ChainError::Validation {
            field: field.to_string(),
            reason: "must be strictly positive, got 0".to_string(),
        });
    }
    Ok(())
}

/// Rejects a spread coefficient above the cap the request path applies.
///
/// The cap is declared as an `f64` beside the model, so it is converted here
/// rather than duplicated as a `Decimal`: one definition, and a conversion that
/// cannot silently disagree with it.
fn reject_above_cap(field: &str, value: Decimal, maximum: f64) -> Result<(), ChainError> {
    let Ok(maximum) = Decimal::try_from(maximum) else {
        // Unreachable: every cap is a small literal. A cap that cannot be
        // represented is not a bound anything can be checked against, so the
        // value is refused rather than admitted unchecked.
        return Err(ChainError::Validation {
            field: field.to_string(),
            reason: "the configured maximum is not representable".to_string(),
        });
    };
    if value > maximum {
        return Err(ChainError::Validation {
            field: field.to_string(),
            reason: format!("must not exceed {maximum}, got {value}"),
        });
    }
    Ok(())
}

/// Validates an explicitly-supplied step interval.
fn validate_step_interval_seconds(seconds: u64) -> Result<u64, ChainError> {
    if !(MIN_STEP_INTERVAL_SECONDS..=MAX_STEP_INTERVAL_SECONDS).contains(&seconds) {
        return Err(ChainError::Validation {
            field: "step_interval_seconds".to_string(),
            reason: format!(
                "must be within [{MIN_STEP_INTERVAL_SECONDS}, {MAX_STEP_INTERVAL_SECONDS}], got {seconds}"
            ),
        });
    }
    Ok(seconds)
}

/// Parses the local expiration time, accepting `HH:MM` and `HH:MM:SS`.
fn parse_expiration_time(raw: &str) -> Result<NaiveTime, ChainError> {
    NaiveTime::parse_from_str(raw, "%H:%M:%S")
        .or_else(|_| NaiveTime::parse_from_str(raw, "%H:%M"))
        .map_err(|_| ChainError::Validation {
            field: "expiration_time".to_string(),
            reason: format!("must be a local time as HH:MM or HH:MM:SS, got {raw:?}"),
        })
}

/// Parses an IANA time-zone name.
fn parse_timezone(raw: &str) -> Result<Tz, ChainError> {
    Tz::from_str(raw).map_err(|_| ChainError::Validation {
        field: "timezone".to_string(),
        reason: format!("must be a known IANA time-zone name, got {raw:?}"),
    })
}

/// Parses the calendar policy version.
fn parse_calendar(raw: Option<&str>) -> Result<CalendarVersion, ChainError> {
    match raw.unwrap_or(SUPPORTED_CALENDAR) {
        SUPPORTED_CALENDAR => Ok(CalendarVersion::WeekdaysV1),
        other => Err(ChainError::Validation {
            field: "calendar".to_string(),
            reason: format!("must be {SUPPORTED_CALENDAR}, got {other:?}"),
        }),
    }
}

/// Normalises an instant to whole-second UTC.
///
/// Truncating the sub-second part is what lets every timestamp in the API and
/// in the exports render as `YYYY-MM-DDTHH:MM:SSZ`, which is in turn what makes
/// a repeated export byte-comparable (ADR 0001 §3.1).
fn to_whole_second_utc(instant: DateTime<Utc>) -> Result<DateTime<Utc>, ChainError> {
    instant
        .with_nanosecond(0)
        .ok_or_else(|| ChainError::Validation {
            field: "start_at".to_string(),
            reason: format!("{instant} cannot be normalised to a whole second"),
        })
}

/// The resolved parameters of a v2 rolling simulation.
///
/// Every field is effective: nothing here is still waiting to be defaulted. The
/// set of fields is exactly the replay input list of ADR 0001 §8, which is what
/// lets a client reproduce a run from the creation response alone.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(try_from = "SimulationParametersV2Wire")]
pub struct SimulationParametersV2 {
    /// Ticker symbol of the underlying.
    pub symbol: String,
    /// Number of steps the simulation runs for.
    pub steps: usize,
    /// The resolved simulated start, in whole-second UTC.
    pub effective_start: DateTime<Utc>,
    /// The resolved interval between simulated steps, in seconds.
    pub step_interval_seconds: u64,
    /// Time frame the stochastic model is scaled by.
    pub time_frame: TimeFrame,
    /// The normalised rolling expiration schedule.
    pub schedule: ExpirationSchedule,
    /// The IANA time-zone database release the expirations were resolved
    /// against, e.g. `2025b`. A replay against a different release is still a
    /// replay — it is just one the client can detect (ADR 0001 §8).
    pub tzdb_version: String,
    /// Initial price of the underlying.
    pub initial_price: Positive,
    /// Initial volatility.
    pub volatility: Positive,
    /// Annualised risk-free rate.
    pub risk_free_rate: Decimal,
    /// Annualised dividend yield.
    pub dividend_yield: Positive,
    /// The stochastic model driving the underlying path.
    pub method: SimulationMethod,
    /// Number of strikes per chain.
    pub chain_size: Option<usize>,
    /// Interval between strikes.
    pub strike_interval: Option<Positive>,
    /// Slope of the volatility skew.
    pub skew_slope: Option<Decimal>,
    /// Curvature of the volatility smile.
    pub smile_curve: Option<Decimal>,
    /// Which strikes the simulation quotes: `rolling` rebuilds the ladder
    /// around the spot at every step, `pinned` fixes it at creation so a
    /// contract quoted once stays quoted. Defaults to `rolling`, which is what
    /// the service did before the field existed.
    #[serde(default)]
    pub strike_ladder: StrikeLadder,
    /// The constant term of the spread model, and the whole of it when no
    /// widening coefficient is set. The field predates the model and keeps its
    /// meaning: one absolute spread, applied to every contract.
    pub spread: Option<Positive>,
    /// How much of a contract's mid price is added to its spread. Zero when
    /// absent, which is what makes an untouched request behave as it did.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub spread_proportional: Option<Decimal>,
    /// How fast the spread widens away from the money, per unit of
    /// `|ln(strike / underlying)|`. Zero when absent.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub spread_moneyness_widening: Option<Decimal>,
    /// How fast the spread widens with time to expiry, per `sqrt(years)`. Zero
    /// when absent.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub spread_tenor_widening: Option<Decimal>,
    /// The smallest quotable increment, and the floor under every bid. One cent
    /// when absent.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub spread_tick: Option<Positive>,
    /// The effective RNG seed. Non-optional: a v2 simulation is always
    /// reproducible, so the seed is resolved at conversion and never `None`.
    pub seed: u64,
    /// How far a pinned ladder may make a step widen the chain it asks
    /// upstream for, in strikes per side.
    ///
    /// Resolved ONCE, at creation, from the per-snapshot contract cap and this
    /// simulation's own schedule, exactly as the effective seed is resolved
    /// once and then carried. The domain reads it from here rather than from
    /// the environment, so a session keeps the ceiling it was created under for
    /// its whole life: the same parameters and seed then reach the same step on
    /// any instance, whatever an operator has since done to
    /// `OCS_MAX_SNAPSHOT_CONTRACTS`.
    ///
    /// A `rolling` simulation carries it too and never consults it.
    pub pinned_width_ceiling: usize,
}

/// The deserialization shape of [`SimulationParametersV2`].
///
/// Exists so that a stored document is validated exactly like a request. The
/// fields are public and the type derives `Deserialize`, so without this a
/// hand-edited or corrupted document in Redis would sail past every check the
/// request path performs: a `step_interval_seconds` of `0` freezes the
/// simulated clock, a `steps` above the cap drives an unbounded factor tape, a
/// sub-second `effective_start` breaks the whole-second rendering that makes
/// exports byte-comparable. Redis is an outer layer, and
/// `rules/global_rules.md` is explicit that domain types must not trust one.
///
/// It also carries `deny_unknown_fields`, which turns the rolling-deploy
/// hazard into a loud one: an old binary reading a document written by a newer
/// one fails instead of silently dropping the new fields and writing the
/// truncated document back.
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct SimulationParametersV2Wire {
    symbol: String,
    steps: usize,
    effective_start: DateTime<Utc>,
    step_interval_seconds: u64,
    time_frame: TimeFrame,
    schedule: ExpirationSchedule,
    tzdb_version: String,
    initial_price: Positive,
    volatility: Positive,
    risk_free_rate: Decimal,
    dividend_yield: Positive,
    method: SimulationMethod,
    chain_size: Option<usize>,
    strike_interval: Option<Positive>,
    skew_slope: Option<Decimal>,
    smile_curve: Option<Decimal>,
    spread: Option<Positive>,
    #[serde(default)]
    strike_ladder: StrikeLadder,
    #[serde(default)]
    spread_proportional: Option<Decimal>,
    #[serde(default)]
    spread_moneyness_widening: Option<Decimal>,
    #[serde(default)]
    spread_tenor_widening: Option<Decimal>,
    #[serde(default)]
    spread_tick: Option<Positive>,
    seed: u64,
    /// Absent in a document written before the ceiling was stored. Such a
    /// session is resolved once here, on load, which is the ceiling it would
    /// have had, and keeps it from then on.
    #[serde(default)]
    pinned_width_ceiling: Option<usize>,
}

impl TryFrom<SimulationParametersV2Wire> for SimulationParametersV2 {
    type Error = ChainError;

    fn try_from(wire: SimulationParametersV2Wire) -> Result<Self, Self::Error> {
        // A document written before the ceiling was stored resolves it once,
        // here, and carries it from then on.
        let pinned_width_ceiling = match wire.pinned_width_ceiling {
            Some(ceiling) => ceiling,
            None => crate::domain::resolve_pinned_ceiling(&wire.schedule)?,
        };
        let parameters = Self {
            symbol: wire.symbol,
            steps: wire.steps,
            effective_start: wire.effective_start,
            step_interval_seconds: wire.step_interval_seconds,
            time_frame: wire.time_frame,
            schedule: wire.schedule,
            tzdb_version: wire.tzdb_version,
            initial_price: wire.initial_price,
            volatility: wire.volatility,
            risk_free_rate: wire.risk_free_rate,
            dividend_yield: wire.dividend_yield,
            method: wire.method,
            chain_size: wire.chain_size,
            strike_interval: wire.strike_interval,
            skew_slope: wire.skew_slope,
            smile_curve: wire.smile_curve,
            spread: wire.spread,
            strike_ladder: wire.strike_ladder,
            spread_proportional: wire.spread_proportional,
            spread_moneyness_widening: wire.spread_moneyness_widening,
            spread_tenor_widening: wire.spread_tenor_widening,
            spread_tick: wire.spread_tick,
            seed: wire.seed,
            pinned_width_ceiling,
        };
        parameters.validate_stored()?;
        Ok(parameters)
    }
}

impl SimulationParametersV2 {
    /// Re-checks every invariant the request path establishes.
    ///
    /// Called from the `Deserialize` path, so a stored document is held to the
    /// same standard as a request. Cheap: a handful of comparisons and one
    /// symbol check, run once per load.
    ///
    /// A `tzdb_version` that differs from the running binary's is a **warning**,
    /// not a rejection: the simulation is still coherent, it was simply resolved
    /// against a different IANA release, and refusing to load it would turn a
    /// dependency bump into an outage. Issue #46 decides whether a mid-tape
    /// divergence should be escalated.
    ///
    /// # Errors
    ///
    /// Returns [`ChainError::Validation`] naming the offending field when
    /// `steps` is outside `1..=MAX_STEPS`, `chain_size` exceeds
    /// `MAX_CHAIN_SIZE`, the symbol violates the identifier format,
    /// `step_interval_seconds` is outside its documented range,
    /// `effective_start` is not on a whole second, `initial_price`,
    /// `volatility` or `strike_interval` is not strictly positive, the walk
    /// model fails its own invariants, `volatility` disagrees with the walk
    /// model's own volatility, or the schedule is invalid.
    pub fn validate(&self) -> Result<(), ChainError> {
        self.validate_inner(true)
    }

    /// The same invariants, for a document coming back from the STORE rather
    /// than from a request.
    ///
    /// One check differs, deliberately. `OCS_MAX_SNAPSHOT_CONTRACTS` is
    /// admission control: it decides what this instance is willing to accept,
    /// which is a question about the future. A session that already exists
    /// answered it when it was created, and re-answering it on load with
    /// whatever cap the instance happens to run today would mean a simulation
    /// created under a higher cap fails to DESERIALIZE on a smaller instance,
    /// taking the tape with it. That is exactly the cross-instance breakage
    /// issue #109 set out to remove.
    ///
    /// So a stored document whose work exceeds this instance's cap loads with
    /// a warning, the same treatment a `tzdb_version` from another release
    /// gets, and every deployment-independent invariant still applies —
    /// including the absolute ceiling on `pinned_width_ceiling`, which is what
    /// keeps a corrupted document from buying unbounded pricing.
    ///
    /// # Errors
    ///
    /// As [`SimulationParametersV2::validate`], minus the per-snapshot cap.
    pub fn validate_stored(&self) -> Result<(), ChainError> {
        self.validate_inner(false)
    }

    /// The shared body. `enforce_cap` is what separates a request from a
    /// document that already exists.
    fn validate_inner(&self, enforce_cap: bool) -> Result<(), ChainError> {
        if self.steps < 1 {
            return Err(ChainError::Validation {
                field: "steps".to_string(),
                reason: "must be at least 1".to_string(),
            });
        }
        if self.steps > *MAX_STEPS {
            return Err(ChainError::Validation {
                field: "steps".to_string(),
                reason: format!("must not exceed {}, got {}", *MAX_STEPS, self.steps),
            });
        }
        if let Some(chain_size) = self.chain_size
            && chain_size > *MAX_CHAIN_SIZE
        {
            return Err(ChainError::Validation {
                field: "chain_size".to_string(),
                reason: format!("must not exceed {}, got {chain_size}", *MAX_CHAIN_SIZE),
            });
        }
        symbol_field("symbol", &self.symbol)?;
        validate_step_interval_seconds(self.step_interval_seconds)?;

        // `Positive` admits zero, so the strict-positive constraints the
        // request path enforces have to be rechecked here: a stored price or
        // volatility of zero is a chain of zero-value options, and a stored
        // `strike_interval` of zero collapses every strike onto one.
        for (field, value) in [
            ("initial_price", self.initial_price),
            ("volatility", self.volatility),
        ] {
            reject_zero(field, value)?;
        }
        if let Some(strike_interval) = self.strike_interval {
            reject_zero("strike_interval", strike_interval)?;
        }
        // A pinned ladder is a fixed set of strikes on a fixed grid, and
        // without an explicit interval upstream derives a different one per
        // expiration: there would be no single ladder to pin. Refused at the
        // boundary rather than resolved to some arbitrary expiration's
        // interval, which would be a number nobody asked for.
        if self.strike_ladder.is_pinned() {
            let Some(interval) = self.strike_interval else {
                return Err(ChainError::Validation {
                    field: "strike_interval".to_string(),
                    reason: "a pinned strike_ladder requires an explicit strike_interval"
                        .to_string(),
                });
            };
            // And the ladder has to be one upstream can actually build: it
            // stops extending downwards past the anchor, so a ladder wider
            // than the spot loses its lowest strikes. Refused here rather than
            // on the first step of a simulation the client already holds.
            ensure_ladder_fits(
                self.initial_price,
                interval,
                self.chain_size.unwrap_or(DEFAULT_CHAIN_SIZE),
            )?;
        }
        // The spread model, BOTH bounds, exactly as the request path applies
        // them. A stored `spread_tick` of zero silently disables the floor that
        // keeps a wing quotable; a stored negative coefficient — which
        // `Decimal` admits — would narrow a quote away from the money; and a
        // stored coefficient above its cap would price a chain the REST
        // boundary refuses to create. Redis is an outer layer, so a document
        // that never passed through a request must clear the same bar.
        if let Some(tick) = self.spread_tick {
            reject_zero("spread_tick", tick)?;
            reject_above_cap("spread_tick", tick.to_dec(), MAX_TICK)?;
        }
        for (field, value, maximum) in [
            (
                "spread_proportional",
                self.spread_proportional,
                MAX_PROPORTIONAL,
            ),
            (
                "spread_moneyness_widening",
                self.spread_moneyness_widening,
                MAX_WIDENING,
            ),
            (
                "spread_tenor_widening",
                self.spread_tenor_widening,
                MAX_WIDENING,
            ),
        ] {
            let Some(value) = value else {
                continue;
            };
            if value < Decimal::ZERO {
                return Err(ChainError::Validation {
                    field: field.to_string(),
                    reason: format!("must not be negative, got {value}"),
                });
            }
            reject_above_cap(field, value, maximum)?;
        }
        validate_walk_type(&self.method)?;

        if self.effective_start.nanosecond() != 0 {
            return Err(ChainError::Validation {
                field: "effective_start".to_string(),
                reason: format!(
                    "must be on a whole second, got {}",
                    self.effective_start.to_rfc3339()
                ),
            });
        }
        self.schedule.validate()?;
        self.enforce_snapshot_work(enforce_cap)?;

        // Deployment-independent, so it holds on every path: a stored ceiling
        // above the absolute maximum would let a corrupted document buy a
        // widening no request could ever ask for.
        if self.pinned_width_ceiling > MAX_PINNED_WIDTH {
            return Err(ChainError::Validation {
                field: "pinned_width_ceiling".to_string(),
                reason: format!(
                    "must not exceed {MAX_PINNED_WIDTH}, got {}",
                    self.pinned_width_ceiling
                ),
            });
        }

        // A simulation has exactly one base volatility. v1 accepts a top-level
        // value and a walk model carrying a different one, and silently prices
        // step zero at the first while walking on the second; v2 refuses the
        // contradiction at the boundary rather than letting the domain pick a
        // winner later. `Historical` carries no model volatility, so there is
        // nothing to disagree with.
        if let Some(model_volatility) = self.method.volatility()
            && model_volatility != self.volatility
        {
            return Err(ChainError::Validation {
                field: "volatility".to_string(),
                reason: format!(
                    "must match the walk model's volatility ({model_volatility}), got {}; \
                     a simulation has exactly one base volatility",
                    self.volatility
                ),
            });
        }

        let running = tzdb_version();
        if self.tzdb_version != running {
            warn!(
                stored = %self.tzdb_version,
                running = %running,
                "simulation was resolved against a different IANA tzdb release"
            );
        }

        Ok(())
    }

    /// Rejects a configuration whose every snapshot would price more contracts
    /// than the service is willing to build.
    ///
    /// The two caps that bound this individually — the chain size and the
    /// per-snapshot expiration count — are each reasonable, and their product
    /// is not: 1 001 strikes across 512 live expirations is half a million
    /// Black-Scholes evaluations for one `/snapshot` call, from a request that
    /// violates neither. The bound is on the product, checked once at creation
    /// rather than per step, and it is deliberately generous: the reference
    /// configuration in ADR 0001 prices about 500 contracts a snapshot.
    ///
    /// `Σ target_count` is the tight upper bound on live expirations — rules
    /// that claim the same date are priced once, so the real count is at most
    /// this — which means a configuration this accepts can never exceed the
    /// cap, and one it rejects genuinely asked for more.
    ///
    /// # Errors
    ///
    /// Returns [`ChainError::Validation`] naming `chain_size`, which is the
    /// field a client can lower without changing what the simulation means.
    fn enforce_snapshot_work(&self, enforce_cap: bool) -> Result<(), ChainError> {
        let contracts = self.snapshot_contracts()?;
        let cap = max_snapshot_contracts();

        if contracts <= cap {
            return Ok(());
        }
        if !enforce_cap {
            // A session that already exists keeps loading; see
            // `validate_stored`.
            warn!(
                contracts,
                cap,
                "a stored simulation prices more contracts per snapshot than this instance would accept"
            );
            return Ok(());
        }

        Err(ChainError::Validation {
            field: "chain_size".to_string(),
            reason: format!(
                "every snapshot would price {contracts} contracts, above the {cap} maximum; \
                 lower chain_size or the schedules' target_count"
            ),
        })
    }

    /// How many contracts one snapshot of this configuration prices.
    ///
    /// # Errors
    ///
    /// Returns [`ChainError::Validation`] when the count is not representable.
    fn snapshot_contracts(&self) -> Result<usize, ChainError> {
        let requested = self.chain_size.unwrap_or(DEFAULT_CHAIN_SIZE);
        let strikes = strikes_per_chain(requested).ok_or_else(|| ChainError::Validation {
            field: "chain_size".to_string(),
            reason: format!("a chain of {requested} does not have a representable strike count"),
        })?;
        let expirations = self
            .schedule
            .rules()
            .iter()
            .try_fold(0usize, |total, rule| {
                total.checked_add(rule.target_count().get())
            })
            .ok_or_else(|| ChainError::Validation {
                field: "schedules".to_string(),
                reason: "the requested expiration counts overflow".to_string(),
            })?;

        strikes
            .checked_mul(expirations)
            .ok_or_else(|| ChainError::Validation {
                field: "chain_size".to_string(),
                reason: format!(
                    "{strikes} strikes across {expirations} expirations overflows the \
                     contract count"
                ),
            })
    }

    /// The simulated instant at `cursor`.
    ///
    /// `effective_start + cursor × step_interval`, with checked arithmetic
    /// throughout. Never reads the wall clock, so the same parameters derive
    /// the same instant on every call, in every process.
    ///
    /// # Errors
    ///
    /// Returns [`ChainError::Validation`] naming `steps` when the product or
    /// the sum leaves the representable range — an overflow is a rejected
    /// request, never a wrapped timestamp.
    pub fn simulated_at(&self, cursor: usize) -> Result<DateTime<Utc>, ChainError> {
        let overflow = || ChainError::Validation {
            field: "steps".to_string(),
            reason: format!(
                "simulated time overflows at cursor {cursor} with a {} s interval",
                self.step_interval_seconds
            ),
        };

        let cursor = i64::try_from(cursor).map_err(|_| overflow())?;
        let interval = i64::try_from(self.step_interval_seconds).map_err(|_| overflow())?;
        let offset = cursor.checked_mul(interval).ok_or_else(overflow)?;
        let delta = TimeDelta::try_seconds(offset).ok_or_else(overflow)?;

        self.effective_start
            .checked_add_signed(delta)
            .ok_or_else(overflow)
    }

    /// The simulated instant one step past the last one served, i.e. the end of
    /// the simulated horizon.
    ///
    /// # Errors
    ///
    /// Returns [`ChainError::Validation`] when the horizon overflows.
    pub fn simulated_end(&self) -> Result<DateTime<Utc>, ChainError> {
        self.simulated_at(self.steps)
    }
}

impl fmt::Display for SimulationParametersV2 {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let json = serde_json::to_string(self).map_err(|_| fmt::Error)?;
        write!(f, "{json}")
    }
}

impl TryFrom<CreateSimulationRequest> for SimulationParametersV2 {
    type Error = ChainError;

    /// Validates a client-supplied [`CreateSimulationRequest`] and resolves it
    /// into effective parameters.
    ///
    /// This is the single place where the v2 REST `f64` / string boundary
    /// becomes `Positive` / `Decimal` / `Tz` / `NaiveTime`, mirroring what
    /// `TryFrom<CreateSessionRequest>` does for v1. Three values are *resolved*
    /// here rather than merely converted, and all three are surfaced back to the
    /// client so the tape can be replayed:
    ///
    /// - the **seed**, generated when absent, as in v1;
    /// - the **effective start**, generated when absent and normalised to
    ///   whole-second UTC — the only wall-clock read in a v2 simulation's whole
    ///   life;
    /// - the **step interval**, derived from `time_frame` when absent.
    ///
    /// # Errors
    ///
    /// Returns [`ChainError::Validation`] naming the first field that fails.
    fn try_from(request: CreateSimulationRequest) -> Result<Self, Self::Error> {
        if request.steps < 1 {
            return Err(ChainError::Validation {
                field: "steps".to_string(),
                reason: "must be at least 1".to_string(),
            });
        }
        if request.steps > *MAX_STEPS {
            return Err(ChainError::Validation {
                field: "steps".to_string(),
                reason: format!("must not exceed {}, got {}", *MAX_STEPS, request.steps),
            });
        }
        if let Some(chain_size) = request.chain_size
            && chain_size > *MAX_CHAIN_SIZE
        {
            return Err(ChainError::Validation {
                field: "chain_size".to_string(),
                reason: format!("must not exceed {}, got {}", *MAX_CHAIN_SIZE, chain_size),
            });
        }
        symbol_field("symbol", &request.symbol)?;

        let time_frame = time_frame_field("time_frame", request.time_frame)?;
        let step_interval_seconds = match request.step_interval_seconds {
            Some(seconds) => validate_step_interval_seconds(seconds)?,
            None => derive_step_interval_seconds(time_frame)?,
        };

        // The only wall-clock read that reaches simulation OUTPUT. Everything
        // downstream — every simulated_at, expires_at and days_to_expiration —
        // is a function of this value and the cursor. (`created_at` and
        // `updated_at` also read the clock, but they are operational metadata
        // and never enter the tape.)
        let effective_start = to_whole_second_utc(request.start_at.unwrap_or_else(Utc::now))?;

        let schedule = ExpirationSchedule::new(
            parse_calendar(request.calendar.as_deref())?,
            parse_timezone(&request.timezone)?,
            parse_expiration_time(&request.expiration_time)?,
            request.schedules,
        )?;

        // The pinned widening ceiling is resolved here, once, from the cap
        // this instance runs and this simulation's own schedule, and stored.
        let pinned_width_ceiling = crate::domain::resolve_pinned_ceiling(&schedule)?;

        let parameters = Self {
            symbol: request.symbol,
            steps: request.steps,
            effective_start,
            step_interval_seconds,
            time_frame,
            schedule,
            tzdb_version: tzdb_version().to_string(),
            initial_price: strictly_positive_field("initial_price", request.initial_price)?,
            volatility: strictly_positive_field("volatility", request.volatility)?,
            risk_free_rate: decimal_field("risk_free_rate", request.risk_free_rate)?,
            dividend_yield: positive_field("dividend_yield", request.dividend_yield)?,
            method: request.method.try_into()?,
            chain_size: request.chain_size,
            strike_interval: request
                .strike_interval
                .map(|value| strictly_positive_field("strike_interval", value))
                .transpose()?,
            skew_slope: request
                .skew_slope
                .map(|value| decimal_field("skew_slope", value))
                .transpose()?,
            smile_curve: request
                .smile_curve
                .map(|value| decimal_field("smile_curve", value))
                .transpose()?,
            spread: request
                .spread
                .map(|value| positive_field("spread", value))
                .transpose()?,
            strike_ladder: request.strike_ladder.unwrap_or_default(),
            pinned_width_ceiling,
            // The widening coefficients are rates, not prices: zero is the
            // documented default and a negative one would NARROW a quote away
            // from the money, which is not a market anyone trades.
            // Bounded above as well as below. The caps are economic — a
            // proportional term of 1 means the spread IS the mid — and they
            // also keep the model's arithmetic inside `Decimal`'s range for any
            // price a chain can carry, which matters because `Decimal`
            // multiplication panics on overflow.
            spread_proportional: request
                .spread_proportional
                .map(|value| {
                    bounded_decimal_field("spread_proportional", value, 0.0, MAX_PROPORTIONAL)
                })
                .transpose()?,
            spread_moneyness_widening: request
                .spread_moneyness_widening
                .map(|value| {
                    bounded_decimal_field("spread_moneyness_widening", value, 0.0, MAX_WIDENING)
                })
                .transpose()?,
            spread_tenor_widening: request
                .spread_tenor_widening
                .map(|value| {
                    bounded_decimal_field("spread_tenor_widening", value, 0.0, MAX_WIDENING)
                })
                .transpose()?,
            // Strictly positive: a tick of zero is not an increment, and it is
            // the floor every bid is held above. Bounded above for the same
            // reason a tick of a thousand dollars is not a tick.
            spread_tick: request
                .spread_tick
                .map(|value| {
                    let tick = strictly_positive_field("spread_tick", value)?;
                    if value > MAX_TICK {
                        return Err(ChainError::Validation {
                            field: "spread_tick".to_string(),
                            reason: format!("must not exceed {MAX_TICK}, got {value}"),
                        });
                    }
                    Ok(tick)
                })
                .transpose()?,
            seed: request.seed.unwrap_or_else(|| rand::rng().random()),
        };

        // Run the same checks the stored-document path runs, so a request and a
        // reloaded document are held to one standard and there is one place to
        // add the next invariant. The field-specific checks above stay where
        // they are: they can name the *request* field, which `validate` cannot.
        parameters.validate()?;
        Ok(parameters)
    }
}

/// A v2 rolling simulation session.
///
/// Reuses the v1 [`SessionState`] machine, but only three of its states are
/// reachable: `Initialized → InProgress → Completed`. A v2 simulation is
/// immutable after creation, so `Modified` and `Reinitialized` — the PATCH and
/// PUT branches — cannot occur.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(try_from = "SessionV2Wire")]
pub struct SessionV2 {
    /// Unique identifier.
    pub id: Uuid,
    /// The stored-schema version this document was written under.
    #[serde(default = "default_schema_version")]
    pub schema_version: u32,
    /// When the simulation was created, in real time. Unrelated to the
    /// simulated clock, which may span years.
    pub created_at: SystemTime,
    /// When the simulation was last written, in real time.
    pub updated_at: SystemTime,
    /// The resolved parameters. Immutable for the simulation's lifetime.
    pub parameters: SimulationParametersV2,
    /// The 0-based index of the next snapshot to serve.
    pub current_step: usize,
    /// The total number of snapshots the simulation serves.
    pub total_steps: usize,
    /// The lifecycle state.
    pub state: SessionState,
    /// Optimistic-concurrency revision, bumped immediately before every
    /// compare-and-swap save, exactly as in v1.
    pub version: u64,
}

/// The deserialization shape of [`SessionV2`], validated on the way in.
///
/// Mirrors the parameters' wire type for the same reason: a stored document is
/// an outer-layer input. It additionally rejects the two states a v2 simulation
/// can never legitimately be in, a cursor past its own horizon, and — the one
/// that matters most operationally — a `schema_version` from the future, which
/// during a rolling deploy would otherwise let an old replica read a newer
/// document, drop what it does not understand, and write the truncated version
/// back with an intact revision so the compare-and-swap succeeds.
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct SessionV2Wire {
    id: Uuid,
    #[serde(default = "default_schema_version")]
    schema_version: u32,
    created_at: SystemTime,
    updated_at: SystemTime,
    parameters: SimulationParametersV2,
    current_step: usize,
    total_steps: usize,
    state: SessionState,
    version: u64,
}

impl TryFrom<SessionV2Wire> for SessionV2 {
    type Error = ChainError;

    fn try_from(wire: SessionV2Wire) -> Result<Self, Self::Error> {
        let simulation = Self {
            id: wire.id,
            schema_version: wire.schema_version,
            created_at: wire.created_at,
            updated_at: wire.updated_at,
            parameters: wire.parameters,
            current_step: wire.current_step,
            total_steps: wire.total_steps,
            state: wire.state,
            version: wire.version,
        };
        simulation.validate()?;
        Ok(simulation)
    }
}

/// The schema version assumed for a stored document written before the field
/// existed. There is no such document today — v2 has shipped with the field
/// from its first release — but defaulting keeps a future reader honest.
fn default_schema_version() -> u32 {
    SESSION_V2_SCHEMA_VERSION
}

impl SessionV2 {
    /// Creates a simulation from resolved parameters.
    ///
    /// The id is **random** (`Uuid::new_v4`), for the reason v1 already
    /// documents: [`crate::utils::UuidGenerator`] derives from a process-local counter that
    /// starts at zero, so a restarted service or a second replica would reissue
    /// the same id sequence. That was survivable while an id only had to be
    /// unique among live sessions; it is not now that persisted snapshots are
    /// filed under `(simulation, generation, step)` for as long as the retention
    /// window, where a repeated id means one run's tape silently replacing
    /// another's.
    ///
    /// The id is not an input to anything seeded — the tape and the snapshots
    /// are functions of the parameters alone — so randomising it leaves the
    /// reproducibility contract exactly where it was.
    #[must_use]
    pub fn new(parameters: SimulationParametersV2) -> Self {
        let now = SystemTime::now();
        Self {
            id: Uuid::new_v4(),
            schema_version: SESSION_V2_SCHEMA_VERSION,
            created_at: now,
            updated_at: now,
            current_step: 0,
            total_steps: parameters.steps,
            parameters,
            state: SessionState::Initialized,
            version: 0,
        }
    }

    /// Re-checks every invariant a freshly-created simulation satisfies.
    ///
    /// Called from the `Deserialize` path, so a stored document cannot present
    /// a state the lifecycle forbids.
    ///
    /// # Errors
    ///
    /// Returns [`ChainError::Validation`] when the document carries a
    /// `schema_version` this binary does not understand, a `total_steps` that
    /// disagrees with its parameters, a cursor past its own horizon, or a state
    /// unreachable for a v2 simulation (`Modified` and `Reinitialized` are the
    /// PATCH and PUT branches, and a v2 simulation is immutable). Also
    /// propagates the parameters' own validation.
    pub fn validate(&self) -> Result<(), ChainError> {
        if self.schema_version > SESSION_V2_SCHEMA_VERSION {
            return Err(ChainError::Validation {
                field: "schema_version".to_string(),
                reason: format!(
                    "document was written under schema {} but this binary understands at most {SESSION_V2_SCHEMA_VERSION}",
                    self.schema_version
                ),
            });
        }
        self.parameters.validate()?;
        if self.total_steps != self.parameters.steps {
            return Err(ChainError::Validation {
                field: "total_steps".to_string(),
                reason: format!(
                    "must equal the parameters' steps ({}), got {}",
                    self.parameters.steps, self.total_steps
                ),
            });
        }
        if self.current_step > self.total_steps {
            return Err(ChainError::Validation {
                field: "current_step".to_string(),
                reason: format!(
                    "must not exceed total_steps ({}), got {}",
                    self.total_steps, self.current_step
                ),
            });
        }
        self.validate_state()
    }

    /// Checks the lifecycle state against the cursor.
    ///
    /// A v2 simulation walks `Initialized` at cursor 0, `InProgress` while it
    /// has snapshots left, and `Completed` once the cursor reaches the horizon.
    /// Every other combination — `Completed` at step 0, `Initialized` halfway
    /// through — is a document no code path can write, so accepting one would
    /// let a corrupted or hand-edited simulation into the manager and serve
    /// snapshots from a state the rest of the code assumes away.
    fn validate_state(&self) -> Result<(), ChainError> {
        let unreachable = |reason: String| ChainError::Validation {
            field: "state".to_string(),
            reason,
        };

        match self.state {
            SessionState::Modified | SessionState::Reinitialized | SessionState::Error => {
                Err(unreachable(format!(
                    "{} is unreachable for a v2 simulation, which is immutable after creation",
                    self.state
                )))
            }
            SessionState::Initialized if self.current_step != 0 => Err(unreachable(format!(
                "{} requires a cursor of 0, got {}",
                self.state, self.current_step
            ))),
            SessionState::Completed if self.current_step != self.total_steps => {
                Err(unreachable(format!(
                    "{} requires the cursor to have reached total_steps ({}), got {}",
                    self.state, self.total_steps, self.current_step
                )))
            }
            SessionState::InProgress
                if self.current_step == 0 || self.current_step >= self.total_steps =>
            {
                Err(unreachable(format!(
                    "{} requires a cursor between 1 and total_steps ({}) exclusive, got {}",
                    self.state, self.total_steps, self.current_step
                )))
            }
            _ => Ok(()),
        }
    }

    /// The simulated instant of the snapshot the cursor currently points at.
    ///
    /// # Errors
    ///
    /// Returns [`ChainError::Validation`] when the simulated clock overflows.
    pub fn simulated_at(&self) -> Result<DateTime<Utc>, ChainError> {
        self.parameters.simulated_at(self.current_step)
    }

    /// Whether the cursor has served every snapshot.
    #[must_use]
    pub fn is_complete(&self) -> bool {
        self.current_step >= self.total_steps
    }

    /// Bumps the optimistic-concurrency revision, returning the value the
    /// caller must pass as `expected_version` to the compare-and-swap save.
    ///
    /// Mirrors the v1 helper: the caller reads a simulation, captures its
    /// `version`, mutates a clone, bumps, and saves with the captured value.
    ///
    /// # Errors
    ///
    /// Returns [`ChainError::Internal`] when the revision counter would
    /// overflow, which is unreachable in practice but must not wrap.
    pub fn bump_version(&mut self) -> Result<u64, ChainError> {
        let expected = self.version;
        self.version = self.version.checked_add(1).ok_or_else(|| {
            ChainError::Internal(format!(
                "version counter overflowed for simulation {}",
                self.id
            ))
        })?;
        self.updated_at = SystemTime::now();
        Ok(expected)
    }
}

impl fmt::Display for SessionV2 {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let json = serde_json::to_string(self).map_err(|_| fmt::Error)?;
        write!(f, "{json}")
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::api::rest::models::{ApiTimeFrame, ApiWalkType};
    use crate::domain::expiry::{ExpiryRule, ExpiryRuleKind, MAX_TARGET_COUNT};
    use chrono::{TimeZone, Weekday};
    use positive::pos_or_panic;
    use rust_decimal_macros::dec;

    /// The reference configuration from ADR 0001 §14.1, as a request.
    /// Two rules at the per-rule cap: the most expirations a schedule can keep
    /// alive, and still under the per-snapshot inventory cap.
    fn maximal_schedules() -> Vec<ExpiryRule> {
        vec![
            rule("zero_dte", ExpiryRuleKind::Daily, MAX_TARGET_COUNT),
            rule(
                "weeklies",
                ExpiryRuleKind::weekly([Weekday::Mon, Weekday::Wed, Weekday::Fri]),
                MAX_TARGET_COUNT,
            ),
        ]
    }

    fn reference_request() -> CreateSimulationRequest {
        CreateSimulationRequest {
            symbol: "SPX".to_string(),
            steps: 500,
            start_at: Some(instant(2026, 1, 5, 14, 30)),
            step_interval_seconds: Some(86_400),
            timezone: "America/New_York".to_string(),
            calendar: Some("weekdays_v1".to_string()),
            expiration_time: "17:00".to_string(),
            schedules: vec![
                rule("zero_dte", ExpiryRuleKind::Daily, 1),
                rule(
                    "weeklies",
                    ExpiryRuleKind::weekly([Weekday::Mon, Weekday::Wed, Weekday::Fri]),
                    3,
                ),
                rule(
                    "monthlies",
                    ExpiryRuleKind::Monthly {
                        weekday: Weekday::Fri,
                    },
                    12,
                ),
            ],
            initial_price: 5000.0,
            volatility: 0.18,
            risk_free_rate: 0.04,
            dividend_yield: 0.012,
            method: ApiWalkType::GeometricBrownian {
                dt: 0.004,
                drift: 0.05,
                volatility: 0.18,
            },
            time_frame: ApiTimeFrame::Day,
            chain_size: Some(15),
            strike_interval: Some(25.0),
            skew_slope: Some(-0.2),
            smile_curve: Some(0.4),
            spread: Some(0.02),
            strike_ladder: Default::default(),
            spread_proportional: None,
            spread_moneyness_widening: None,
            spread_tenor_widening: None,
            spread_tick: None,
            seed: Some(42),
        }
    }

    fn rule(id: &str, kind: ExpiryRuleKind, count: usize) -> ExpiryRule {
        match ExpiryRule::new(id, kind, count) {
            Ok(rule) => rule,
            Err(error) => panic!("test rule must be valid: {error}"),
        }
    }

    fn instant(year: i32, month: u32, day: u32, hour: u32, minute: u32) -> DateTime<Utc> {
        match Utc
            .with_ymd_and_hms(year, month, day, hour, minute, 0)
            .single()
        {
            Some(instant) => instant,
            None => panic!("test instant must be valid"),
        }
    }

    fn parameters(request: CreateSimulationRequest) -> SimulationParametersV2 {
        match SimulationParametersV2::try_from(request) {
            Ok(parameters) => parameters,
            Err(error) => panic!("the request must convert: {error}"),
        }
    }

    // ---- resolution ------------------------------------------------------

    /// The reference request converts, and every effective value is resolved.
    #[test]
    fn test_reference_request_converts_to_effective_parameters() {
        let parameters = parameters(reference_request());

        assert_eq!(parameters.symbol, "SPX");
        assert_eq!(parameters.steps, 500);
        assert_eq!(parameters.seed, 42);
        assert_eq!(parameters.effective_start, instant(2026, 1, 5, 14, 30));
        assert_eq!(parameters.step_interval_seconds, 86_400);
        assert_eq!(parameters.time_frame, TimeFrame::Day);
        assert_eq!(parameters.schedule.rules().len(), 3);
        assert_eq!(parameters.initial_price, pos_or_panic!(5000.0));
        assert!(!parameters.tzdb_version.is_empty());
    }

    /// An omitted seed is generated and surfaced, exactly as in v1.
    #[test]
    fn test_omitted_seed_is_generated_and_surfaced() {
        let mut request = reference_request();
        request.seed = None;

        // The generated seed is random, so the observable guarantee is that a
        // seed exists and is echoed — not that it has a particular value.
        let first = parameters(request.clone());
        let second = parameters(request);

        assert_ne!(
            first.seed, second.seed,
            "two unseeded requests must not share a seed"
        );
    }

    /// An omitted start is generated once, normalised to whole-second UTC, and
    /// then never re-derived.
    #[test]
    fn test_omitted_start_is_generated_once_and_normalised() {
        let mut request = reference_request();
        request.start_at = None;

        let parameters = parameters(request);

        assert_eq!(parameters.effective_start.nanosecond(), 0);
        // The start is stored, not recomputed: cursor 0 resolves back to it.
        match parameters.simulated_at(0) {
            Ok(at) => assert_eq!(at, parameters.effective_start),
            Err(error) => panic!("cursor 0 must resolve: {error}"),
        }
    }

    /// A supplied start with sub-second precision is truncated, so every
    /// timestamp renders without a fractional part.
    #[test]
    fn test_supplied_start_is_truncated_to_whole_seconds() {
        let mut request = reference_request();
        request.start_at = Some(instant(2026, 1, 5, 14, 30) + TimeDelta::milliseconds(750));

        let parameters = parameters(request);

        assert_eq!(parameters.effective_start, instant(2026, 1, 5, 14, 30));
    }

    /// An omitted interval is derived from the time frame.
    #[test]
    fn test_omitted_step_interval_is_derived_from_the_time_frame() {
        let mut request = reference_request();
        request.step_interval_seconds = None;
        request.time_frame = ApiTimeFrame::Hour;

        let parameters = parameters(request);

        assert_eq!(parameters.step_interval_seconds, 3_600);
    }

    /// A time frame shorter than a second cannot derive an interval, and says
    /// so rather than clamping to one second.
    #[test]
    fn test_sub_second_time_frame_cannot_derive_an_interval() {
        let mut request = reference_request();
        request.step_interval_seconds = None;
        request.time_frame = ApiTimeFrame::Microsecond;

        match SimulationParametersV2::try_from(request) {
            Err(ChainError::Validation { field, .. }) => {
                assert_eq!(field, "step_interval_seconds");
            }
            other => panic!("expected a validation error, got {other:?}"),
        }
    }

    /// A custom time frame coarser than a year cannot derive an interval
    /// either.
    #[test]
    fn test_custom_time_frame_beyond_a_year_cannot_derive_an_interval() {
        let mut request = reference_request();
        request.step_interval_seconds = None;
        // Half a period per year is a two-year step.
        request.time_frame = ApiTimeFrame::Custom(0.5);

        match SimulationParametersV2::try_from(request) {
            Err(ChainError::Validation { field, .. }) => {
                assert_eq!(field, "step_interval_seconds");
            }
            other => panic!("expected a validation error, got {other:?}"),
        }
    }

    /// A custom time frame within range derives cleanly.
    #[test]
    fn test_custom_time_frame_within_range_derives_an_interval() {
        let mut request = reference_request();
        request.step_interval_seconds = None;
        request.time_frame = ApiTimeFrame::Custom(365.0);

        let parameters = parameters(request);

        assert_eq!(parameters.step_interval_seconds, 86_400);
    }

    /// An explicit interval outside the accepted range is rejected.
    #[test]
    fn test_out_of_range_step_interval_is_rejected() {
        for seconds in [0, MAX_STEP_INTERVAL_SECONDS + 1] {
            let mut request = reference_request();
            request.step_interval_seconds = Some(seconds);

            match SimulationParametersV2::try_from(request) {
                Err(ChainError::Validation { field, .. }) => {
                    assert_eq!(field, "step_interval_seconds");
                }
                other => panic!("expected a validation error for {seconds}, got {other:?}"),
            }
        }
    }

    // ---- the simulated clock --------------------------------------------

    /// `simulated_at` is `effective_start + cursor × interval`, and never reads
    /// the wall clock.
    #[test]
    fn test_simulated_at_is_start_plus_cursor_times_interval() {
        let parameters = parameters(reference_request());

        match (parameters.simulated_at(0), parameters.simulated_at(3)) {
            (Ok(first), Ok(fourth)) => {
                assert_eq!(first, instant(2026, 1, 5, 14, 30));
                assert_eq!(fourth, instant(2026, 1, 8, 14, 30));
            }
            (first, fourth) => panic!("both cursors must resolve: {first:?} {fourth:?}"),
        }
    }

    /// The same parameters derive the same instant for every cursor, call after
    /// call.
    #[test]
    fn test_simulated_at_is_stable_across_calls() {
        let parameters = parameters(reference_request());

        for cursor in [0, 1, 17, 499] {
            match (
                parameters.simulated_at(cursor),
                parameters.simulated_at(cursor),
            ) {
                (Ok(first), Ok(second)) => assert_eq!(first, second),
                (first, second) => panic!("cursor {cursor} must resolve: {first:?} {second:?}"),
            }
        }
    }

    /// The horizon is one interval past the last served snapshot.
    #[test]
    fn test_simulated_end_is_one_interval_past_the_last_step() {
        let mut request = reference_request();
        request.steps = 3;
        let parameters = parameters(request);

        match parameters.simulated_end() {
            Ok(end) => assert_eq!(end, instant(2026, 1, 8, 14, 30)),
            Err(error) => panic!("the horizon must resolve: {error}"),
        }
    }

    /// A cursor that would overflow the simulated clock is a typed error, never
    /// a wrapped timestamp.
    #[test]
    fn test_simulated_at_overflow_is_a_typed_error() {
        let mut parameters = parameters(reference_request());
        parameters.step_interval_seconds = MAX_STEP_INTERVAL_SECONDS;

        match parameters.simulated_at(usize::MAX) {
            Err(ChainError::Validation { field, reason }) => {
                assert_eq!(field, "steps");
                assert!(reason.contains("overflow"));
            }
            other => panic!("expected an overflow error, got {other:?}"),
        }
    }

    // ---- validation ------------------------------------------------------

    /// Each invalid field is rejected by name, and no path panics.
    #[test]
    fn test_invalid_fields_are_rejected_by_name() {
        /// One invalid-field case: the field the error must name, and the
        /// mutation that makes the request invalid.
        type Case = (&'static str, Box<dyn Fn(&mut CreateSimulationRequest)>);

        let cases: Vec<Case> = vec![
            (
                "steps",
                Box::new(|r: &mut CreateSimulationRequest| r.steps = 0),
            ),
            (
                "symbol",
                Box::new(|r: &mut CreateSimulationRequest| r.symbol = "bad symbol!".to_string()),
            ),
            (
                "timezone",
                Box::new(|r: &mut CreateSimulationRequest| r.timezone = "Mars/Olympus".to_string()),
            ),
            (
                "expiration_time",
                Box::new(|r: &mut CreateSimulationRequest| r.expiration_time = "25:00".to_string()),
            ),
            (
                "calendar",
                Box::new(|r: &mut CreateSimulationRequest| {
                    r.calendar = Some("weekdays_v9".to_string())
                }),
            ),
            (
                "initial_price",
                Box::new(|r: &mut CreateSimulationRequest| r.initial_price = 0.0),
            ),
            (
                "volatility",
                Box::new(|r: &mut CreateSimulationRequest| r.volatility = f64::NAN),
            ),
            (
                "risk_free_rate",
                Box::new(|r: &mut CreateSimulationRequest| r.risk_free_rate = f64::INFINITY),
            ),
            (
                "dividend_yield",
                Box::new(|r: &mut CreateSimulationRequest| r.dividend_yield = -1.0),
            ),
            (
                "strike_interval",
                Box::new(|r: &mut CreateSimulationRequest| r.strike_interval = Some(0.0)),
            ),
            (
                "chain_size",
                Box::new(|r: &mut CreateSimulationRequest| r.chain_size = Some(usize::MAX)),
            ),
        ];

        for (field, mutate) in cases {
            let mut request = reference_request();
            mutate(&mut request);

            match SimulationParametersV2::try_from(request) {
                Err(ChainError::Validation { field: named, .. }) => {
                    assert_eq!(named, field, "wrong field named for {field}");
                }
                other => panic!("expected a validation error for {field}, got {other:?}"),
            }
        }
    }

    /// An empty schedule is rejected through the same domain validation the
    /// planner uses.
    #[test]
    fn test_empty_schedule_is_rejected() {
        let mut request = reference_request();
        request.schedules = Vec::new();

        match SimulationParametersV2::try_from(request) {
            Err(ChainError::Validation { field, .. }) => assert_eq!(field, "schedules"),
            other => panic!("expected a validation error, got {other:?}"),
        }
    }

    /// `HH:MM:SS` is accepted as well as `HH:MM`.
    #[test]
    fn test_expiration_time_accepts_both_precisions() {
        for raw in ["17:00", "17:00:00"] {
            let mut request = reference_request();
            request.expiration_time = raw.to_string();

            let parameters = parameters(request);
            match NaiveTime::from_hms_opt(17, 0, 0) {
                Some(expected) => {
                    assert_eq!(parameters.schedule.expiration_time(), expected)
                }
                None => panic!("17:00:00 must be a valid time"),
            }
        }
    }

    /// An omitted calendar defaults to the only supported policy.
    #[test]
    fn test_omitted_calendar_defaults_to_weekdays_v1() {
        let mut request = reference_request();
        request.calendar = None;

        let parameters = parameters(request);

        assert_eq!(parameters.schedule.calendar(), CalendarVersion::WeekdaysV1);
    }

    // ---- persistence shape ----------------------------------------------

    /// The parameters round-trip through serde, which is what the stores rely
    /// on.
    #[test]
    fn test_parameters_round_trip_through_serde() {
        let parameters = parameters(reference_request());

        let json = match serde_json::to_string(&parameters) {
            Ok(json) => json,
            Err(error) => panic!("must serialize: {error}"),
        };
        match serde_json::from_str::<SimulationParametersV2>(&json) {
            Ok(round_tripped) => assert_eq!(round_tripped, parameters),
            Err(error) => panic!("must deserialize: {error}"),
        }
    }

    /// A simulation created WITH a spread model round-trips through the store.
    ///
    /// The outer fields skip serializing when absent and the wire struct denies
    /// unknown ones, so a field present on one and missing from the other would
    /// produce a document that writes fine and can never be read back. With the
    /// reference request every coefficient is `None`, which is exactly the case
    /// that would not catch it.
    #[test]
    fn test_stored_parameters_keep_the_spread_model() {
        let mut request = reference_request();
        request.spread = Some(0.03);
        request.spread_proportional = Some(0.02);
        request.spread_moneyness_widening = Some(0.5);
        request.spread_tenor_widening = Some(0.1);
        request.spread_tick = Some(0.05);
        let parameters = parameters(request);

        let json = match serde_json::to_string(&parameters) {
            Ok(json) => json,
            Err(error) => panic!("must serialize: {error}"),
        };
        match serde_json::from_str::<SimulationParametersV2>(&json) {
            Ok(loaded) => {
                assert_eq!(loaded, parameters);
                assert_eq!(loaded.spread_proportional, Some(dec!(0.02)));
                assert_eq!(loaded.spread_tick, Some(pos_or_panic!(0.05)));
            }
            Err(error) => panic!("a simulation with a spread model must reload: {error}"),
        }
    }

    /// A stored document with an impossible spread model is refused on load.
    ///
    /// Redis is an outer layer: a hand-edited `spread_tick` of zero would
    /// silently disable the floor that keeps a wing quotable.
    #[test]
    fn test_stored_parameters_reject_an_impossible_spread_model() {
        let parameters = parameters(reference_request());
        let json = match serde_json::to_value(&parameters) {
            Ok(serde_json::Value::Object(mut map)) => {
                map.insert("spread_tick".to_string(), serde_json::json!("0"));
                serde_json::Value::Object(map)
            }
            other => panic!("the parameters must serialize to an object, got {other:?}"),
        };

        match serde_json::from_value::<SimulationParametersV2>(json) {
            Ok(loaded) => panic!("a zero tick must be refused, got {loaded:?}"),
            Err(error) => assert!(
                error.to_string().contains("spread_tick"),
                "the failure must name the field: {error}"
            ),
        }
    }

    /// A pinned ladder without an interval is refused, on both paths.
    ///
    /// There is no fixed grid to pin without one, and resolving it to some
    /// expiration's derived interval would be a number the client never chose.
    #[test]
    fn test_a_pinned_ladder_requires_an_explicit_interval() {
        let mut request = reference_request();
        request.strike_ladder = Some(StrikeLadder::Pinned);
        request.strike_interval = None;

        match SimulationParametersV2::try_from(request) {
            Ok(parameters) => {
                panic!("a pinned ladder with no grid must be refused: {parameters:?}")
            }
            Err(ChainError::Validation { field, .. }) => assert_eq!(field, "strike_interval"),
            Err(error) => panic!("expected a validation failure, got {error:?}"),
        }

        // And the stored path, since Redis is an outer layer.
        let mut pinned = reference_request();
        pinned.strike_ladder = Some(StrikeLadder::Pinned);
        pinned.strike_interval = Some(25.0);
        let parameters = parameters(pinned);

        let json = match serde_json::to_value(&parameters) {
            Ok(serde_json::Value::Object(mut map)) => {
                map.remove("strike_interval");
                serde_json::Value::Object(map)
            }
            other => panic!("the parameters must serialize to an object, got {other:?}"),
        };
        match serde_json::from_value::<SimulationParametersV2>(json) {
            Ok(loaded) => panic!("a stored pinned ladder with no grid must be refused: {loaded:?}"),
            Err(error) => assert!(
                error.to_string().contains("strike_interval"),
                "the failure must name the field: {error}"
            ),
        }
    }

    /// The ladder survives a store round trip, and defaults to rolling.
    #[test]
    fn test_the_strike_ladder_round_trips() {
        let mut request = reference_request();
        request.strike_ladder = Some(StrikeLadder::Pinned);
        request.strike_interval = Some(25.0);
        let pinned = parameters(request);

        assert_eq!(pinned.strike_ladder, StrikeLadder::Pinned);

        let json = match serde_json::to_string(&pinned) {
            Ok(json) => json,
            Err(error) => panic!("must serialize: {error}"),
        };
        match serde_json::from_str::<SimulationParametersV2>(&json) {
            Ok(loaded) => assert_eq!(loaded.strike_ladder, StrikeLadder::Pinned),
            Err(error) => panic!("must deserialize: {error}"),
        }

        // A request that says nothing gets the behaviour the service had.
        let untouched = parameters(reference_request());
        assert_eq!(untouched.strike_ladder, StrikeLadder::Rolling);
    }

    /// A document written before the field still loads, as rolling.
    #[test]
    fn test_stored_parameters_without_a_strike_ladder_still_load() {
        let parameters = parameters(reference_request());
        let json = match serde_json::to_value(&parameters) {
            Ok(serde_json::Value::Object(mut map)) => {
                map.remove("strike_ladder");
                serde_json::Value::Object(map)
            }
            other => panic!("the parameters must serialize to an object, got {other:?}"),
        };

        match serde_json::from_value::<SimulationParametersV2>(json) {
            Ok(loaded) => assert_eq!(
                loaded.strike_ladder,
                StrikeLadder::Rolling,
                "an older document must read as the behaviour it was written under"
            ),
            Err(error) => panic!("an older document must still load: {error}"),
        }
    }

    /// A document written before the ceiling was stored still loads, and gets
    /// the ceiling it would have had, resolved once on the way in.
    #[test]
    fn test_stored_parameters_without_a_pinned_ceiling_still_load() {
        let parameters = parameters(reference_request());
        let json = match serde_json::to_value(&parameters) {
            Ok(serde_json::Value::Object(mut map)) => {
                assert!(
                    map.remove("pinned_width_ceiling").is_some(),
                    "the ceiling must be part of the stored shape"
                );
                serde_json::Value::Object(map)
            }
            other => panic!("the parameters must serialize to an object, got {other:?}"),
        };

        match serde_json::from_value::<SimulationParametersV2>(json) {
            Ok(loaded) => assert_eq!(
                loaded.pinned_width_ceiling, parameters.pinned_width_ceiling,
                "an older document must read as the ceiling it would have had"
            ),
            Err(error) => panic!("an older document must still load: {error}"),
        }
    }

    /// A stored ceiling survives the round trip rather than being re-resolved,
    /// which is the whole point: a session created under a tighter cap keeps
    /// that ceiling on an instance whose own cap is wider.
    #[test]
    fn test_a_stored_pinned_ceiling_survives_the_round_trip() {
        let mut parameters = parameters(reference_request());
        parameters.pinned_width_ceiling = 7;

        let json = match serde_json::to_string(&parameters) {
            Ok(json) => json,
            Err(error) => panic!("must serialize: {error}"),
        };
        match serde_json::from_str::<SimulationParametersV2>(&json) {
            Ok(loaded) => assert_eq!(loaded.pinned_width_ceiling, 7),
            Err(error) => panic!("must deserialize: {error}"),
        }
    }

    /// A corrupted document cannot buy a widening no request could ask for.
    ///
    /// Redis is an outer layer: the stored ceiling is a number the domain then
    /// obeys, so a hand-edited document raising it past the absolute maximum
    /// would bypass the guard on how much the service prices per step.
    #[test]
    fn test_a_stored_pinned_ceiling_above_the_maximum_is_refused() {
        let parameters = parameters(reference_request());

        let json = match serde_json::to_value(&parameters) {
            Ok(serde_json::Value::Object(mut map)) => {
                map.insert(
                    "pinned_width_ceiling".to_string(),
                    serde_json::json!(MAX_PINNED_WIDTH + 1),
                );
                serde_json::Value::Object(map)
            }
            other => panic!("the parameters must serialize to an object, got {other:?}"),
        };

        match serde_json::from_value::<SimulationParametersV2>(json) {
            Ok(loaded) => panic!(
                "a ceiling of {} must be refused, loaded {loaded:?}",
                MAX_PINNED_WIDTH + 1
            ),
            Err(error) => {
                let rendered = error.to_string();
                assert!(
                    rendered.contains("pinned_width_ceiling"),
                    "the failure must name the field: {rendered}"
                );
            }
        }
    }

    /// A session created under a higher per-snapshot cap keeps LOADING on an
    /// instance running a lower one.
    ///
    /// The cap is admission control, a decision about what to accept next. A
    /// session that already exists answered it at creation, and re-answering it
    /// on load would mean a smaller instance cannot even deserialize a tape a
    /// bigger one is serving, which is the cross-instance breakage #109 exists
    /// to remove. The creation path still refuses the same configuration.
    #[test]
    fn test_a_stored_simulation_over_the_cap_still_loads_but_is_not_creatable() {
        let mut parameters = parameters(reference_request());

        // 1 001 strikes across 512 live expirations is half a million
        // contracts a snapshot, far above any sane cap.
        parameters.chain_size = Some(*MAX_CHAIN_SIZE);
        let rules = match (
            ExpiryRule::new("dailies", ExpiryRuleKind::Daily, 256),
            ExpiryRule::new("more_dailies", ExpiryRuleKind::Daily, 256),
        ) {
            (Ok(first), Ok(second)) => vec![first, second],
            (first, second) => panic!("the test rules must be valid: {first:?} {second:?}"),
        };
        parameters.schedule = match ExpirationSchedule::new(
            parameters.schedule.calendar(),
            parameters.schedule.timezone(),
            parameters.schedule.expiration_time(),
            rules,
        ) {
            Ok(schedule) => schedule,
            Err(error) => panic!("the test schedule must be valid: {error}"),
        };

        match parameters.validate() {
            Ok(()) => panic!("a request for that much work must be refused at creation"),
            Err(ChainError::Validation { field, reason }) => {
                assert_eq!(field, "chain_size");
                assert!(reason.contains("above the"), "{reason}");
            }
            Err(error) => panic!("expected a validation failure, got {error:?}"),
        }

        match parameters.validate_stored() {
            Ok(()) => {}
            Err(error) => panic!(
                "a session that already exists must keep loading on a smaller instance: {error}"
            ),
        }
    }

    /// A stored coefficient above its cap is refused on load.
    ///
    /// Redis is an outer layer: a document that never passed through a request
    /// must clear the same bar, or a hand-edited one prices a chain the REST
    /// boundary would refuse to create.
    #[test]
    fn test_stored_parameters_reject_a_coefficient_above_its_cap() {
        let parameters = parameters(reference_request());

        for (field, over_cap) in [
            ("spread_proportional", MAX_PROPORTIONAL + 1.0),
            ("spread_moneyness_widening", MAX_WIDENING + 1.0),
            ("spread_tenor_widening", MAX_WIDENING + 1.0),
            ("spread_tick", MAX_TICK + 1.0),
        ] {
            let json = match serde_json::to_value(&parameters) {
                Ok(serde_json::Value::Object(mut map)) => {
                    map.insert(field.to_string(), serde_json::json!(over_cap.to_string()));
                    serde_json::Value::Object(map)
                }
                other => panic!("the parameters must serialize to an object, got {other:?}"),
            };

            match serde_json::from_value::<SimulationParametersV2>(json) {
                Ok(loaded) => panic!("{field} above its cap must be refused, got {loaded:?}"),
                Err(error) => assert!(
                    error.to_string().contains(field),
                    "the failure must name {field}: {error}"
                ),
            }
        }
    }

    /// A coefficient exactly at its cap still loads.
    ///
    /// The bound is inclusive on both paths, so the stored check cannot be
    /// stricter than the request that produced the document.
    #[test]
    fn test_stored_parameters_accept_a_coefficient_at_its_cap() {
        let mut request = reference_request();
        request.spread_proportional = Some(MAX_PROPORTIONAL);
        request.spread_moneyness_widening = Some(MAX_WIDENING);
        request.spread_tick = Some(MAX_TICK);
        let parameters = parameters(request);

        let json = match serde_json::to_string(&parameters) {
            Ok(json) => json,
            Err(error) => panic!("must serialize: {error}"),
        };
        match serde_json::from_str::<SimulationParametersV2>(&json) {
            Ok(loaded) => assert_eq!(loaded, parameters),
            Err(error) => panic!("a document at the cap must load: {error}"),
        }
    }

    /// A document written before the spread model still loads.
    ///
    /// The four coefficients are `#[serde(default)]`, so a simulation stored by
    /// an older binary keeps deserializing and reads as the single scalar it
    /// was created with. Losing that would strand every live simulation on a
    /// deploy.
    #[test]
    fn test_stored_parameters_without_the_spread_model_still_load() {
        let parameters = parameters(reference_request());
        let json = match serde_json::to_value(&parameters) {
            Ok(serde_json::Value::Object(mut map)) => {
                for field in [
                    "spread_proportional",
                    "spread_moneyness_widening",
                    "spread_tenor_widening",
                    "spread_tick",
                ] {
                    map.remove(field);
                }
                serde_json::Value::Object(map)
            }
            other => panic!("the parameters must serialize to an object, got {other:?}"),
        };

        match serde_json::from_value::<SimulationParametersV2>(json) {
            Ok(loaded) => {
                assert_eq!(loaded.spread, parameters.spread, "the scalar survives");
                assert_eq!(loaded.spread_proportional, None);
                assert_eq!(loaded.spread_moneyness_widening, None);
                assert_eq!(loaded.spread_tenor_widening, None);
                assert_eq!(loaded.spread_tick, None);
            }
            Err(error) => panic!("an older document must still load: {error}"),
        }
    }

    /// The spread coefficients are bounded at the boundary, both ways.
    ///
    /// A negative widening term would NARROW a quote away from the money, a
    /// tick of zero is not an increment, and an unbounded coefficient would
    /// reach `Decimal`'s range against a large mid.
    #[test]
    fn test_an_out_of_range_spread_coefficient_is_refused() {
        for (field, mutate) in [
            (
                "spread_proportional",
                (|request: &mut CreateSimulationRequest| request.spread_proportional = Some(-0.01))
                    as fn(&mut CreateSimulationRequest),
            ),
            ("spread_moneyness_widening", |request| {
                request.spread_moneyness_widening = Some(-0.5)
            }),
            ("spread_tenor_widening", |request| {
                request.spread_tenor_widening = Some(-0.2)
            }),
            ("spread_tick", |request| request.spread_tick = Some(0.0)),
            ("spread_proportional", |request| {
                request.spread_proportional = Some(MAX_PROPORTIONAL + 1.0)
            }),
            ("spread_moneyness_widening", |request| {
                request.spread_moneyness_widening = Some(MAX_WIDENING + 1.0)
            }),
            ("spread_tick", |request| {
                request.spread_tick = Some(MAX_TICK + 1.0)
            }),
        ] {
            let mut request = reference_request();
            mutate(&mut request);

            match SimulationParametersV2::try_from(request) {
                Ok(parameters) => panic!("{field} must be refused, got {parameters:?}"),
                Err(ChainError::Validation { field: named, .. }) => assert_eq!(named, field),
                Err(error) => panic!("expected a validation failure for {field}, got {error:?}"),
            }
        }
    }

    /// A stored simulation round-trips, keeping its schema version, cursor and
    /// revision.
    #[test]
    fn test_simulation_round_trips_through_serde() {
        let simulation = SessionV2::new(parameters(reference_request()));

        let json = match serde_json::to_string(&simulation) {
            Ok(json) => json,
            Err(error) => panic!("must serialize: {error}"),
        };
        match serde_json::from_str::<SessionV2>(&json) {
            Ok(round_tripped) => {
                assert_eq!(round_tripped, simulation);
                assert_eq!(round_tripped.schema_version, SESSION_V2_SCHEMA_VERSION);
            }
            Err(error) => panic!("must deserialize: {error}"),
        }
    }

    /// The stored document carries an explicit schema version, so a future
    /// migration can tell documents apart without inferring from their shape.
    #[test]
    fn test_stored_document_carries_an_explicit_schema_version() {
        let simulation = SessionV2::new(parameters(reference_request()));

        let value = match serde_json::to_value(&simulation) {
            Ok(value) => value,
            Err(error) => panic!("must serialize: {error}"),
        };
        assert_eq!(
            value
                .get("schema_version")
                .and_then(serde_json::Value::as_u64),
            Some(u64::from(SESSION_V2_SCHEMA_VERSION))
        );
    }

    /// The stored schedule keeps its documented wire shape.
    ///
    /// This is the *storage* form: the schedule nests under `schedule` with its
    /// rules under `rules`. ADR 0001 §14.2 shows the *response* form, which is
    /// flat inside `parameters` with the array named `schedules`; #47 owns that
    /// mapping.
    #[test]
    fn test_stored_schedule_keeps_the_documented_shape() {
        let parameters = parameters(reference_request());

        let value = match serde_json::to_value(&parameters) {
            Ok(value) => value,
            Err(error) => panic!("must serialize: {error}"),
        };
        let schedule = match value.get("schedule") {
            Some(schedule) => schedule,
            None => panic!("the parameters must carry a schedule"),
        };
        assert_eq!(
            schedule.get("timezone").and_then(serde_json::Value::as_str),
            Some("America/New_York")
        );
        assert_eq!(
            schedule.get("calendar").and_then(serde_json::Value::as_str),
            Some("weekdays_v1")
        );
        assert_eq!(
            schedule
                .get("expiration_time")
                .and_then(serde_json::Value::as_str),
            Some("17:00:00")
        );
    }

    // ---- stored input is not trusted -------------------------------------

    /// Deserialization runs the same validation as the request path.
    ///
    /// Without it a hand-edited or corrupted document in Redis sails past every
    /// check: `rules/global_rules.md` is explicit that a domain type must not
    /// trust an outer layer, and Redis is one.
    #[test]
    fn test_stored_parameters_are_validated_on_load() {
        let parameters = parameters(reference_request());
        let json = match serde_json::to_string(&parameters) {
            Ok(json) => json,
            Err(error) => panic!("must serialize: {error}"),
        };

        // Each tamper is a field the request path checks and a stored document
        // could otherwise smuggle past.
        let tampers = [
            // A zero interval freezes the simulated clock: every snapshot would
            // carry the same instant, the cutoff would never advance, and the
            // rolling inventory would never roll.
            (
                r#""step_interval_seconds":86400"#,
                r#""step_interval_seconds":0"#,
                "step_interval_seconds",
            ),
            // A steps count above the cap drives an unbounded factor tape.
            (r#""steps":500"#, r#""steps":100000000"#, "steps"),
            // A sub-second start breaks the whole-second rendering that makes
            // exports byte-comparable.
            (
                r#""effective_start":"2026-01-05T14:30:00Z""#,
                r#""effective_start":"2026-01-05T14:30:00.5Z""#,
                "effective_start",
            ),
            // A symbol carrying the CSV separators would corrupt the export
            // column the rule-id charset was narrowed to protect.
            (r#""symbol":"SPX""#, r#""symbol":"SPX,\"x\"|y""#, "symbol"),
            // A chain size above the cap drives an unbounded strike ladder.
            (r#""chain_size":15"#, r#""chain_size":100000"#, "chain_size"),
            // `Positive` admits zero, so these three survive the type and have
            // to be rejected by the validator: a zero price or volatility is a
            // chain of worthless options, and a zero interval collapses every
            // strike onto one. Every `Positive` is quoted since `positive` 0.6
            // writes the exact decimal as a string.
            (
                r#""initial_price":"5000""#,
                r#""initial_price":"0""#,
                "initial_price",
            ),
            (
                r#""tzdb_version":"2025b","initial_price":"5000","volatility":"0.18""#,
                r#""tzdb_version":"2025b","initial_price":"5000","volatility":"0""#,
                "volatility",
            ),
            (
                r#""strike_interval":"25""#,
                r#""strike_interval":"0""#,
                "strike_interval",
            ),
            // The walk's own invariants are not re-derived here: the stored
            // method round-trips through the same mirror the request path
            // validates with, so a `dt` of zero is caught by that check.
            (r#""dt":"0.004""#, r#""dt":"0.0""#, "dt"),
            // A simulation has exactly one base volatility, and the check that
            // enforces it has to hold on the stored path too — otherwise a
            // document can carry a top-level value the walk never uses. The
            // anchor is needed because "volatility":0.18 appears twice.
            (
                r#""tzdb_version":"2025b","initial_price":"5000","volatility":"0.18""#,
                r#""tzdb_version":"2025b","initial_price":"5000","volatility":"0.25""#,
                "volatility",
            ),
        ];

        for (from, to, field) in tampers {
            let tampered = json.replace(from, to);
            assert_ne!(tampered, json, "the tamper for {field} must have applied");

            let error = match serde_json::from_str::<SimulationParametersV2>(&tampered) {
                Ok(_) => panic!("a tampered {field} must be rejected on load"),
                Err(error) => error.to_string(),
            };
            assert!(
                error.contains(field),
                "the error must name {field}, got {error}"
            );
        }
    }

    /// A configuration whose snapshots would price more contracts than the cap
    /// is refused at creation, naming the field a client can lower.
    #[test]
    fn test_a_configuration_above_the_snapshot_contract_cap_is_rejected() {
        let mut request = reference_request();
        // 500 is the chain-size cap: 1 001 strikes. Two rules at the per-rule
        // cap keep 512 expirations alive, which is the per-snapshot inventory
        // cap. Their product is half a million contracts, and neither field
        // alone is out of range.
        request.chain_size = Some(500);
        request.schedules = maximal_schedules();

        match SimulationParametersV2::try_from(request) {
            Err(ChainError::Validation { field, reason }) => {
                assert_eq!(field, "chain_size");
                assert!(
                    reason.contains("would price"),
                    "the reason must say what it refused, got {reason}"
                );
            }
            other => panic!("the product of the two caps must be refused, got {other:?}"),
        }
    }

    /// The reference configuration is nowhere near the cap, so the bound costs
    /// a realistic client nothing.
    #[test]
    fn test_the_reference_configuration_is_far_below_the_contract_cap() {
        match SimulationParametersV2::try_from(reference_request()) {
            Ok(parameters) => match parameters.validate() {
                Ok(()) => {}
                Err(error) => panic!("the reference configuration must validate: {error}"),
            },
            Err(error) => panic!("the reference request must convert: {error}"),
        }
    }

    /// An unknown field in a stored document is an error, not a silent drop.
    ///
    /// During a rolling deploy an old replica would otherwise read a newer
    /// document, discard what it does not understand, and write the truncated
    /// version back with an intact revision — so the compare-and-swap succeeds
    /// and the data is simply gone.
    #[test]
    fn test_stored_parameters_reject_an_unknown_field() {
        let parameters = parameters(reference_request());
        let json = match serde_json::to_string(&parameters) {
            Ok(json) => json,
            Err(error) => panic!("must serialize: {error}"),
        };
        let tampered = json.replace(r#""symbol":"SPX""#, r#""symbol":"SPX","a_future_field":1"#);

        assert!(serde_json::from_str::<SimulationParametersV2>(&tampered).is_err());
    }

    /// A stored simulation from a newer schema is refused rather than silently
    /// downgraded.
    #[test]
    fn test_stored_simulation_rejects_a_future_schema_version() {
        let simulation = SessionV2::new(parameters(reference_request()));
        let json = match serde_json::to_string(&simulation) {
            Ok(json) => json,
            Err(error) => panic!("must serialize: {error}"),
        };
        let tampered = json.replace(
            &format!(r#""schema_version":{SESSION_V2_SCHEMA_VERSION}"#),
            r#""schema_version":99"#,
        );

        let error = match serde_json::from_str::<SessionV2>(&tampered) {
            Ok(_) => panic!("a future schema version must be rejected"),
            Err(error) => error.to_string(),
        };
        assert!(error.contains("schema_version"), "got {error}");
    }

    /// A stored simulation in a state the v2 lifecycle cannot reach is refused.
    #[test]
    fn test_stored_simulation_rejects_an_unreachable_state() {
        let simulation = SessionV2::new(parameters(reference_request()));
        let json = match serde_json::to_string(&simulation) {
            Ok(json) => json,
            Err(error) => panic!("must serialize: {error}"),
        };

        for state in ["Modified", "Reinitialized", "Error"] {
            let tampered =
                json.replace(r#""state":"Initialized""#, &format!(r#""state":"{state}""#));
            assert_ne!(tampered, json, "the tamper for {state} must have applied");

            let error = match serde_json::from_str::<SessionV2>(&tampered) {
                Ok(_) => panic!("{state} must be rejected for a v2 simulation"),
                Err(error) => error.to_string(),
            };
            assert!(error.contains("state"), "got {error}");
        }
    }

    /// A state that contradicts the cursor is refused.
    ///
    /// `Completed` at step 0 and `Initialized` halfway through are documents no
    /// code path writes, so accepting one would let a corrupted simulation
    /// serve snapshots from a state the rest of the code assumes away.
    #[test]
    fn test_stored_simulation_rejects_a_state_the_cursor_contradicts() {
        let mut request = reference_request();
        request.steps = 4;
        let simulation = SessionV2::new(parameters(request));
        let json = match serde_json::to_string(&simulation) {
            Ok(json) => json,
            Err(error) => panic!("must serialize: {error}"),
        };

        let contradictions = [
            (
                r#""current_step":0,"total_steps":4,"state":"Completed""#,
                "Completed at step 0",
            ),
            (
                r#""current_step":2,"total_steps":4,"state":"Initialized""#,
                "Initialized mid-run",
            ),
            (
                r#""current_step":0,"total_steps":4,"state":"InProgress""#,
                "InProgress at step 0",
            ),
            (
                r#""current_step":4,"total_steps":4,"state":"InProgress""#,
                "InProgress at the horizon",
            ),
        ];

        for (replacement, what) in contradictions {
            let tampered = json.replace(
                r#""current_step":0,"total_steps":4,"state":"Initialized""#,
                replacement,
            );
            assert_ne!(tampered, json, "the tamper for {what} must have applied");

            let error = match serde_json::from_str::<SessionV2>(&tampered) {
                Ok(_) => panic!("{what} must be rejected"),
                Err(error) => error.to_string(),
            };
            assert!(
                error.contains("state"),
                "{what} must name state, got {error}"
            );
        }
    }

    /// A cursor past its own horizon is refused, so no caller has to defend
    /// against one.
    #[test]
    fn test_stored_simulation_rejects_a_cursor_past_the_horizon() {
        let mut request = reference_request();
        request.steps = 2;
        let simulation = SessionV2::new(parameters(request));
        let json = match serde_json::to_string(&simulation) {
            Ok(json) => json,
            Err(error) => panic!("must serialize: {error}"),
        };
        let tampered = json.replace(r#""current_step":0"#, r#""current_step":9999"#);

        let error = match serde_json::from_str::<SessionV2>(&tampered) {
            Ok(_) => panic!("a cursor past the horizon must be rejected"),
            Err(error) => error.to_string(),
        };
        assert!(error.contains("current_step"), "got {error}");
    }

    /// A `total_steps` that disagrees with the parameters is refused.
    #[test]
    fn test_stored_simulation_rejects_a_mismatched_total_steps() {
        let simulation = SessionV2::new(parameters(reference_request()));
        let json = match serde_json::to_string(&simulation) {
            Ok(json) => json,
            Err(error) => panic!("must serialize: {error}"),
        };
        let tampered = json.replace(r#""total_steps":500"#, r#""total_steps":7"#);

        let error = match serde_json::from_str::<SessionV2>(&tampered) {
            Ok(_) => panic!("a mismatched total_steps must be rejected"),
            Err(error) => error.to_string(),
        };
        assert!(error.contains("total_steps"), "got {error}");
    }

    /// A valid stored document still loads, so the validation does not reject
    /// what it should accept.
    #[test]
    fn test_a_valid_stored_simulation_still_loads() {
        let simulation = SessionV2::new(parameters(reference_request()));
        let json = match serde_json::to_string(&simulation) {
            Ok(json) => json,
            Err(error) => panic!("must serialize: {error}"),
        };

        match serde_json::from_str::<SessionV2>(&json) {
            Ok(loaded) => assert_eq!(loaded, simulation),
            Err(error) => panic!("a valid document must load: {error}"),
        }
    }

    // ---- lifecycle -------------------------------------------------------

    /// A new simulation starts at cursor zero, revision zero, Initialized.
    #[test]
    fn test_new_simulation_starts_initialized_at_cursor_zero() {
        let simulation = SessionV2::new(parameters(reference_request()));

        assert_eq!(simulation.current_step, 0);
        assert_eq!(simulation.total_steps, 500);
        assert_eq!(simulation.version, 0);
        assert_eq!(simulation.state, SessionState::Initialized);
        assert!(!simulation.is_complete());
    }

    /// Bumping the revision returns the value a compare-and-swap must expect.
    #[test]
    fn test_bump_version_returns_the_expected_revision() {
        let mut simulation = SessionV2::new(parameters(reference_request()));

        match simulation.bump_version() {
            Ok(expected) => {
                assert_eq!(expected, 0);
                assert_eq!(simulation.version, 1);
            }
            Err(error) => panic!("must bump: {error}"),
        }
    }

    /// A revision counter at its maximum is an error rather than a wrap that
    /// would let a stale writer pass the compare-and-swap.
    #[test]
    fn test_bump_version_overflow_is_a_typed_error() {
        let mut simulation = SessionV2::new(parameters(reference_request()));
        simulation.version = u64::MAX;

        match simulation.bump_version() {
            Err(ChainError::Internal(reason)) => assert!(reason.contains("overflow")),
            other => panic!("expected an internal error, got {other:?}"),
        }
    }

    /// Completion is a cursor comparison, not a stored flag.
    #[test]
    fn test_is_complete_tracks_the_cursor() {
        let mut request = reference_request();
        request.steps = 2;
        let mut simulation = SessionV2::new(parameters(request));

        assert!(!simulation.is_complete());
        simulation.current_step = 2;
        assert!(simulation.is_complete());
    }

    /// The simulation's own `simulated_at` follows its cursor.
    #[test]
    fn test_simulation_simulated_at_follows_the_cursor() {
        let mut simulation = SessionV2::new(parameters(reference_request()));
        simulation.current_step = 2;

        match simulation.simulated_at() {
            Ok(at) => assert_eq!(at, instant(2026, 1, 7, 14, 30)),
            Err(error) => panic!("must resolve: {error}"),
        }
    }

    /// The same wire change on v2's largest stored payload: a `Historical`
    /// method's `prices: Vec<Positive>`, written as bare JSON numbers.
    ///
    /// `Historical` carries no model volatility, so it is also the variant
    /// that skips the "one base volatility" cross-check — the load has to
    /// succeed for a different reason than the synthetic models do.
    #[test]
    fn test_a_stored_historical_price_series_still_loads() {
        const STORED_BY_AN_OLDER_BINARY: &str = concat!(
            r#"{"id":"6ba7b813-9dad-11d1-80b4-00c04fd430c8","schema_version":1,"#,
            r#""created_at":{"secs_since_epoch":1735689600,"nanos_since_epoch":0},"#,
            r#""updated_at":{"secs_since_epoch":1735689660,"nanos_since_epoch":0},"#,
            r#""parameters":{"symbol":"SPX","steps":4,"#,
            r#""effective_start":"2026-01-05T14:30:00Z","step_interval_seconds":86400,"#,
            r#""time_frame":"Day","schedule":{"calendar":"weekdays_v1","#,
            r#""timezone":"America/New_York","expiration_time":"17:00:00","rules":["#,
            r#"{"rule_id":"zero_dte","kind":"daily","target_count":1}]},"#,
            r#""tzdb_version":"2025b","initial_price":5000,"volatility":0.18,"#,
            r#""risk_free_rate":"0.04","dividend_yield":0.012,"#,
            r#""method":{"Historical":{"timeframe":"Day","#,
            r#""prices":[5000,5012.5,4987.25,5030],"symbol":"SPX"}},"#,
            r#""chain_size":15,"strike_interval":25,"skew_slope":"-0.2","#,
            r#""smile_curve":"0.4","spread":0.02,"seed":42},"#,
            r#""current_step":0,"total_steps":4,"state":"Initialized","version":0}"#,
        );

        let simulation: SessionV2 = match serde_json::from_str(STORED_BY_AN_OLDER_BINARY) {
            Ok(simulation) => simulation,
            Err(error) => panic!("a stored historical series must load: {error}"),
        };

        match simulation.parameters.method {
            SimulationMethod::Historical {
                ref timeframe,
                ref prices,
                ..
            } => {
                assert_eq!(*timeframe, TimeFrame::Day);
                assert_eq!(prices.len(), 4);
                assert_eq!(prices[0], pos_or_panic!(5000.0));
                assert_eq!(prices[1], pos_or_panic!(5012.5));
                assert_eq!(prices[2], pos_or_panic!(4987.25));
                assert_eq!(prices[3], pos_or_panic!(5030.0));
            }
            ref other => panic!("the stored method must survive the load, got {other:?}"),
        }
        assert_eq!(simulation.parameters.seed, 42);

        let rewritten = match serde_json::to_string(&simulation) {
            Ok(rewritten) => rewritten,
            Err(error) => panic!("must re-serialize: {error}"),
        };
        assert!(
            rewritten.contains(r#""prices":["5000","5012.5","4987.25","5030"]"#),
            "a rewritten series must carry the 0.6 string form, got {rewritten}"
        );
        match serde_json::from_str::<SessionV2>(&rewritten) {
            Ok(reloaded) => assert_eq!(reloaded.parameters.method, simulation.parameters.method),
            Err(error) => panic!("the rewritten simulation must load: {error}"),
        }
    }

    /// Stored-simulation compatibility across the `positive` 0.6 wire change.
    ///
    /// Up to `positive` 0.5 a `Positive` serialised as a JSON number; since 0.6
    /// it serialises as its exact decimal in a string. Simulations written by
    /// an earlier binary sit in Redis in the numeric form and must keep
    /// loading, validation and all — the v2 store reads them back through this
    /// same `Deserialize`. The document below is that numeric form, value for
    /// value as `positive` 0.5.1 emitted it.
    #[test]
    fn test_a_simulation_written_before_positive_0_6_still_loads() {
        const STORED_BY_AN_OLDER_BINARY: &str = concat!(
            r#"{"id":"6ba7b811-9dad-11d1-80b4-00c04fd430c8","schema_version":1,"#,
            r#""created_at":{"secs_since_epoch":1735689600,"nanos_since_epoch":0},"#,
            r#""updated_at":{"secs_since_epoch":1735689660,"nanos_since_epoch":0},"#,
            r#""parameters":{"symbol":"SPX","steps":500,"#,
            r#""effective_start":"2026-01-05T14:30:00Z","step_interval_seconds":86400,"#,
            r#""time_frame":"Day","schedule":{"calendar":"weekdays_v1","#,
            r#""timezone":"America/New_York","expiration_time":"17:00:00","rules":["#,
            r#"{"rule_id":"monthlies","kind":"monthly","target_count":12,"weekday":"Fri"},"#,
            r#"{"rule_id":"weeklies","kind":"weekly","target_count":3,"#,
            r#""weekdays":["Mon","Wed","Fri"]},"#,
            r#"{"rule_id":"zero_dte","kind":"daily","target_count":1}]},"#,
            r#""tzdb_version":"2025b","initial_price":5000,"volatility":0.18,"#,
            r#""risk_free_rate":"0.04","dividend_yield":0.012,"#,
            r#""method":{"GeometricBrownian":{"dt":0.004,"drift":"0.05","volatility":0.18}},"#,
            r#""chain_size":15,"strike_interval":25,"skew_slope":"-0.2","#,
            r#""smile_curve":"0.4","spread":0.02,"seed":42},"#,
            r#""current_step":7,"total_steps":500,"state":"InProgress","version":9}"#,
        );

        let simulation: SessionV2 = match serde_json::from_str(STORED_BY_AN_OLDER_BINARY) {
            Ok(simulation) => simulation,
            Err(error) => panic!("a simulation stored by an older binary must load: {error}"),
        };

        assert_eq!(simulation.parameters.initial_price, pos_or_panic!(5000.0));
        assert_eq!(simulation.parameters.volatility, pos_or_panic!(0.18));
        assert_eq!(simulation.parameters.dividend_yield, pos_or_panic!(0.012));
        assert_eq!(
            simulation.parameters.strike_interval,
            Some(pos_or_panic!(25.0))
        );
        assert_eq!(simulation.parameters.spread, Some(pos_or_panic!(0.02)));
        match simulation.parameters.method {
            SimulationMethod::GeometricBrownian {
                dt,
                drift,
                volatility,
            } => {
                assert_eq!(dt, pos_or_panic!(0.004));
                assert_eq!(drift, dec!(0.05));
                assert_eq!(volatility, pos_or_panic!(0.18));
            }
            ref other => panic!("the stored method must survive the load, got {other:?}"),
        }

        // The seed is the reproducibility contract; a stored simulation that
        // came back with a different one would serve a different tape.
        assert_eq!(simulation.parameters.seed, 42);
        assert_eq!(simulation.current_step, 7);
        assert_eq!(simulation.state, SessionState::InProgress);
        assert_eq!(simulation.version, 9);

        // Rewriting it stores the 0.6 shape, which the same reader accepts, so
        // the store converges on one form with no migration step.
        let rewritten = match serde_json::to_string(&simulation) {
            Ok(rewritten) => rewritten,
            Err(error) => panic!("must re-serialize: {error}"),
        };
        assert!(
            rewritten.contains(r#""initial_price":"5000""#),
            "a rewritten simulation must carry the 0.6 string form, got {rewritten}"
        );
        match serde_json::from_str::<SessionV2>(&rewritten) {
            Ok(reloaded) => assert_eq!(
                reloaded.parameters.initial_price,
                simulation.parameters.initial_price
            ),
            Err(error) => panic!("the rewritten simulation must load: {error}"),
        }
    }
}