autumn_web/

sim.rs

//! Deterministic simulation testing (sim-testing, issue #1797).
//!
//! This module is the public 0.7.0 developer-experience surface for writing
//! **deterministic** simulation tests: a single [`#[sim_test]`](crate::sim_test)
//! attribute gives you a seeded [`Sim`] handle and a paused runtime, so a test
//! runs identically on every machine and every run, and a failure prints a
//! copy-pasteable line that reproduces it exactly.
//!
//! # Quick start
//!
//! ```rust,ignore
//! use autumn_web::sim::Sim;
//! use autumn_web::sim_test;
//!
//! #[sim_test]
//! async fn deterministic(mut sim: Sim) {
//!     // The seed comes from `AUTUMN_SIM_SEED` (hex `0x..` or decimal,
//!     // default 0). Everything derived from `sim` is seed-driven and
//!     // reproducible.
//!     assert_eq!(sim.seed, 0);
//!     let _rng = sim.rng();
//! }
//! ```
//!
//! Reproduce a failing run by copying the replay line printed on panic, e.g.:
//!
//! ```text
//! AUTUMN_SIM_SEED=0x9f3a cargo test -p my-crate deterministic
//! ```
//!
//! # What a `Sim` gives you
//!
//! - **Virtual time.** [`Sim::build`] mounts a [`crate::test::TestApp`] with a
//!   virtual clock. [`Sim::advance`] and [`Sim::advance_to`] step it together
//!   with tokio's paused timer, so `#[job]` backoff and `#[scheduled]` ticks fire
//!   in virtual time. [`Sim::run_to_idle`] drains jobs, due ticks and durable
//!   commit hooks.
//! - **Seeded identity.** [`Sim::rng`] ([`SimRng`]) draws deterministic values.
//!   [`Sim::build`] seeds the app's [`crate::entropy::Entropy`] from the seed,
//!   so framework-minted ids (jobs, request ids, idempotency keys, sessions)
//!   replay too. [`Sim::seeded_entropy`] returns the same source.
//! - **Faults.** [`Chaos`] ([`Sim::chaos`]) injects seed-sampled faults.
//!   [`FaultPlan`] (#1680), attached with
//!   [`crate::test::TestApp::with_fault_plan`], injects authored ones and
//!   records a serializable [`FaultOutcome`]. [`Sim::kill`] and
//!   [`Sim::restart`] model a process crash; `sim::llm` is a seeded LLM stub.
//! - **Assertions and sweeps.** [`always!`](crate::always) and
//!   [`sometimes!`](crate::sometimes) ([`mod@assert`]). Behind the
//!   `sim-testing` feature, `sim::op` generates workloads (`Sim::gen_ops`,
//!   `Sim::run_proptest` with shrinking) and `sim::sweep` runs one scenario
//!   across many seeds (`sweep_proptest`, driven in CI by the `sim-sweep` bin).
//! - **Deadlocks.** With `AUTUMN_SIM_LIVENESS_BUDGET_SECS` set, a `#[sim_test]`
//!   whose tasks all park panics with its replay line instead of hanging. See
//!   [`__with_liveness_budget`] for the limits.
//! - **Phase 2 (issue #2967).** `Sim::net` routes outbound HTTP through a
//!   seeded `SimNet`. [`Sim::interleave`] and [`Sim::spawn`] reorder ready work
//!   from the seed. [`crash_at`] drops an operation at any await.
//!   [`Sim::try_run_to_idle`] reports a drain that does not settle. Framework
//!   code with no clock in scope reads [`crate::time::ambient_now`] and its
//!   siblings, which follow the sim clock.
//! - **Replicas (issue #3067).** [`Sim::mount_replica`] mounts several apps on
//!   one clock, each with its own [`Replica`] clock (offset, drift, steps).
//!   `Sim::db_link` (with `sqlite`) faults one replica's database session
//!   and writes. [`Sim::run_for`] moves time one event at a time. [`runtime`] gates
//!   database work, so a seed replays it. [`mod@trace`] compares two runs of
//!   one seed.
//!
//! The clock and app handles inside a `Sim` are crate-private:
//!
//! ```compile_fail
//! use autumn_web::sim::SimClock;
//! ```
//!
//! ```compile_fail
//! use autumn_web::sim::SimApp;
//! ```
//!
//! Everything here is designed to grow additively (builder-style) without
//! breaking the frozen surface — hence the `#[non_exhaustive]` markers.

// Several thin accessors here could be `const fn`; they stay non-const so the
// frozen public surface can grow a non-const body later without a break.
#![allow(clippy::missing_const_for_fn)]

use std::sync::Arc;

use chrono::{DateTime, LocalResult, NaiveDateTime, Offset, TimeZone, Utc};
use rand::{Rng, SeedableRng};
use rand_chacha::ChaCha8Rng;
use uuid::Uuid;

use crate::entropy::{Entropy, SeededEntropy, uuid_v4_from_bytes, uuid_v7_from_parts};
use crate::time::TickingClock;

// The per-sim SQLite DB lane (W4, issue #1797): a fresh, migrated, in-process
// SQLite substrate the sim builds its app on. Self-contained and additive — W2
// consumes its pool when it mounts the app on `SimApp`. Only meaningful under the
// `sqlite` feature (the sim's DB substrate is SQLite by design).
//
// Exposed as `#[doc(hidden)] pub` — unstable test/sim plumbing, not the stable
// surface — mirroring this module's `__seed_from_env` / `__replay_line` hidden
// hooks. It is `pub` (not `pub(crate)`) purely so the W4 DoD consolidated
// integration test, which is an external crate, can drive the end-to-end drain;
// the stable mount API remains W2's `Sim::build`.
#[cfg(feature = "sqlite")]
#[doc(hidden)]
pub mod substrate;

// The chaos lane (W5, issue #1797): deterministic fault injection wired into
// `Sim::build`. Additive and opt-in — a default (empty) `Chaos` installs
// nothing. See the module docs for the determinism contract.
pub mod chaos;

#[cfg(feature = "mail")]
pub use chaos::MailFault;
pub use chaos::{Chaos, ChaosEvent, ChaosHook};

// The authored fault lane (issue #1680): `FaultPlan`, an ordinal-targeted,
// seed-deterministic fault schedule installed through `TestApp::with_fault_plan`
// (not through `Sim::chaos`), plus the serializable `FaultOutcome` a scenario
// asserts on. Additive and opt-in — a `TestApp` with no plan is untouched. See
// the module docs for the determinism contract and how it differs from `Chaos`.
pub mod fault;

pub use fault::{
    FaultEffect, FaultLedger, FaultOutcome, FaultPlan, FinalState, FiredFault, PlannedFault,
    ReportedError,
};

// The seeded LLM stub (W5.b, item 6, issue #1797): a deterministic fake
// completion client — canned responses + a seeded fault/latency schedule — for
// exercising agent retry/fallback paths under the virtual clock. Standalone and
// additive; it does not route through the `Chaos` builder. See the module docs
// for the determinism contract.
pub mod llm;

pub use llm::{LlmCall, LlmClient, LlmError, LlmRequest, LlmResponse, SeededLlm, SeededLlmBuilder};

// The crash lane (W5.c item 7, issue #1797): a seed-derived crash schedule plus
// the `Sim` kill/restart primitive for durable crash-recovery tests. Additive —
// the schedule is a pure function of the seed and installs nothing at build.
pub mod crash;

// The W6 semantic core (issue #1797): the `always!` / `sometimes!` assertion
// macros and the thread-local non-vacuity registry. Public (documented) module —
// the macros are `#[macro_export]`ed at the crate root (`autumn_web::always` /
// `autumn_web::sometimes`), and their hidden plumbing plus the sweep-facing
// registry API live here.
pub mod assert;

// The simulated network (issue #2967): seeded latency, drops and partitions
// for outbound `http_client::Client` calls, served by in-process hosts.
#[cfg(feature = "http-client")]
pub mod net;

#[cfg(feature = "http-client")]
pub use net::{NetEvent, NetFault, SimNet};

// The shared demo scenario (issue #2967): one `Op` vocabulary for the
// `sim-sweep` proptest sweep and the `sim_ops` cargo-fuzz target. Hidden,
// unstable harness plumbing.
#[doc(hidden)]
pub mod scenario;

// The interleaving shuffler (issue #2967): seeded poll order for
// `Sim::interleave` and seeded yields for `Sim::spawn`.
mod shuffle;

// Several apps on one sim clock (issue #3067).
pub mod replica;

pub use replica::{Replica, ReplicaClock};

// Faults on one replica's link to the shared SQLite database (issue #3067).
#[cfg(feature = "sqlite")]
pub mod dblink;

#[cfg(feature = "sqlite")]
pub use dblink::{DbFaultEvent, DbFaultKind, DbLink};

// The sim runtime and its blocking-work gate (issue #3067).
pub(crate) mod gate;

pub use gate::{SimRuntime, runtime};

// Same-seed trace capture and diff (issue #3067).
pub mod trace;

pub use trace::{Trace, TraceDiff};

// Framework scenarios (jobs, scheduler, lock) for the seed sweep (issue #3067).
#[cfg(feature = "sqlite")]
#[doc(hidden)]
pub mod fleet;

pub use assert::{
    SometimesRegistry, assert_all_sometimes_satisfied, reset_sometimes_registry,
    sometimes_snapshot, sometimes_unsatisfied,
};
pub use crash::{CrashOutcome, CrashPoint, CrashSchedule, crash_at};

// The W6 op-driver (PR2, issue #1797): `Sim::gen_ops`/`Sim::gen_ops_with` (deterministic,
// non-shrinking generation) and `Sim::run_proptest` (the shrink-capable
// runner-owning entrypoint). Behind the `sim-testing` feature because it needs
// `proptest` as a library (not just dev) dependency — see `autumn/Cargo.toml`.
#[cfg(feature = "sim-testing")]
pub mod op;

// The W6 seed-sweep runner (PR3, issue #1797): `sweep_proptest` runs
// `Sim::run_proptest` sequentially across a batch of seeds, reporting the
// first failing seed (if any), and folds `sometimes!` reachability — across
// every proptest case in every seed — across the whole swept range so a
// green sweep is provably non-vacuous. Sequential, not parallel: see
// `sim::sweep`'s module docs for why a `body` that mounts a real app makes
// OS-thread parallelism unsafe here. The `sim-sweep` `[[bin]]`
// (`autumn/src/bin/sim_sweep.rs`) is its CI-facing driver. Same
// `sim-testing` feature gate as `op` — it builds directly on
// `Sim::run_proptest_with_case_hook`.
#[cfg(feature = "sim-testing")]
pub mod sweep;

#[cfg(feature = "sim-testing")]
pub use sweep::{SweepFailure, SweepOutcome, sweep_proptest};

/// The fixed, deterministic epoch the simulation clock starts at:
/// `2020-01-01T00:00:00Z`.
///
/// Every sim run starts its virtual clock here so wall-clock-derived values are
/// reproducible across machines and runs. W2 drives this clock forward via
/// `SimClock`.
const SIM_EPOCH_UNIX_SECS: i64 = 1_577_836_800; // 2020-01-01T00:00:00Z

/// A deterministic simulation handle, constructed from a single `u64` seed.
///
/// `Sim` is the day-one **public, stability-frozen** entry point handed to a
/// [`#[sim_test]`](crate::sim_test) body. Its [`seed`](Sim::seed) is public so a
/// test can assert on or thread it; the injection handles it owns
/// ([`SimRng`], the clock, [`Chaos`] and the mounted app) are private and reached
/// through accessors, so their internals can evolve wave-over-wave without a
/// breaking change.
///
/// Marked `#[non_exhaustive]` so future waves can add handles without breaking
/// construction — always build one via [`Sim::from_seed`].
#[non_exhaustive]
pub struct Sim {
    /// The seed this simulation was constructed from.
    ///
    /// Reproduce a run by exporting `AUTUMN_SIM_SEED=0x<seed>` (the replay line
    /// printed on panic does this for you).
    pub seed: u64,

    /// Seeded deterministic RNG, reached through [`Sim::rng`].
    rng: SimRng,

    /// Virtual clock, started at the fixed sim epoch
    /// (`2020-01-01T00:00:00Z`). Stepped by [`Sim::advance`] in lockstep with
    /// tokio's paused timer.
    clock: SimClock,

    /// Fault-injection configuration installed at [`Sim::build`]. A default
    /// (empty) [`Chaos`] is inactive and installs nothing.
    chaos: Chaos,

    /// Shared chaos runtime state (decision stream + event log), populated by
    /// [`Sim::build`] when [`chaos`](Self::chaos) is active. Read through
    /// [`Sim::__chaos_events`].
    chaos_state: Option<Arc<chaos::ChaosState>>,

    /// Built [`crate::test::TestClient`] handle, mounted by [`Sim::build`] on
    /// the paused runtime with the virtual clock installed.
    app: SimApp,

    /// Real wall-clock budget for the `strict_wall_clock` real-time leak guard,
    /// or `None` (the default) when the guard is off. Set once via
    /// [`strict_wall_clock`](Sim::strict_wall_clock) /
    /// [`strict_wall_clock_budget`](Sim::strict_wall_clock_budget) *before* the
    /// run and only **read** by [`advance`](Sim::advance) /
    /// [`run_to_idle`](Sim::run_to_idle), so it is a plain `Option` with no
    /// interior mutability — that keeps `Sim: Sync` and the `&self`
    /// advance/drain futures `Send` (a `Cell`/`RefCell` field would break both).
    strict_budget: Option<std::time::Duration>,

    /// The simulated network installed at each mount, if any.
    #[cfg(feature = "http-client")]
    net: Option<net::SimNet>,

    /// How many shuffler calls ([`interleave`](Sim::interleave),
    /// [`spawn`](Sim::spawn)) this sim made. Each call draws its own seeded
    /// stream from the seed and this count. Atomic so `&self` stays `Sync`.
    shuffle_calls: std::sync::atomic::AtomicU64,

    /// The clock installed as this thread's ambient clock (issue #2967). See
    /// [`crate::time::ambient_now`].
    ambient: Arc<AmbientSimClock>,

    /// Keeps [`ambient`](Self::ambient) installed while the sim lives.
    _ambient_guard: crate::time::AmbientGuard,

    /// Keeps this sim on its thread's sim stack, which splits tokio time
    /// between the sims that share a runtime.
    stack_guard: SimStackGuard,

    /// How many times [`mount`](Sim::mount) has run. The first mount seeds the
    /// app's entropy from [`seed`](Sim::seed); each restart derives a new seed
    /// from it, so a restarted process does not replay the crashed one's ids.
    mounts: u64,

    /// The apps from [`mount_replica`](Sim::mount_replica), in mount order
    /// (issue #3067).
    replicas: Vec<ReplicaSlot>,

    /// The database link of each replica, made on first use.
    #[cfg(feature = "sqlite")]
    db_links: std::sync::Mutex<std::collections::BTreeMap<String, DbLink>>,

    /// This sim's place in its sim runtime's live count (issue #3067).
    gate_seat: gate::SimSeat,
}

/// The name a replica has as a network host. Two replica names with one key
/// would share one `SimNet` host, so they cannot both be mounted.
fn replica_host_key(name: &str) -> String {
    #[cfg(feature = "http-client")]
    {
        net::canonical_host(name)
    }
    #[cfg(not(feature = "http-client"))]
    {
        name.to_ascii_lowercase()
    }
}

/// One named replica: its spec, its clock, and its app while it is alive.
struct ReplicaSlot {
    name: String,
    clock: replica::NodeClock,
    client: Option<crate::test::TestClient>,
    /// The network this replica is a host of. [`Sim::net`] can set a new
    /// network later, so the kill and the drop clean this one.
    #[cfg(feature = "http-client")]
    net: Option<net::SimNet>,
}

impl Sim {
    /// Construct a simulation from `seed`.
    ///
    /// Infallible and cheap: it seeds the RNG and starts the virtual clock but
    /// does **not** boot a database or an app, so an empty
    /// [`#[sim_test]`](crate::sim_test) runs with zero setup. Mount an app with
    /// [`build`](Sim::build).
    ///
    /// Built inside its paused runtime, the sim's elapsed time starts here.
    /// Built before that runtime, call [`anchor`](Sim::anchor) first thing in
    /// it; `#[sim_test]` does this for you.
    #[must_use]
    pub fn from_seed(seed: u64) -> Self {
        // Each seed run starts with a clean reachability registry, so the sweep
        // can attribute observed/satisfied `sometimes!` labels to exactly one
        // seed before folding them into its cross-seed aggregate (W6, #1797).
        assert::reset_sometimes_registry();
        let clock = SimClock::new(TickingClock::starting_at(sim_epoch()));
        let ambient_clock = Arc::new(AmbientSimClock {
            wall: std::sync::RwLock::new(Arc::new(clock.ticking())),
            own_advanced: std::sync::Mutex::new(std::time::Duration::ZERO),
            auto_advanced: std::sync::Mutex::new(std::time::Duration::ZERO),
            alive: std::sync::atomic::AtomicBool::new(true),
        });
        let gate_seat = gate::SimSeat::default();
        gate_seat.take();
        let ambient_guard = crate::time::install_ambient(ambient_clock.clone());
        let sim_stack = SimStackGuard::enter(Arc::clone(&ambient_clock));
        Self {
            seed,
            rng: SimRng::new(seed),
            clock,
            ambient: ambient_clock,
            _ambient_guard: ambient_guard,
            stack_guard: sim_stack,
            chaos: Chaos::default(),
            chaos_state: None,
            app: SimApp::default(),
            strict_budget: None,
            #[cfg(feature = "http-client")]
            net: None,
            shuffle_calls: std::sync::atomic::AtomicU64::new(0),
            mounts: 0,
            replicas: Vec::new(),
            #[cfg(feature = "sqlite")]
            db_links: std::sync::Mutex::default(),
            gate_seat,
        }
    }

    /// Start this sim's elapsed time at tokio's current instant, if it has not
    /// started yet. Call it first thing in the runtime that drives a sim built
    /// before that runtime: tokio time that passes before the sim is anchored
    /// is not in its elapsed time. Calling it again changes nothing, and
    /// outside a runtime it does nothing.
    ///
    /// A sim built inside its runtime, or run by
    /// [`#[sim_test]`](crate::sim_test), is anchored already.
    pub fn anchor(&self) {
        if tokio::runtime::Handle::try_current().is_ok() {
            lock_sim_stack(&self.stack_guard.home, SimStack::settle);
            // A sim built before `sim::runtime()` joins its gate here.
            self.gate_seat.take();
        }
    }

    /// Configure deterministic fault injection for this simulation.
    ///
    /// The `chaos` builder's hooks (transient DB checkout errors, job duplicate
    /// delivery, clock skew) are installed at [`build`](Sim::build) time, each
    /// fault decision drawn from a dedicated seed-derived stream so the same
    /// seed and configuration replay the same fault schedule. A default
    /// [`Chaos`] is inactive and changes nothing.
    ///
    /// ```rust,ignore
    /// use autumn_web::sim::Chaos;
    /// sim.chaos(Chaos::default().db_transient_errors(0.1).job_duplicate_delivery(0.2));
    /// let client = sim.build(TestApp::new().routes(routes![touch]).jobs(jobs![work]));
    /// ```
    pub fn chaos(&mut self, chaos: Chaos) -> &mut Self {
        self.chaos = chaos;
        self
    }

    /// Route the app's outbound HTTP through a simulated network (issue
    /// #2967).
    ///
    /// [`build`](Sim::build) installs `net` in the app, so the
    /// [`crate::http_client::Client`] extractor sends through it: seeded
    /// latency and drops, partitions, and in-process hosts. Keep a clone to
    /// partition hosts and read [`SimNet::events`] later. See [`mod@net`].
    #[cfg(feature = "http-client")]
    pub fn net(&mut self, net: SimNet) -> &mut Self {
        self.net = Some(net);
        self
    }

    /// The seed this simulation was constructed from.
    ///
    /// Provided for API symmetry alongside the public [`seed`](Sim::seed)
    /// field.
    #[must_use]
    pub const fn seed(&self) -> u64 {
        self.seed
    }

    /// Borrow the seeded deterministic RNG handle.
    ///
    /// Draw deterministic values and UUIDs through it — e.g.
    /// [`SimRng::uuid_v4`] / [`SimRng::next_u64`]. The same seed always yields
    /// the same draw sequence.
    #[must_use]
    pub fn rng(&mut self) -> &mut SimRng {
        &mut self.rng
    }

    /// Build a shared, seeded [`Entropy`] source for this simulation's seed,
    /// ready to inject into a mounted app via
    /// [`crate::state::AppState::with_entropy`].
    ///
    /// [`build`](Self::build) already installs this source in the app it
    /// mounts, unless the test passed its own with
    /// [`crate::test::TestApp::with_entropy`]. Use this to seed state you build
    /// yourself.
    ///
    /// This is the bridge W3 provides for W2's app mounting: the app the
    /// simulation drives resolves the [`crate::entropy::Rng`] extractor and
    /// every framework-minted identifier (job ids, request ids, idempotency
    /// lock owners, session ids) from this source, so a fixed seed replays the
    /// whole identifier stream byte-for-byte.
    ///
    /// The returned source is seeded independently from the [`rng`](Self::rng)
    /// handle (both from [`seed`](Self::seed)), so drawing from one does not
    /// perturb the other's sequence.
    #[must_use]
    pub fn seeded_entropy(&self) -> Arc<dyn Entropy> {
        SeededEntropy::shared(self.seed)
    }

    /// Mount `app` on the paused runtime with the simulation's virtual clock
    /// installed, and return the resulting [`crate::test::TestClient`].
    ///
    /// The simulation's virtual clock is threaded in via
    /// [`crate::test::TestApp::with_clock`], so every handler that reads a
    /// [`crate::time::Clock`] extractor sees the virtual instant — starting at
    /// the fixed sim epoch (`2020-01-01T00:00:00Z`) and moving only when
    /// [`Sim::advance`] steps it. The built app also starts the in-process job
    /// runtime (the in-memory backend), so [`run_to_idle`](Sim::run_to_idle)
    /// can drain enqueued jobs deterministically, and starts its `#[scheduled]`
    /// tasks, whose ticks fire as virtual time crosses their deadlines.
    ///
    /// Unless `app` already has an entropy source from
    /// [`crate::test::TestApp::with_entropy`], `build` installs one seeded from
    /// [`seed`](Sim::seed), so framework-minted ids replay from the seed. A
    /// later mount through [`restart`](Sim::restart) derives a new seed from it,
    /// so a restarted process does not repeat the crashed one's ids.
    ///
    /// Configure `app` fully before handing it over: routes, jobs, tasks and a
    /// sim database are attached to the [`crate::test::TestApp`] before this
    /// call. Do **not** call [`crate::test::TestApp::with_clock`] yourself;
    /// `build` owns the clock so time stays in lockstep.
    ///
    /// The returned borrow is convenient for an immediate request; to interleave
    /// requests with [`advance`](Sim::advance) / [`run_to_idle`](Sim::run_to_idle)
    /// calls, reach the client through [`client`](Sim::client) instead.
    ///
    /// ```rust,ignore
    /// let client = sim.build(TestApp::new().routes(routes![hello]).jobs(jobs![work]));
    /// client.get("/hello").send().await.assert_ok();
    /// ```
    pub fn build(&mut self, app: crate::test::TestApp) -> &crate::test::TestClient {
        self.mount(app)
    }

    /// Mount `app` on the paused runtime with the simulation's virtual clock (and
    /// active chaos hooks) installed, replacing any previously-mounted client.
    ///
    /// Shared by [`build`](Self::build) and [`restart`](Self::restart) so the
    /// initial mount and a post-crash restart go through byte-for-byte the same
    /// path. When chaos is active this re-derives the chaos decision state from
    /// the seed, so a restart's fault schedule replays deterministically.
    fn mount(&mut self, app: crate::test::TestApp) -> &crate::test::TestClient {
        // Seed the app's entropy unless the test injected its own source, so
        // framework-minted ids replay from the seed with no extra call.
        let mount_seed = mount_entropy_seed(self.seed, self.mounts);
        let app = app.with_default_entropy(SeededEntropy::shared(mount_seed));
        self.mounts += 1;
        // Install the simulated network, with a stream fresh for this mount.
        #[cfg(feature = "http-client")]
        let app = match &self.net {
            Some(net) => {
                net.reseed(mount_seed);
                app.with_sim_net(net.clone())
            }
            None => app,
        };
        // When chaos is active, install its deterministic hooks (which also own
        // the clock so a skew wrapper can be applied); otherwise the build is
        // byte-for-byte the pre-W5 path — just the virtual clock.
        // The ambient wall clock matches the app clock, skewed or not.
        *self
            .ambient
            .wall
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner) =
            chaos::wall_clock(&self.chaos, self.seed, self.clock.ticking());
        let app = if self.chaos.is_active() {
            let state = chaos::ChaosState::new(self.seed, &self.chaos);
            self.chaos_state = Some(Arc::clone(&state));
            chaos::install(app, &self.chaos, self.seed, self.clock.ticking(), state)
        } else {
            app.with_clock(self.clock.ticking())
        };
        let client = app.build();
        self.app.client = Some(client);
        self.app.client()
    }

    /// Simulate a process crash: stop the app's job and scheduler tasks and
    /// drop the mounted app, so in-flight work is **cancelled without
    /// completing** (the client's [`Drop`] also cancels the runtime's shutdown
    /// token and clears the global job client), ready for durable recovery on
    /// [`restart`](Self::restart).
    ///
    /// This is the kill half of the W5.c crash-recovery primitive (item 7). It
    /// deliberately drops **only** the app/runtime, never the durable database:
    /// the caller holds the sim's DB substrate (e.g. an
    /// `SqliteSubstrate`) and its `pool()`, so every committed row — crucially
    /// the durable `autumn_repository_commit_hooks` queue — survives the crash
    /// and is still there when a fresh app is mounted on the same pool.
    ///
    /// A crash after [`build`](Self::build) has not run is a no-op.
    ///
    /// # Durability boundary (stated plainly)
    ///
    /// Under the `sqlite` sim substrate the app runs the **in-memory `local`
    /// job backend**, which is **not durable** — a kill drops its mid-flight and
    /// still-queued jobs by design, exactly as a real process crash would drop an
    /// in-memory queue. Item 7's durable guarantee is therefore asserted against
    /// the DB-backed repository commit-hook queue, **not** the local job queue;
    /// the in-memory job queue's by-design loss is documented, never pretended
    /// durable. See the [`crash`] module docs.
    pub fn kill(&mut self) {
        // Stop the app's tasks, as a process death does. Dropping the client
        // then runs `TestJobRuntime::drop` (shutdown.cancel() +
        // clear_global_job_client()).
        if let Some(client) = self.app.client.take() {
            abort_app_tasks(&client);
            drop(client);
        }
        // A fresh process has no in-memory chaos decision log; a restart
        // re-derives it deterministically from the seed.
        self.chaos_state = None;
    }

    /// Restart after a [`kill`](Self::kill): mount a fresh `app` on the paused
    /// runtime, modelling a process restart on the **same durable database**.
    ///
    /// The caller rebuilds the `TestApp` against the *same* substrate pool
    /// (`TestApp::new()…with_db(substrate.pool())`), so the restarted app sees
    /// every row the crashed process committed. Following the restart with
    /// [`run_to_idle`](Self::run_to_idle) drains the durable repository
    /// commit-hook queue, recovering and running any hook the crash left
    /// un-drained (at-least-once / idempotent). Registering the app's hook
    /// runners on the fresh app models a real app re-registering them on boot.
    pub fn restart(&mut self, app: crate::test::TestApp) -> &crate::test::TestClient {
        self.mount(app)
    }

    /// Kill the running app and immediately [`restart`](Self::restart) it on a
    /// fresh `app` — the kill-then-restart convenience over
    /// [`kill`](Self::kill) + [`restart`](Self::restart).
    ///
    /// The `app` must be rebuilt against the same durable substrate pool so the
    /// restarted process recovers the crashed one's committed rows.
    pub fn crash_and_restart(&mut self, app: crate::test::TestApp) -> &crate::test::TestClient {
        self.kill();
        self.restart(app)
    }

    /// Mount `app` as the replica `replica`, next to the other replicas
    /// (issue #3067).
    ///
    /// Each replica is its own app on the sim's one clock: its own state, job
    /// runtime and scheduled tasks. It reads its own [`Replica`] clock. Give
    /// each replica its own pool on one database, so they share data as a
    /// fleet does:
    ///
    /// ```rust,ignore
    /// let substrate = SqliteSubstrate::new()?;
    /// for name in ["a", "b"] {
    ///     let pool = substrate.replica_pool(&sim.db_link(name))?;
    ///     sim.mount_replica(name, TestApp::new().with_db(pool).jobs(jobs![work]));
    /// }
    /// sim.run_for(Duration::from_secs(30)).await;
    /// ```
    ///
    /// With [`net`](Sim::net) set, the replica is also a [`SimNet`] host under
    /// its name, so `http://b/` reaches the replica `b`.
    ///
    /// Process-wide state is not per replica: the cache, the event bus and the
    /// global job client belong to the replica mounted last. Enqueue through a
    /// replica's own `JobClient` extension instead of `job::enqueue`.
    ///
    /// # Panics
    ///
    /// Panics if a replica with this name is already mounted. A name that
    /// is the same host name, such as one that differs only in case, is the
    /// same name. Use
    /// [`restart_replica`](Sim::restart_replica) to mount it again.
    pub fn mount_replica(
        &mut self,
        replica: impl Into<Replica>,
        app: crate::test::TestApp,
    ) -> &crate::test::TestClient {
        let replica = replica.into();
        let key = replica_host_key(replica.name());
        if let Some(slot) = self
            .replicas
            .iter()
            .find(|slot| replica_host_key(&slot.name) == key)
        {
            panic!(
                "replica `{}` is already mounted as `{}`; use `restart_replica` to mount it again",
                replica.name(),
                slot.name
            );
        }
        let clock = replica::NodeClock::new(
            Arc::clone(&self.ambient) as Arc<dyn crate::time::ClockSource>,
            sim_epoch(),
            replica.resolve(self.seed),
        );
        self.replicas.push(ReplicaSlot {
            name: replica.name().to_owned(),
            clock,
            client: None,
            #[cfg(feature = "http-client")]
            net: None,
        });
        let index = self.replicas.len() - 1;
        self.mount_replica_at(index, app)
    }

    /// Kill the replica `name` if it is alive, then mount `app` in its place.
    /// The replica keeps its clock, as a restarted process keeps its host's
    /// clock. Rebuild `app` on the same database to recover its rows.
    ///
    /// # Panics
    ///
    /// Panics if no replica with this name was mounted.
    pub fn restart_replica(
        &mut self,
        name: &str,
        app: crate::test::TestApp,
    ) -> &crate::test::TestClient {
        self.kill_replica(name);
        let index = self.expect_replica_index(name);
        self.mount_replica_at(index, app)
    }

    /// Kill the replica `name`: stop its tasks and drop its app, as a process
    /// crash does. A job handler that was running stops at its next await and
    /// does not settle its claim. The database keeps every committed row. A
    /// kill of a dead replica changes nothing.
    ///
    /// # Panics
    ///
    /// Panics if no replica with this name was mounted.
    pub fn kill_replica(&mut self, name: &str) {
        let index = self.expect_replica_index(name);
        let Some(client) = self.replicas[index].client.take() else {
            return;
        };
        #[cfg(feature = "http-client")]
        if let Some(net) = self.replicas[index].net.take() {
            net.remove_host(name);
        }
        abort_app_tasks(&client);
        drop(client);
    }

    /// The client of the live replica `name`.
    ///
    /// # Panics
    ///
    /// Panics if no live replica has this name.
    #[must_use]
    pub fn replica(&self, name: &str) -> &crate::test::TestClient {
        self.try_replica(name)
            .unwrap_or_else(|| panic!("no live replica `{name}`: mount it with `mount_replica`"))
    }

    /// The client of the live replica `name`, or `None`.
    #[must_use]
    pub fn try_replica(&self, name: &str) -> Option<&crate::test::TestClient> {
        self.replica_index(name)
            .and_then(|index| self.replicas[index].client.as_ref())
    }

    /// The names of the live replicas, in mount order.
    #[must_use]
    pub fn replica_names(&self) -> Vec<&str> {
        self.replicas
            .iter()
            .filter(|slot| slot.client.is_some())
            .map(|slot| slot.name.as_str())
            .collect()
    }

    /// The clock settings of the replica `name` (alive or killed), or `None`.
    #[must_use]
    pub fn replica_clock(&self, name: &str) -> Option<ReplicaClock> {
        self.replica_index(name)
            .map(|index| self.replicas[index].clock.settings())
    }

    /// Jump the wall clock of the replica `name` by `by`, as an NTP step does.
    /// `by` can be negative: a step can go back. The monotonic clock does not
    /// move. Timers the replica already set do not move.
    ///
    /// # Panics
    ///
    /// Panics if no replica with this name was mounted.
    pub fn step_replica_clock(&self, name: &str, by: chrono::TimeDelta) {
        let index = self.expect_replica_index(name);
        self.replicas[index].clock.step(by);
    }

    /// Let `duration` of virtual time pass, one event at a time (issue #3067).
    ///
    /// The sim sleeps. Tokio moves its paused clock to the next timer only
    /// when all tasks wait and no blocking work is queued, such as a `SQLite`
    /// query. So database work ends before time moves on, and a seed replays
    /// the same order. With replicas on `SQLite`, use it, not
    /// [`advance`](Sim::advance).
    ///
    /// Replica clocks move with each step. The clock of the app from
    /// [`build`](Sim::build) moves once, at the end.
    pub async fn run_for(&self, duration: std::time::Duration) {
        let guard_start = self.wall_clock_guard_start();
        let start = tokio::time::Instant::now();
        tokio::time::sleep(duration).await;
        let moved = tokio::time::Instant::now().saturating_duration_since(start);
        // Tokio time already moved, so this is not an explicit advance: the
        // ambient clock counts it as auto-advanced time.
        self.clock.advance(moved);
        self.enforce_wall_clock_budget(guard_start);
    }

    /// The database link of the replica `name`, made on first use and seeded
    /// from the sim seed and the name. Pass it to
    /// [`SqliteSubstrate::replica_pool`](substrate::SqliteSubstrate::replica_pool),
    /// then fault the replica's database through it. See [`DbLink`].
    ///
    /// The link must exist before the mount, so this does not check the name.
    /// A wrong name gives a link that no replica uses.
    #[cfg(feature = "sqlite")]
    #[must_use]
    pub fn db_link(&self, name: &str) -> DbLink {
        self.db_links
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .entry(name.to_owned())
            .or_insert_with(|| {
                DbLink::new(name, derive_seed(self.seed, &format!("sim-db-link:{name}")))
            })
            .clone()
    }

    /// The app from [`build`](Sim::build), then each live replica.
    fn live_clients(&self) -> impl Iterator<Item = &crate::test::TestClient> {
        self.app
            .try_client()
            .into_iter()
            .chain(self.replicas.iter().filter_map(|slot| slot.client.as_ref()))
    }

    fn replica_index(&self, name: &str) -> Option<usize> {
        self.replicas.iter().position(|slot| slot.name == name)
    }

    fn expect_replica_index(&self, name: &str) -> usize {
        self.replica_index(name)
            .unwrap_or_else(|| panic!("no replica `{name}` was mounted"))
    }

    /// Build `app` for the replica slot `index` and store its client.
    fn mount_replica_at(
        &mut self,
        index: usize,
        app: crate::test::TestApp,
    ) -> &crate::test::TestClient {
        let mount_seed = mount_entropy_seed(self.seed, self.mounts);
        let first_mount = self.mounts == 0;
        let app = app.with_default_entropy(SeededEntropy::shared(mount_seed));
        self.mounts += 1;
        // One network for every replica: only the first mount seeds it.
        #[cfg(feature = "http-client")]
        let app = match &self.net {
            Some(net) => {
                if first_mount {
                    net.reseed(mount_seed);
                }
                app.with_sim_net(net.clone())
            }
            None => app,
        };
        #[cfg(not(feature = "http-client"))]
        let _ = first_mount;
        let slot = &mut self.replicas[index];
        #[cfg(feature = "http-client")]
        let app = if self.net.is_some() {
            app.trust_host(&slot.name)
        } else {
            app
        };
        let client = app.with_clock(slot.clock.clone()).build();
        #[cfg(feature = "http-client")]
        if let Some(net) = &self.net {
            net.insert_host(&slot.name, client.router());
            slot.net = Some(net.clone());
        }
        slot.client.insert(client)
    }

    /// Run `ops` concurrently and poll them in a seeded order (issue #2967).
    ///
    /// Each round polls the pending ops in an order drawn from the seed, and
    /// can hold one back for a round. So ops that share state (a lock, a row, a
    /// counter) meet in a different order per seed, and the same seed replays
    /// the same order. Outputs keep the input order.
    ///
    /// ```rust,ignore
    /// let client = sim.client();
    /// let ops = vec![client.post("/a").send(), client.post("/b").send()];
    /// let responses = sim.interleave(ops).await;
    /// ```
    pub async fn interleave<F: std::future::Future>(&self, ops: Vec<F>) -> Vec<F::Output> {
        shuffle::Interleave::new(ops, self.next_shuffle_stream()).await
    }

    /// Spawn `op` on the sim runtime with seeded yields (issue #2967).
    ///
    /// Before a poll, the task can yield from the seeded stream, which moves
    /// it behind the other ready tasks. Tasks spawned this way therefore run
    /// in a seed-driven order. Tasks the framework spawns keep tokio's order.
    pub fn spawn<F>(&self, op: F) -> tokio::task::JoinHandle<F::Output>
    where
        F: std::future::Future + Send + 'static,
        F::Output: Send + 'static,
    {
        tokio::spawn(shuffle::Shuffled::new(op, self.next_shuffle_stream()))
    }

    /// The seeded stream for the next shuffler call.
    fn next_shuffle_stream(&self) -> Arc<dyn Entropy> {
        let call = self
            .shuffle_calls
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        shuffle::stream(self.seed, call)
    }

    /// The seed-derived [`CrashSchedule`] for this simulation.
    ///
    /// A pure function of the [`seed`](Self::seed): two same-seed sims return an
    /// equal schedule. Pass a point's [`CrashPoint::await_index`] to
    /// [`crash_at`] to crash there.
    #[must_use]
    pub fn crash_schedule(&self) -> CrashSchedule {
        CrashSchedule::derive(self.seed, crash::DEFAULT_CRASH_SCHEDULE_LEN)
    }

    /// The first entry of the seed-derived
    /// [`crash_schedule`](Self::crash_schedule).
    ///
    /// `None` only if the schedule is empty (it never is under the default
    /// length). Deterministic for a given seed.
    #[must_use]
    pub fn crash_point(&self) -> Option<CrashPoint> {
        self.crash_schedule().first().cloned()
    }

    /// The recorded chaos fault schedule for this simulation.
    ///
    /// Returns one [`ChaosEvent`] per chaos-hook invocation, in the order the
    /// hooks fired — the reproducible *fault schedule* the run produced. Empty
    /// when chaos was inactive or [`build`](Sim::build) has not run.
    ///
    /// Unstable sim plumbing (hidden from the stable surface, like the module's
    /// other `__`-prefixed hooks); the W5 Definition-of-Done test asserts two
    /// same-seed runs return equal schedules.
    #[doc(hidden)]
    #[must_use]
    pub fn __chaos_events(&self) -> Vec<ChaosEvent> {
        self.chaos_state
            .as_ref()
            .map(|state| state.events())
            .unwrap_or_default()
    }

    /// Borrow the [`crate::test::TestClient`] mounted by [`build`](Sim::build).
    ///
    /// # Panics
    ///
    /// Panics if [`build`](Sim::build) has not been called yet.
    #[must_use]
    pub fn client(&self) -> &crate::test::TestClient {
        self.app.client()
    }

    /// Borrow the mounted [`crate::test::TestClient`], or `None` before
    /// [`build`](Sim::build).
    #[must_use]
    pub const fn try_client(&self) -> Option<&crate::test::TestClient> {
        self.app.try_client()
    }

    /// Enable the **real-time leak guard** (`strict_wall_clock`) with the
    /// default budget, panicking if a paused-sim step burns more than that much
    /// *real* wall-clock time.
    ///
    /// # What this guards (and what it deliberately does not)
    ///
    /// A paused sim runs on tokio's virtual timer: a
    /// [`tokio::time::sleep`](https://docs.rs/tokio) / job backoff / delayed
    /// enqueue consumes **zero** real time and only advances when
    /// [`advance`](Sim::advance) steps the clock. The one thing that breaks that
    /// invariant is code that escapes the virtual timer and blocks the real
    /// thread — a `std::thread::sleep`, a `spawn_blocking`, or a blocking
    /// syscall. This guard catches the **observable consequence** of that: real
    /// wall-clock time leaking into a step that should have taken virtually none.
    ///
    /// It is **not** off-seam-read detection. It cannot tell you that some code
    /// read `Utc::now()` / `Instant::now()` directly instead of through the
    /// injected clock — a free-function `now()` call has no runtime interception
    /// point in safe Rust, so its *absence* is not observable at runtime. Finding
    /// those reads is a static-analysis (deny-lint) job; this guard is the
    /// complementary runtime backstop for the worst pattern (a real blocking
    /// sleep). Enabling it does not slow a legitimate virtual advance: jumping a
    /// day of virtual time still costs microseconds of real time, well under
    /// budget.
    ///
    /// # Budget & the `AUTUMN_SIM_STRICT_WALL_CLOCK_BUDGET_MS` override
    ///
    /// The default budget is deliberately generous (2000 ms) so ordinary CI
    /// scheduling jitter never trips it — the target is a *real* sleep (seconds),
    /// not sub-millisecond noise. Set the environment variable
    /// `AUTUMN_SIM_STRICT_WALL_CLOCK_BUDGET_MS` (whole milliseconds) to override
    /// the budget globally for a run; a blank or unparseable value is ignored and
    /// the default (or the value passed to
    /// [`strict_wall_clock_budget`](Sim::strict_wall_clock_budget)) applies. This
    /// mirrors the `AUTUMN_SIM_SEED` idiom, so a too-tight budget on a slow
    /// runner can be loosened without editing test code.
    ///
    /// The guard is read-only during the run: enable it (chainably) *before* the
    /// first [`advance`](Sim::advance) / [`run_to_idle`](Sim::run_to_idle).
    ///
    /// ```rust,ignore
    /// let mut sim = Sim::from_seed(0);
    /// sim.strict_wall_clock();
    /// sim.build(TestApp::new());
    /// sim.advance(std::time::Duration::from_secs(24 * 3600)).await; // virtual, cheap
    /// sim.run_to_idle().await;
    /// ```
    pub fn strict_wall_clock(&mut self) -> &mut Self {
        self.strict_budget = Some(strict_budget_from_env_or(DEFAULT_STRICT_WALL_CLOCK_BUDGET));
        self
    }

    /// Enable the real-time leak guard with an explicit `budget`.
    ///
    /// Identical to [`strict_wall_clock`](Sim::strict_wall_clock) but starts from
    /// `budget` instead of the 100 ms default. The
    /// `AUTUMN_SIM_STRICT_WALL_CLOCK_BUDGET_MS` environment variable still
    /// overrides `budget` when it holds a valid whole-millisecond value (so CI
    /// can loosen the guard globally); a blank/unparseable value leaves `budget`
    /// in effect. See [`strict_wall_clock`](Sim::strict_wall_clock) for what the
    /// guard does and does not catch.
    ///
    /// ```rust,ignore
    /// let mut sim = Sim::from_seed(0);
    /// sim.strict_wall_clock_budget(std::time::Duration::from_millis(5));
    /// ```
    pub fn strict_wall_clock_budget(&mut self, budget: std::time::Duration) -> &mut Self {
        self.strict_budget = Some(strict_budget_from_env_or(budget));
        self
    }

    /// Sample a real [`std::time::Instant`] at the start of a guarded step, or
    /// `None` when the leak guard is off. Held across the step's `.await`s (an
    /// `Instant` is `Send + Sync`, so the guarded `&self` futures stay `Send`).
    fn wall_clock_guard_start(&self) -> Option<std::time::Instant> {
        self.strict_budget.map(|_| std::time::Instant::now())
    }

    /// Enforce the real-time leak guard at the end of a step: panic if more than
    /// the configured budget of *real* wall-clock time elapsed since `start`.
    ///
    /// A no-op when the guard is off (`start` / `strict_budget` are `None`). The
    /// panic flows up through the [`#[sim_test]`](crate::sim_test) macro's
    /// `catch_unwind`, so the `AUTUMN_SIM_SEED=…` replay line still prints.
    ///
    /// # Panics
    ///
    /// Panics when the leak guard is enabled and real elapsed time exceeds the
    /// budget (see [`strict_wall_clock`](Sim::strict_wall_clock)).
    fn enforce_wall_clock_budget(&self, start: Option<std::time::Instant>) {
        if let (Some(budget), Some(start)) = (self.strict_budget, start) {
            let elapsed = start.elapsed();
            assert!(
                elapsed <= budget,
                "Sim::strict_wall_clock real-time leak guard tripped: {elapsed:?} of real \
                 wall-clock time elapsed inside a paused-sim step, exceeding the {budget:?} \
                 budget. This means real time leaked into the virtual timeline — usually a real \
                 `std::thread::sleep`, blocking I/O, or `spawn_blocking` that escaped tokio's \
                 paused timer. If this is CI scheduling jitter rather than a genuine leak, raise \
                 the budget via the AUTUMN_SIM_STRICT_WALL_CLOCK_BUDGET_MS environment variable."
            );
        }
    }

    /// Advance virtual time by `duration`, stepping the injected wall clock and
    /// tokio's paused timer wheel **together**.
    ///
    /// The framework [`crate::time::Clock`] extractor (backed by the
    /// simulation's virtual clock) and tokio's virtual timer (`tokio::time::sleep`,
    /// job backoff delays, delayed enqueues) move by exactly the same amount, so
    /// `Utc::now()`-via-extractor and a sleeping task stay in lockstep — a job
    /// whose retry backs off 24 hours fires the instant this advances 24 hours,
    /// with zero wall-clock delay. Timers that come due within the window fire
    /// and their tasks are polled before this returns.
    ///
    /// Pair with [`run_to_idle`](Sim::run_to_idle) to then drain the work the
    /// fired timers enqueued.
    pub async fn advance(&self, duration: std::time::Duration) {
        // Real-time leak guard (no-op unless `strict_wall_clock` is enabled):
        // sample a REAL instant (not tokio's paused time) at entry and check the
        // real elapsed against the budget before returning. `advance_to` routes
        // through here, so it inherits the guard for free.
        let guard_start = self.wall_clock_guard_start();
        // `advance` yields and never parks, so database work runs without the
        // gate's turns, as before the gate (#3067).
        let _open = gate::Open::new();
        // Let every ready task take one step at the current instant before time
        // moves. A task spawned since the last yield (a `#[scheduled]` loop that
        // `build` started, a job worker) then registers its first timer at the
        // instant it was started, not at the end of this advance.
        tokio::task::yield_now().await;
        // Step the framework clock first so any task woken by the tokio timer
        // that reads the clock observes the already-advanced instant.
        self.clock.advance(duration);
        // Advance tokio's paused timer wheel; this fires due timers and yields
        // so their tasks are polled before returning.
        // `tokio::time::advance` moves the clock in its first poll, the same
        // poll as this note, and only then yields.
        note_own_advance(&self.ambient, duration);
        tokio::time::advance(duration).await;
        self.enforce_wall_clock_budget(guard_start);
    }

    /// Advance virtual time **to** a specific zoned instant, resolving the
    /// timezone (including any DST transition) to the correct UTC instant and
    /// then stepping forward by the delta from the current sim instant.
    ///
    /// This is the timezone/DST-aware companion to [`advance`](Sim::advance):
    /// where `advance` takes a raw [`std::time::Duration`], `advance_to` takes a
    /// *wall-clock target in any timezone* and computes the real (UTC) delta for
    /// you. It is generic over any [`chrono::TimeZone`] — pass a
    /// `chrono::DateTime<Utc>`, a `chrono::DateTime<chrono::FixedOffset>`, or a
    /// `chrono::DateTime<chrono_tz::Tz>` from the `chrono-tz` crate — so a caller
    /// can express a zoned target without this crate hard-depending on any
    /// particular timezone database.
    ///
    /// The target is converted to UTC via [`chrono::DateTime::with_timezone`],
    /// which is unambiguous (every zoned `DateTime` already names a single
    /// instant), and the sim then reuses [`advance`](Sim::advance) internally so
    /// the injected wall clock and tokio's paused timer wheel stay in **exact
    /// lockstep** — any `tokio::time::sleep` / job-backoff timer whose deadline
    /// falls inside the crossed wall-clock window (a DST "spring-forward" gap
    /// included) fires during the advance, then its task is polled before this
    /// returns. Because the injected clock is UTC and monotonic, a spring-forward
    /// boundary is just a shorter real interval — the timers inside it still fire
    /// correctly.
    ///
    /// # Forward-only semantics
    ///
    /// Virtual time never moves backward:
    ///
    /// - **Target equals the current sim instant** → this is a **no-op** (no
    ///   clock/timer step at all).
    /// - **Target is strictly before the current sim instant** → this
    ///   **panics** with a clear message. Silently doing nothing would hide a
    ///   test bug (a target computed to be in the past almost always means the
    ///   test's arithmetic is wrong), so the panic is deliberate.
    ///
    /// Pair with [`run_to_idle`](Sim::run_to_idle) afterward to drain the work
    /// the fired timers enqueued.
    ///
    /// # Panics
    ///
    /// Panics if `target` resolves to a UTC instant strictly before the current
    /// sim instant (see *Forward-only semantics*).
    ///
    /// ```rust,ignore
    /// use chrono::{TimeZone, Utc};
    /// // Advance the sim clock to a specific UTC instant.
    /// let target = Utc.with_ymd_and_hms(2020, 3, 8, 12, 0, 0).unwrap();
    /// sim.advance_to(&target).await;
    /// sim.run_to_idle().await;
    /// ```
    ///
    /// Written as a non-`async fn` returning a future so the generic zoned
    /// `target` is resolved to UTC **synchronously** and never captured across an
    /// `.await` — the returned future holds only `&self` and the resolved
    /// `DateTime<Utc>`, so it stays `Send` for any `Tz` (a borrowed or non-`Sync`
    /// `Tz` would otherwise poison the future).
    pub fn advance_to<Tz>(
        &self,
        target: &DateTime<Tz>,
    ) -> impl std::future::Future<Output = ()> + '_
    where
        Tz: TimeZone,
    {
        let target_utc = target.with_timezone(&Utc);
        self.advance_to_utc(target_utc)
    }

    /// Advance virtual time to a **naive local wall-clock time** interpreted in
    /// timezone `tz`, resolving DST edge cases explicitly (no `.unwrap()` on a
    /// [`chrono::LocalResult`]).
    ///
    /// Convenience wrapper over [`advance_to`](Sim::advance_to) for the common
    /// "advance to 2:30 AM local on this date" shape, where the naive local time
    /// must be mapped to a single UTC instant. `tz` is any
    /// [`chrono::TimeZone`] (e.g. a `chrono_tz::Tz`); the forward-only /
    /// panic-on-past semantics of [`advance_to`](Sim::advance_to) apply once the
    /// instant is resolved.
    ///
    /// # DST resolution (deterministic)
    ///
    /// A naive local time need not correspond to exactly one UTC instant:
    ///
    /// - **Unambiguous** ([`LocalResult::Single`])
    ///   → that instant.
    /// - **Fall-back / ambiguous** ([`LocalResult::Ambiguous`],
    ///   the wall time occurs twice as the clock rolls back) → the **earlier**
    ///   of the two UTC instants.
    /// - **Spring-forward gap** ([`LocalResult::None`],
    ///   the wall time never occurs because the clock jumps forward) → the
    ///   nonexistent wall time is carried **forward across** the gap to a single
    ///   deterministic post-transition instant (it is resolved by looking up the
    ///   zone's offset at the matching UTC-clock reading and applying it, which
    ///   shifts the requested time past the boundary rather than erroring). For
    ///   example a request for the nonexistent `02:30` on a US spring-forward
    ///   day resolves to `03:30` local (the same instant, one gap-length later).
    ///
    /// # Panics
    ///
    /// Panics if the resolved instant is strictly before the current sim instant
    /// (see [`advance_to`](Sim::advance_to)).
    ///
    /// ```rust,ignore
    /// use chrono::NaiveDate;
    /// let local = NaiveDate::from_ymd_opt(2020, 3, 8).unwrap()
    ///     .and_hms_opt(3, 30, 0).unwrap();
    /// sim.advance_to_local(local, &chrono_tz::America::New_York).await;
    /// ```
    ///
    /// Like [`advance_to`](Sim::advance_to), this is a non-`async fn` returning a
    /// future: the `&tz` reference is used only while resolving the instant
    /// synchronously and is **not** captured by the returned future, so the
    /// future stays `Send` even though `Tz` need not be `Sync`.
    pub fn advance_to_local<Tz>(
        &self,
        local: NaiveDateTime,
        tz: &Tz,
    ) -> impl std::future::Future<Output = ()> + '_
    where
        Tz: TimeZone,
    {
        self.advance_to_utc(resolve_local_to_utc(local, tz))
    }

    /// Shared UTC-target advance used by [`advance_to`](Sim::advance_to) and
    /// [`advance_to_local`](Sim::advance_to_local): plan the step against the
    /// current sim instant, then reuse [`advance`](Sim::advance) so the clock and
    /// timer wheel stay in lockstep.
    async fn advance_to_utc(&self, target: DateTime<Utc>) {
        match plan_advance_to(self.clock.now(), target) {
            AdvancePlan::NoOp => {}
            AdvancePlan::Advance(delta) => self.advance(delta).await,
        }
    }

    /// Drain all ready work — enqueued jobs the in-process runtime can run now,
    /// plus tasks woken by timers that have already come due — until the runtime
    /// is quiescent.
    ///
    /// The sim runtime is a single-threaded, current-thread runtime with the
    /// clock paused, so background tasks (the job worker consuming its queue, a
    /// retry timer that just fired, a delayed enqueue delivering) make progress
    /// only when the running task yields. This cooperatively yields for
    /// `MAX_DRAIN_STEPS` rounds.
    ///
    /// It does **not** fast-forward to a *future* timer — advancing the clock to
    /// reach a not-yet-due backoff/sleep is [`advance`](Sim::advance)'s job
    /// (tokio exposes no next-deadline hook, and a blind auto-advance would
    /// break the clock lockstep). The idiom is therefore
    /// [`advance`](Sim::advance) to the next interesting instant, then
    /// `run_to_idle` to settle the work it released.
    ///
    /// # Panics
    ///
    /// Panics with the seed when the drain does not settle: work still ran in
    /// its last rounds (issue #2967). A job that enqueues itself again is the
    /// usual cause. Use [`try_run_to_idle`](Sim::try_run_to_idle) to get the
    /// [`SimStall`] instead.
    pub async fn run_to_idle(&self) {
        if let Err(stall) = self.try_run_to_idle().await {
            panic!("{stall}");
        }
    }

    /// Like [`run_to_idle`](Sim::run_to_idle), but returns the stall.
    ///
    /// # Errors
    ///
    /// Returns [`SimStall`] when the drain never sees 64 quiet rounds in a row
    /// within 2048 rounds. A round is quiet when no job starts, no scheduled
    /// tick fires, no commit hook or outbox message drains and the task count
    /// does not change.
    /// A single long-running job, or a task that loops without spawning,
    /// does not count as work.
    pub async fn try_run_to_idle(&self) -> Result<(), SimStall> {
        // Real-time leak guard (no-op unless `strict_wall_clock` is enabled):
        // sample a REAL instant at entry and enforce the budget before
        // returning, catching a real blocking sleep in a drained task.
        let guard_start = self.wall_clock_guard_start();
        // The drain yields and never parks, so database work runs without the
        // gate's turns, as before the gate (#3067).
        let _open = gate::Open::new();

        // Resolve the mounted app's DB pool once (a cheap, cloned Arc-backed
        // handle). Durable repository commit hooks are rows, so there is
        // nothing to drain when the app was built without a database (e.g. the
        // in-memory job DoD path) — the lane is skipped entirely then.
        #[cfg(feature = "db")]
        let commit_hook_pools: Vec<_> = self
            .live_clients()
            .filter_map(|client| crate::db::DbState::pool(client.state()).cloned())
            .collect();

        // Run the full `MAX_DRAIN_STEPS`. Then stop at the first
        // `STALL_WINDOW` quiet rounds in a row; work that ends late still
        // settles. Work that never goes quiet is a stall.
        let mut quiet_rounds = 0;
        let mut rounds = 0;
        while rounds < MAX_DRAIN_STEPS
            || (quiet_rounds < STALL_WINDOW && rounds < MAX_DRAIN_STEPS * 2)
        {
            rounds += 1;
            let before = drain_fingerprint();
            // One yield lets each currently-ready spawned task take a step; a
            // zero-duration timer advance flushes any timers registered for the
            // current instant and yields again, so a chain of ready timer/task
            // wakeups settles without advancing the clock. This quiesces the
            // in-process job runtime (TestJobRuntime / JobAdminMemoryBackend)
            // and any tasks woken by already-due scheduler ticks.
            tokio::task::yield_now().await;
            tokio::time::advance(std::time::Duration::ZERO).await;

            // Third ready-work source: durable repository commit hooks. Drain
            // the ready set through the public test-harness wrapper
            // (`crate::test::drain_ready_repository_commit_hooks`), which runs
            // the same claim → run → ack wiring the background commit-hook
            // worker uses, but deterministically and worker-free. A hook may
            // itself enqueue a job, so draining inside the settle loop lets a
            // subsequent iteration pick that job up.
            #[cfg(feature = "db")]
            for pool in &commit_hook_pools {
                let hooks_drained =
                    crate::test::drain_ready_repository_commit_hooks(pool, MAX_DRAIN_STEPS).await;
                if hooks_drained > 0 {
                    note_drain_progress();
                }
            }

            // Fourth source: the transactional outbox (issue #3062). The same
            // claim → handle → mark path the relay worker runs, worker-free.
            #[cfg(feature = "db")]
            for state in self
                .live_clients()
                .map(crate::test::TestClient::state)
                .filter(|state| state.extension::<crate::outbox::OutboxRelay>().is_some())
            {
                match crate::outbox::drain(state, MAX_DRAIN_STEPS).await {
                    Ok(0) => {}
                    Ok(_) => note_drain_progress(),
                    Err(error) => tracing::warn!(%error, "sim outbox drain failed"),
                }
            }

            if drain_fingerprint() == before {
                quiet_rounds += 1;
            } else {
                quiet_rounds = 0;
            }
        }

        self.enforce_wall_clock_budget(guard_start);
        if quiet_rounds < STALL_WINDOW {
            return Err(SimStall {
                seed: self.seed,
                steps: rounds,
            });
        }
        Ok(())
    }
}

impl Drop for Sim {
    fn drop(&mut self) {
        // A replica is a host of the sim network, and its app holds the
        // network: a cycle. Remove the hosts, so the replica apps drop.
        #[cfg(feature = "http-client")]
        for slot in &mut self.replicas {
            if let Some(net) = slot.net.take() {
                net.remove_host(&slot.name);
            }
        }
        // The runtime drops the app tasks after the sim. A task with a query
        // in flight then waits for it, so let gated queries run (#3067).
        self.gate_seat.leave();
    }
}

/// A [`Sim::run_to_idle`] drain that did not settle (issue #2967).
///
/// Work still ran in the last rounds of the drain, so the app was not idle
/// when the drain gave up.
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SimStall {
    /// The seed of the stalled run.
    pub seed: u64,
    /// The drain rounds that ran.
    pub steps: usize,
}

impl std::fmt::Display for SimStall {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "sim drain stall: work still ran after {steps} drain rounds (seed=0x{seed:x}). \
             A job, hook or task keeps making new work, so the app is never idle. \
             Replay with AUTUMN_SIM_SEED=0x{seed:x}.",
            steps = self.steps,
            seed = self.seed,
        )
    }
}

impl std::error::Error for SimStall {}

thread_local! {
    /// Work the drain can see on this thread: jobs run, ticks fired, hooks
    /// drained. [`Sim::try_run_to_idle`] compares it across a round.
    static DRAIN_PROGRESS: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
}

/// Record one unit of drain-visible work on this thread. The job runner, the
/// task scheduler and the commit-hook drain call it.
pub(crate) fn note_drain_progress() {
    DRAIN_PROGRESS.with(|progress| progress.set(progress.get().wrapping_add(1)));
}

/// This thread's drain progress plus the runtime's live task count. A change
/// across a drain round means work ran in it.
fn drain_fingerprint() -> (u64, usize) {
    let tasks = tokio::runtime::Handle::try_current()
        .map_or(0, |handle| handle.metrics().num_alive_tasks());
    (DRAIN_PROGRESS.with(std::cell::Cell::get), tasks)
}

/// The fixed sim epoch as a `DateTime`.
fn sim_epoch() -> DateTime<Utc> {
    Utc.timestamp_opt(SIM_EPOCH_UNIX_SECS, 0)
        .single()
        .unwrap_or_else(|| Utc.timestamp_nanos(0))
}

/// A `u64` derived from `seed` and `purpose`.
#[cfg(feature = "sqlite")]
fn derive_seed(seed: u64, purpose: &str) -> u64 {
    SeededEntropy::new(seed)
        .derive_uuid(purpose)
        .as_u64_pair()
        .0
}

/// The background tasks one app spawned (issue #3067).
///
/// `TestApp::build` puts one in the app state. The job and scheduler loops
/// register here, so a sim kill can stop them as a process death does. A
/// production app has none, and its tasks run as before.
#[derive(Default)]
pub(crate) struct AppTasks {
    handles: std::sync::Mutex<Vec<tokio::task::AbortHandle>>,
}

impl AppTasks {
    fn track(&self, handle: tokio::task::AbortHandle) {
        let mut handles = self
            .handles
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        handles.retain(|handle| !handle.is_finished());
        handles.push(handle);
    }

    fn abort_all(&self) {
        let handles = std::mem::take(
            &mut *self
                .handles
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner),
        );
        for handle in handles {
            handle.abort();
        }
    }
}

/// Spawn an app background task. In a test app, a sim kill stops it.
pub(crate) fn spawn_app_task<F>(
    state: &crate::state::AppState,
    future: F,
) -> tokio::task::JoinHandle<F::Output>
where
    F: std::future::Future + Send + 'static,
    F::Output: Send + 'static,
{
    let handle = tokio::spawn(future);
    if let Some(tasks) = state.extension::<AppTasks>() {
        tasks.track(handle.abort_handle());
    }
    handle
}

/// Stop every background task of the app behind `client`.
fn abort_app_tasks(client: &crate::test::TestClient) {
    if let Some(tasks) = client.state().extension::<AppTasks>() {
        tasks.abort_all();
    }
}

/// The entropy seed for the `mount`-th app a simulation mounts.
///
/// Mount 0 uses `seed` itself, so the default equals
/// `with_entropy(SeededEntropy::new(sim.seed))`. A later mount (a restart after
/// [`Sim::kill`]) derives its seed from `seed` and the mount number: a real
/// restarted process draws new ids, and this keeps them seed-driven.
fn mount_entropy_seed(seed: u64, mount: u64) -> u64 {
    if mount == 0 {
        return seed;
    }
    let derived = SeededEntropy::new(seed).derive_uuid(format!("sim-mount-{mount}"));
    let bytes: [u8; 8] = derived.as_bytes()[..8]
        .try_into()
        .expect("a uuid has 16 bytes");
    u64::from_le_bytes(bytes)
}

/// The longest timer tokio accepts (about 2.2 years). A larger liveness
/// budget is clamped to it.
const MAX_LIVENESS_BUDGET: std::time::Duration = std::time::Duration::from_millis(68_719_476_734);

/// Resolve the liveness budget from the raw `AUTUMN_SIM_LIVENESS_BUDGET_SECS`
/// value. A positive whole number of seconds arms the watchdog. Unset, blank,
/// `0` or unparseable leaves it off.
fn parse_liveness_budget(raw: Option<&str>) -> Option<std::time::Duration> {
    let secs = raw.map(str::trim)?.parse::<u64>().ok()?;
    if secs == 0 {
        return None;
    }
    Some(std::time::Duration::from_secs(secs).min(MAX_LIVENESS_BUDGET))
}

/// Run a `#[sim_test]` body under the liveness watchdog, when
/// `AUTUMN_SIM_LIVENESS_BUDGET_SECS` arms it.
///
/// Hidden macro plumbing, like [`__seed_from_env`]. See
/// [`__with_liveness_budget`] for what the watchdog detects.
#[doc(hidden)]
pub async fn __with_liveness_watchdog<F: std::future::Future>(seed: u64, body: F) -> F::Output {
    // `#[sim_test]` builds the sim before its runtime. Start its elapsed
    // time as the runtime starts, before the body runs.
    anchor_current_sim();
    let raw = std::env::var("AUTUMN_SIM_LIVENESS_BUDGET_SECS").ok();
    __with_liveness_budget(seed, parse_liveness_budget(raw.as_deref()), body).await
}

/// Run `body`, and panic with a message that names `seed` when it is still
/// running after `budget` of virtual time. `None` runs `body` with no watchdog.
///
/// A deadlock parks every task with no timer that could wake one, so without a
/// watchdog the test hangs. The watchdog is itself a timer, so the paused
/// runtime advances straight to it and the test fails at once; the
/// `#[sim_test]` macro then prints the replay line.
///
/// Two limits. A busy loop that never parks keeps the runtime from advancing,
/// so it is not detected. And the paused runtime also advances while a task
/// waits on something outside it (a lock another thread holds, real I/O), so
/// arm the watchdog only where sim tests run one at a time
/// (`--test-threads=1`) and do no real I/O.
#[doc(hidden)]
pub async fn __with_liveness_budget<F: std::future::Future>(
    seed: u64,
    budget: Option<std::time::Duration>,
    body: F,
) -> F::Output {
    let Some(budget) = budget else {
        return body.await;
    };
    tokio::time::timeout(budget, body)
        .await
        .unwrap_or_else(|_elapsed| {
            panic!(
                "sim liveness: the test body did not finish within {budget:?} of virtual time \
                 (seed=0x{seed:x}). Every task was parked with no timer to wake one, which is \
                 a deadlock, or the body waited past the budget. Raise \
                 AUTUMN_SIM_LIVENESS_BUDGET_SECS for a legitimately long run."
            )
        })
}

/// Upper bound on cooperative yield rounds [`Sim::run_to_idle`] performs before
/// returning, so a misbehaving always-ready task can never hang the drain.
/// Generous relative to the handful of hops a job takes from the queue through
/// its handler to completion under the single-threaded paused runtime.
const MAX_DRAIN_STEPS: usize = 1024;

/// Quiet rounds in a row that settle a drain after `MAX_DRAIN_STEPS`. A drain
/// that does not reach them by twice `MAX_DRAIN_STEPS` is a [`SimStall`].
const STALL_WINDOW: usize = 64;

/// Default real wall-clock budget for the `strict_wall_clock` leak guard
/// ([`Sim::strict_wall_clock`]).
///
/// Deliberately generous (2000 ms): the guard exists to catch a *real* blocking
/// sleep escaping the virtual timer (seconds of wall time), so the budget must
/// sit far above ordinary current-thread scheduling jitter to avoid false
/// positives on a slow/contended CI runner. A legitimate virtual advance — even
/// jumping a day of virtual time — costs microseconds of real time, orders of
/// magnitude under this.
const DEFAULT_STRICT_WALL_CLOCK_BUDGET: std::time::Duration = std::time::Duration::from_secs(2);

/// Resolve the effective `strict_wall_clock` budget: the
/// `AUTUMN_SIM_STRICT_WALL_CLOCK_BUDGET_MS` environment override when it holds a
/// valid whole-millisecond value, otherwise `default`.
///
/// Mirrors the [`__seed_from_env`] / [`parse_seed`] idiom so a too-tight budget
/// on a slow runner can be loosened globally without editing test code; a blank
/// or unparseable value falls back to `default`.
fn strict_budget_from_env_or(default: std::time::Duration) -> std::time::Duration {
    std::env::var("AUTUMN_SIM_STRICT_WALL_CLOCK_BUDGET_MS")
        .ok()
        .and_then(|raw| parse_strict_budget_ms(&raw))
        .unwrap_or(default)
}

/// Parse a whole-millisecond budget string into a [`std::time::Duration`],
/// returning `None` for empty/whitespace-only or unparseable input (so the
/// caller can fall back to the configured default).
fn parse_strict_budget_ms(raw: &str) -> Option<std::time::Duration> {
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        return None;
    }
    trimmed
        .parse::<u64>()
        .ok()
        .map(std::time::Duration::from_millis)
}

/// The forward-only step [`Sim::advance_to`] resolves a target instant into.
///
/// Kept as a small pure enum (rather than inlining the branch) so the
/// forward-only / no-op / panic-on-past decision is unit-testable without a
/// runtime or a paused clock.
#[derive(Debug, PartialEq, Eq)]
enum AdvancePlan {
    /// Target equals the current instant — advancing does nothing.
    NoOp,
    /// Target is in the future — step forward by exactly this real delta.
    Advance(std::time::Duration),
}

/// Decide how to advance from `now` to `target` under the forward-only clock
/// contract of [`Sim::advance_to`].
///
/// Returns [`AdvancePlan::NoOp`] when `target == now` and
/// [`AdvancePlan::Advance`] with the positive delta when `target` is in the
/// future.
///
/// # Panics
///
/// Panics when `target` is strictly before `now`: virtual time is forward-only,
/// and a past target signals a test-arithmetic bug that silent no-op behavior
/// would hide.
fn plan_advance_to(now: DateTime<Utc>, target: DateTime<Utc>) -> AdvancePlan {
    let delta = target - now;
    match delta.cmp(&chrono::Duration::zero()) {
        std::cmp::Ordering::Equal => AdvancePlan::NoOp,
        std::cmp::Ordering::Less => panic!(
            "Sim::advance_to target {target} is strictly before the current sim instant {now}; \
             virtual time is forward-only (advancing to a past instant is a test bug)"
        ),
        std::cmp::Ordering::Greater => AdvancePlan::Advance(
            delta
                .to_std()
                .expect("a strictly-positive chrono delta always converts to std::time::Duration"),
        ),
    }
}

/// Resolve a naive local wall-clock time in timezone `tz` to a single UTC
/// instant, handling DST edges deterministically (see
/// [`Sim::advance_to_local`] for the documented policy).
///
/// Ambiguous (fall-back) local times resolve to the **earlier** instant; a
/// nonexistent (spring-forward gap) local time is carried across the gap using
/// the post-transition UTC offset. Never `.unwrap()`s a
/// [`chrono::LocalResult`].
fn resolve_local_to_utc<Tz>(local: NaiveDateTime, tz: &Tz) -> DateTime<Utc>
where
    Tz: TimeZone,
{
    match tz.from_local_datetime(&local) {
        LocalResult::Single(dt) => dt.with_timezone(&Utc),
        LocalResult::Ambiguous(earlier, _later) => earlier.with_timezone(&Utc),
        LocalResult::None => {
            // Spring-forward gap: the wall time never occurs. Resolve it by
            // looking up the zone's offset at the UTC-clock reading numerically
            // equal to the wall time and applying it — a total, deterministic
            // mapping that carries the nonexistent time forward across the gap to
            // a single post-transition instant.
            let offset_secs =
                i64::from(tz.offset_from_utc_datetime(&local).fix().local_minus_utc());
            (local - chrono::Duration::seconds(offset_secs)).and_utc()
        }
    }
}

/// A seeded, deterministic random number generator handle.
///
/// Wraps a `ChaCha8Rng` seeded from the simulation seed, so the same seed
/// always yields the same draw sequence. Draw deterministic bytes and UUIDs
/// through the generation helpers below; they share their `Uuid` bit-stamping
/// with the [`crate::entropy::Entropy`] source an app is seeded with, so a
/// `SimRng` draw and an equivalently-seeded app draw agree.
pub struct SimRng {
    seed: u64,
    inner: ChaCha8Rng,
}

impl SimRng {
    /// Seed a fresh deterministic RNG from `seed`.
    pub(crate) fn new(seed: u64) -> Self {
        Self {
            seed,
            inner: ChaCha8Rng::seed_from_u64(seed),
        }
    }

    /// Derive a stable [`Uuid`] from this simulation's seed and a `purpose_tag`
    /// namespace, **independently of the draw stream** (seed-derived ids).
    ///
    /// Unlike [`uuid_v4`](Self::uuid_v4), this does **not** advance the RNG: the
    /// same seed and `purpose_tag` always produce the same UUID no matter how
    /// many other values have been drawn, so `derive_uuid("tenant:acme")` is a
    /// stable, byte-reproducible id for "acme" across runs and machines. Ideal
    /// for seeding multi-tenant fixtures without perturbing the deterministic id
    /// stream. See [`crate::entropy::SeededEntropy::derive_uuid`] for the shared
    /// mechanism and the version bits it sets (v4).
    #[must_use]
    pub fn derive_uuid(&self, purpose_tag: impl AsRef<[u8]>) -> Uuid {
        crate::entropy::derive_uuid_from(self.seed, purpose_tag.as_ref())
    }

    /// Draw the next deterministic `u64`.
    pub fn next_u64(&mut self) -> u64 {
        self.inner.next_u64()
    }

    /// Fill `dest` with deterministic bytes.
    pub fn fill_bytes(&mut self, dest: &mut [u8]) {
        self.inner.fill_bytes(dest);
    }

    /// Draw a deterministic version-4 (fully random) [`Uuid`].
    ///
    /// The same seed and the same number of prior draws always yield the same
    /// UUID.
    #[must_use]
    pub fn uuid_v4(&mut self) -> Uuid {
        let mut bytes = [0u8; 16];
        self.inner.fill_bytes(&mut bytes);
        uuid_v4_from_bytes(bytes)
    }

    /// Draw a version-7 (time-ordered) [`Uuid`] whose 48-bit timestamp is
    /// `unix_millis` and whose remaining bits are drawn deterministically.
    #[must_use]
    pub fn uuid_v7(&mut self, unix_millis: u64) -> Uuid {
        let mut rand_bytes = [0u8; 10];
        self.inner.fill_bytes(&mut rand_bytes);
        uuid_v7_from_parts(unix_millis, rand_bytes)
    }

    /// Borrow the underlying `ChaCha8Rng`.
    ///
    /// Internal escape hatch for the determinism smoke test.
    #[cfg(test)]
    pub(crate) fn inner_mut(&mut self) -> &mut ChaCha8Rng {
        &mut self.inner
    }
}

/// A virtual clock handle for the simulation. Crate-private: no public API
/// returns it (issue #2967).
///
/// Wraps a [`TickingClock`] started at the fixed sim epoch. [`Sim::advance`]
/// steps this clock (the wall-clock time a [`crate::time::Clock`] extractor
/// reports) in lockstep with tokio's paused virtual timer, so
/// `Utc::now()`-via-extractor and `tokio::time::sleep` never drift apart.
pub(crate) struct SimClock {
    inner: TickingClock,
}

impl SimClock {
    /// Wrap a ticking clock as the simulation's virtual clock.
    pub(crate) fn new(inner: TickingClock) -> Self {
        Self { inner }
    }

    /// Step the injected wall clock forward by `duration`.
    ///
    /// This moves only the framework clock (the [`crate::time::Clock`]
    /// extractor / `ClockSource`); [`Sim::advance`] pairs it with
    /// `tokio::time::advance` so the tokio timer wheel moves the same amount.
    pub(crate) fn advance(&self, duration: std::time::Duration) {
        self.inner.advance(duration);
    }

    /// A [`TickingClock`] handle sharing this clock's instant.
    ///
    /// Handed to [`crate::test::TestApp::with_clock`] at [`Sim::build`] time so
    /// mounted handlers read the same virtual instant [`Sim::advance`] steps.
    pub(crate) fn ticking(&self) -> TickingClock {
        self.inner.clone()
    }

    /// The clock's current virtual UTC instant.
    ///
    /// Read by [`Sim::advance_to`] to compute the forward delta to a zoned
    /// target.
    pub(crate) fn now(&self) -> DateTime<Utc> {
        crate::time::ClockSource::now(&self.inner)
    }
}

/// The clock a `Sim` installs as its thread's ambient clock (issue #2967).
///
/// Wall time is the sim clock, so it moves only on [`Sim::advance`].
///
/// Elapsed time is kept per sim. Tokio's paused clock is one per runtime, and
/// several sims can share a runtime, so a sim's elapsed time is:
///
/// - the time its own [`Sim::advance`] calls moved tokio's clock, plus
/// - the time tokio's clock moved by itself (a `sleep` the paused runtime
///   auto-advanced) while this sim was the ambient clock.
///
/// So an ambient deadline and a `tokio::time::sleep` stay on one timeline, and
/// another sim's `advance` never moves this sim's elapsed time.
struct AmbientSimClock {
    /// The sim clock, or its chaos-skewed view once a skewed app mounts.
    wall: std::sync::RwLock<Arc<dyn crate::time::ClockSource>>,
    /// Time this sim's own `Sim::advance` calls moved tokio's clock.
    own_advanced: std::sync::Mutex<std::time::Duration>,
    /// Time tokio's clock moved by itself while this sim was ambient.
    auto_advanced: std::sync::Mutex<std::time::Duration>,
    /// Cleared when the sim drops, on any thread.
    alive: std::sync::atomic::AtomicBool,
}

impl AmbientSimClock {
    fn add(slot: &std::sync::Mutex<std::time::Duration>, duration: std::time::Duration) {
        let mut total = slot
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        *total = total.saturating_add(duration);
    }

    fn read(slot: &std::sync::Mutex<std::time::Duration>) -> std::time::Duration {
        *slot
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    fn is_alive(&self) -> bool {
        self.alive.load(std::sync::atomic::Ordering::Acquire)
    }
}

/// The sims alive on one thread, and tokio's instant at the last settle.
#[derive(Default)]
struct SimStack {
    /// Newest last. The newest live sim is the ambient one.
    clocks: Vec<Arc<AmbientSimClock>>,
    /// Tokio's instant up to which auto-advanced time is attributed. `None`
    /// until a sim's runtime starts on this thread.
    checkpoint: Option<tokio::time::Instant>,
    /// Whether this thread already warned about a sim that was not anchored.
    warned_unanchored: bool,
    /// `Sim::advance` time not yet taken out of a settle. The next settles
    /// take it out of the tokio time they see first, so an advance is never
    /// also counted as auto-advanced time: not when it is cancelled after the
    /// clock moved, and not when advances run at the same time.
    pending_advance: std::time::Duration,
    /// The runtime these sims run on, so a sim dropped on another thread can
    /// still read this runtime's clock and settle up to its drop.
    runtime: Option<tokio::runtime::Handle>,
}

impl SimStack {
    /// [`settle`](Self::settle), from a clock read or an advance. When no
    /// sim on this thread was anchored, warn once: tokio time that passed
    /// before this call is not in the sim's elapsed time.
    fn settle_lazily(&mut self) {
        if self.checkpoint.is_none()
            && !self.warned_unanchored
            && self.clocks.iter().any(|clock| clock.is_alive())
            && tokio::runtime::Handle::try_current().is_ok()
        {
            self.warned_unanchored = true;
            tracing::warn!(
                "a Sim built outside its runtime was not anchored; its elapsed \
                 time starts now. Call `Sim::anchor` first in the runtime, or use \
                 #[sim_test]"
            );
        }
        self.settle();
    }

    /// Give the tokio time since the last settle to the ambient sim, and
    /// drop entries whose sim has dropped (on any thread).
    fn settle(&mut self) {
        if let Ok(current) = tokio::runtime::Handle::try_current() {
            self.runtime = Some(current);
        }
        let Some(runtime) = self.runtime.clone() else {
            self.clocks.retain(|clock| clock.is_alive());
            return;
        };
        // Read this stack's runtime clock, also from another thread.
        let now = {
            let _enter = runtime.enter();
            tokio::time::Instant::now()
        };
        if let Some(checkpoint) = self.checkpoint {
            let moved = now.saturating_duration_since(checkpoint);
            let explicit = moved.min(self.pending_advance);
            self.pending_advance = self.pending_advance.saturating_sub(explicit);
            if let Some(top) = self.clocks.last() {
                AmbientSimClock::add(&top.auto_advanced, moved.saturating_sub(explicit));
            }
        }
        // Prune only after attributing: the time up to now belongs to the sim
        // that was ambient, even if it has since dropped on another thread.
        self.clocks.retain(|clock| clock.is_alive());
        if self.clocks.is_empty() {
            self.checkpoint = None;
            self.pending_advance = std::time::Duration::ZERO;
            self.runtime = None;
        } else {
            self.checkpoint = Some(now);
        }
    }
}

/// A thread's sim stack. Shared, so a sim dropped on another thread can
/// settle the stack of the thread it was built on.
type SharedSimStack = Arc<std::sync::Mutex<SimStack>>;

thread_local! {
    static SIM_STACK: SharedSimStack = SharedSimStack::default();
}

/// Run `f` on `stack`.
fn lock_sim_stack<T>(stack: &SharedSimStack, f: impl FnOnce(&mut SimStack) -> T) -> T {
    f(&mut stack
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner))
}

/// Run `f` on this thread's sim stack. `None` during thread teardown.
fn with_sim_stack<T>(f: impl FnOnce(&mut SimStack) -> T) -> Option<T> {
    SIM_STACK.try_with(|stack| lock_sim_stack(stack, f)).ok()
}

/// Start attributing tokio time on this thread from now, if nothing has
/// yet. `#[sim_test]` builds the sim before its runtime, so it calls this as
/// the runtime starts. Settling first keeps any time already attributed.
fn anchor_current_sim() {
    if tokio::runtime::Handle::try_current().is_err() {
        return;
    }
    with_sim_stack(SimStack::settle);
}

/// Keeps a sim on its thread's sim stack while the sim lives.
struct SimStackGuard {
    own: Arc<AmbientSimClock>,
    /// The stack of the thread the sim was built on.
    home: SharedSimStack,
}

impl SimStackGuard {
    fn enter(own: Arc<AmbientSimClock>) -> Self {
        let home = SIM_STACK.with(Arc::clone);
        lock_sim_stack(&home, |stack| {
            // Time so far belongs to the sim that was ambient until now.
            stack.settle();
            stack.clocks.push(Arc::clone(&own));
            if stack.checkpoint.is_none() && tokio::runtime::Handle::try_current().is_ok() {
                stack.checkpoint = Some(tokio::time::Instant::now());
            }
        });
        Self { own, home }
    }
}

impl Drop for SimStackGuard {
    fn drop(&mut self) {
        // On the home stack, from any thread: settle, so time up to the drop
        // goes to this sim while it is still ambient; then mark it dead and
        // settle again, which removes it. Later time goes to the next sim.
        lock_sim_stack(&self.home, |stack| {
            stack.settle();
            self.own
                .alive
                .store(false, std::sync::atomic::Ordering::Release);
            stack.settle();
        });
    }
}

/// Record `duration` as `sim`'s own advance. Call in the same poll that moves
/// tokio's clock, so a cancel cannot split the two.
fn note_own_advance(sim: &AmbientSimClock, duration: std::time::Duration) {
    with_sim_stack(|stack| {
        stack.settle_lazily();
        stack.pending_advance = stack.pending_advance.saturating_add(duration);
    });
    AmbientSimClock::add(&sim.own_advanced, duration);
}

impl crate::time::ClockSource for AmbientSimClock {
    fn now(&self) -> DateTime<Utc> {
        self.wall
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .now()
    }

    fn monotonic(&self) -> crate::time::MonotonicInstant {
        // With no runtime there is no paused clock to settle from. The total
        // as last settled still holds the auto-advanced time, so a thread
        // with no runtime reads the same time as the sim's.
        if tokio::runtime::Handle::try_current().is_ok() {
            with_sim_stack(SimStack::settle_lazily);
        }
        crate::time::MonotonicInstant::from_origin_elapsed(
            Self::read(&self.own_advanced).saturating_add(Self::read(&self.auto_advanced)),
        )
    }
}

/// The built application handle for a simulation.
///
/// Holds the [`crate::test::TestClient`] mounted by [`Sim::build`] on the paused
/// runtime with the simulation's virtual clock installed. Empty until
/// [`Sim::build`] is called (an empty [`#[sim_test]`](crate::sim_test) that only
/// drives time / RNG never mounts an app).
///
/// Crate-private: no public API returns it (issue #2967).
#[derive(Default)]
pub(crate) struct SimApp {
    /// The mounted test client, or `None` before [`Sim::build`].
    client: Option<crate::test::TestClient>,
}

impl SimApp {
    /// Borrow the mounted [`crate::test::TestClient`].
    ///
    /// # Panics
    ///
    /// Panics if no app has been mounted yet — call [`Sim::build`] first.
    #[must_use]
    pub fn client(&self) -> &crate::test::TestClient {
        self.try_client()
            .expect("no app mounted: call `sim.build(TestApp::new()...)` before `client()`")
    }

    /// Borrow the mounted [`crate::test::TestClient`], or `None` before
    /// [`Sim::build`].
    #[must_use]
    pub const fn try_client(&self) -> Option<&crate::test::TestClient> {
        self.client.as_ref()
    }
}

/// Read and parse the simulation seed from the `AUTUMN_SIM_SEED` environment
/// variable.
///
/// Accepts a hex (`0x`-prefixed) or decimal `u64`; an absent or unparseable
/// value falls back to `0`. Called by the [`#[sim_test]`](crate::sim_test)
/// macro so the parsing is unit-tested in one place and the macro stays tiny.
#[doc(hidden)]
#[must_use]
pub fn __seed_from_env() -> u64 {
    std::env::var("AUTUMN_SIM_SEED").map_or(0, |raw| parse_seed(&raw))
}

/// Parse a seed string: hex (`0x`/`0X` prefixed) or decimal, defaulting to `0`
/// on any parse failure or empty input.
fn parse_seed(raw: &str) -> u64 {
    let trimmed = raw.trim();
    trimmed
        .strip_prefix("0x")
        .or_else(|| trimmed.strip_prefix("0X"))
        .map_or_else(
            || trimmed.parse::<u64>().unwrap_or(0),
            |hex| u64::from_str_radix(hex, 16).unwrap_or(0),
        )
}

/// Build the deterministic replay line printed on a sim-test panic.
///
/// Returns exactly
/// `AUTUMN_SIM_SEED=0x<seed-hex> cargo test -p <pkg> <test>` — copy-paste it to
/// reproduce the failing run bit-for-bit. Called by the
/// [`#[sim_test]`](crate::sim_test) macro.
#[doc(hidden)]
#[must_use]
pub fn __replay_line(seed: u64, pkg: &str, test: &str) -> String {
    format!("AUTUMN_SIM_SEED=0x{seed:x} cargo test -p {pkg} {test}")
}

#[cfg(test)]
mod tests {
    use super::{
        __replay_line, AdvancePlan, DEFAULT_STRICT_WALL_CLOCK_BUDGET, MAX_LIVENESS_BUDGET, Sim,
        mount_entropy_seed, parse_liveness_budget, parse_seed, parse_strict_budget_ms,
        plan_advance_to, resolve_local_to_utc, strict_budget_from_env_or,
    };
    use chrono::{NaiveDate, TimeZone, Utc};
    use rand::Rng;

    #[tokio::test(start_paused = true)]
    async fn blocking_work_reads_auto_advanced_sim_time() {
        let sim = Sim::from_seed(11);
        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
        sim.advance(std::time::Duration::from_secs(2)).await;
        let here = (
            crate::time::ambient_monotonic(),
            crate::time::ambient_instant(),
            crate::time::ambient_now(),
        );
        assert_eq!(here.0.since_origin(), std::time::Duration::from_secs(7));
        let there = crate::time::spawn_blocking(|| {
            (
                crate::time::ambient_monotonic(),
                crate::time::ambient_instant(),
                crate::time::ambient_now(),
            )
        })
        .await
        .unwrap();
        assert_eq!(there, here, "a blocking worker reads the caller's sim time");
    }

    #[tokio::test(start_paused = true)]
    async fn a_thread_with_no_runtime_reads_auto_advanced_sim_time() {
        let sim = Sim::from_seed(12);
        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
        let here = crate::time::ambient_monotonic();
        assert_eq!(here.since_origin(), std::time::Duration::from_secs(5));
        let clock: std::sync::Arc<dyn crate::time::ClockSource> = sim.ambient.clone();
        let there = std::thread::spawn(move || clock.monotonic())
            .join()
            .unwrap();
        assert_eq!(there, here);
    }

    #[test]
    fn liveness_budget_is_off_unless_armed() {
        assert_eq!(parse_liveness_budget(None), None);
        assert_eq!(parse_liveness_budget(Some("")), None);
        assert_eq!(parse_liveness_budget(Some("0")), None);
        assert_eq!(parse_liveness_budget(Some("soon")), None);
        assert_eq!(parse_liveness_budget(Some("-5")), None);
    }

    #[test]
    fn liveness_budget_takes_seconds_and_clamps_to_the_tokio_maximum() {
        assert_eq!(
            parse_liveness_budget(Some(" 90 ")),
            Some(std::time::Duration::from_secs(90))
        );
        assert_eq!(
            parse_liveness_budget(Some("18446744073709551615")),
            Some(MAX_LIVENESS_BUDGET)
        );
    }

    #[test]
    fn mount_entropy_seed_is_the_sim_seed_first_then_derived() {
        assert_eq!(mount_entropy_seed(7, 0), 7);
        let restart = mount_entropy_seed(7, 1);
        assert_ne!(restart, 7, "a restart draws a new stream");
        assert_eq!(restart, mount_entropy_seed(7, 1), "and it is deterministic");
        assert_ne!(restart, mount_entropy_seed(7, 2));
        assert_ne!(restart, mount_entropy_seed(8, 1));
    }

    #[test]
    fn replay_line_zero_seed_is_exact() {
        assert_eq!(
            __replay_line(0, "autumn-web", "my_test"),
            "AUTUMN_SIM_SEED=0x0 cargo test -p autumn-web my_test"
        );
    }

    #[test]
    fn replay_line_formats_seed_as_hex() {
        let line = __replay_line(0x9f3a, "autumn-web", "my_test");
        assert!(
            line.contains("0x9f3a"),
            "seed must be rendered in hex: {line}"
        );
        assert_eq!(
            line,
            "AUTUMN_SIM_SEED=0x9f3a cargo test -p autumn-web my_test"
        );
    }

    #[test]
    fn parse_seed_covers_hex_decimal_and_garbage() {
        assert_eq!(parse_seed("0"), 0);
        assert_eq!(parse_seed("0x9f3a"), 0x9f3a);
        assert_eq!(parse_seed("0X9F3A"), 0x9f3a);
        assert_eq!(parse_seed("42"), 42);
        assert_eq!(parse_seed("garbage"), 0);
        assert_eq!(parse_seed(""), 0);
        assert_eq!(parse_seed("  0x10  "), 0x10);
    }

    #[test]
    fn from_seed_exposes_the_seed() {
        assert_eq!(Sim::from_seed(0).seed, 0);
        assert_eq!(Sim::from_seed(7).seed(), 7);
    }

    #[test]
    fn same_seed_produces_identical_first_draw() {
        let mut a = Sim::from_seed(7);
        let mut b = Sim::from_seed(7);
        let da = a.rng().inner_mut().next_u64();
        let db = b.rng().inner_mut().next_u64();
        assert_eq!(da, db, "same seed must yield the same first RNG draw");
    }

    #[test]
    fn plan_advance_to_equal_target_is_noop() {
        let now = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
        assert_eq!(plan_advance_to(now, now), AdvancePlan::NoOp);
    }

    #[test]
    fn plan_advance_to_future_target_is_exact_delta() {
        let now = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
        let target = Utc.with_ymd_and_hms(2020, 1, 1, 1, 0, 0).unwrap();
        assert_eq!(
            plan_advance_to(now, target),
            AdvancePlan::Advance(std::time::Duration::from_secs(3600))
        );
    }

    #[test]
    #[should_panic(expected = "forward-only")]
    fn plan_advance_to_past_target_panics() {
        let now = Utc.with_ymd_and_hms(2020, 1, 1, 1, 0, 0).unwrap();
        let target = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
        let _ = plan_advance_to(now, target);
    }

    #[test]
    fn parse_strict_budget_ms_covers_valid_and_garbage() {
        use std::time::Duration;
        assert_eq!(
            parse_strict_budget_ms("100"),
            Some(Duration::from_millis(100))
        );
        assert_eq!(
            parse_strict_budget_ms("  250  "),
            Some(Duration::from_millis(250))
        );
        assert_eq!(parse_strict_budget_ms("0"), Some(Duration::ZERO));
        assert_eq!(parse_strict_budget_ms(""), None);
        assert_eq!(parse_strict_budget_ms("   "), None);
        assert_eq!(parse_strict_budget_ms("garbage"), None);
        assert_eq!(parse_strict_budget_ms("-5"), None);
        assert_eq!(parse_strict_budget_ms("1.5"), None);
    }

    #[test]
    fn strict_budget_from_env_falls_back_to_default_when_unset() {
        // The env var is not set in this unit-test process, so the default is
        // returned verbatim (the override path is exercised via the public
        // integration tests, which never set the var).
        let default = std::time::Duration::from_millis(7);
        assert_eq!(strict_budget_from_env_or(default), default);
        assert_eq!(
            strict_budget_from_env_or(DEFAULT_STRICT_WALL_CLOCK_BUDGET),
            DEFAULT_STRICT_WALL_CLOCK_BUDGET
        );
    }

    #[test]
    fn strict_wall_clock_builders_set_the_budget() {
        let mut sim = Sim::from_seed(0);
        assert!(sim.strict_budget.is_none(), "guard is off by default");
        sim.strict_wall_clock();
        assert_eq!(sim.strict_budget, Some(DEFAULT_STRICT_WALL_CLOCK_BUDGET));

        let mut custom = Sim::from_seed(0);
        custom.strict_wall_clock_budget(std::time::Duration::from_millis(5));
        assert_eq!(
            custom.strict_budget,
            Some(std::time::Duration::from_millis(5))
        );
    }

    #[test]
    fn resolve_local_unambiguous_maps_to_single_instant() {
        // A plain UTC-offset zone: 12:00 at +00:00 is exactly 12:00Z.
        let local = NaiveDate::from_ymd_opt(2020, 6, 1)
            .unwrap()
            .and_hms_opt(12, 0, 0)
            .unwrap();
        let got = resolve_local_to_utc(local, &Utc);
        assert_eq!(got, Utc.with_ymd_and_hms(2020, 6, 1, 12, 0, 0).unwrap());
    }

    /// A replica on the sim network is a host of that network, and its app
    /// holds the network. The sim must break that cycle when it drops, or a
    /// seed sweep keeps every replica app of every seed in memory.
    #[cfg(feature = "http-client")]
    #[tokio::test(start_paused = true)]
    async fn a_dropped_sim_frees_its_replica_apps() {
        let marker = std::sync::Arc::new(());
        let freed = std::sync::Arc::downgrade(&marker);
        {
            let mut sim = Sim::from_seed(21);
            sim.net(super::net::SimNet::new());
            let held = std::sync::Arc::clone(&marker);
            let router = axum::Router::new().route(
                "/held",
                axum::routing::get(move || {
                    let holders = std::sync::Arc::strong_count(&held);
                    async move { holders.to_string() }
                }),
            );
            sim.mount_replica("a", crate::test::TestApp::new().merge(router));
            sim.mount_replica("b", crate::test::TestApp::new());
        }
        drop(marker);
        assert!(
            freed.upgrade().is_none(),
            "the replica app is still alive after its sim dropped"
        );
    }

    /// A replica stays a host of the network it joined, even after
    /// [`Sim::net`] sets a new network. The kill and the drop must clean that
    /// network, and leave a host of the same name on the new one.
    #[cfg(feature = "http-client")]
    #[tokio::test(start_paused = true)]
    async fn a_replica_is_cleaned_from_the_network_it_joined() {
        let marker = std::sync::Arc::new(());
        let freed = std::sync::Arc::downgrade(&marker);
        let first = super::net::SimNet::new();
        let held = std::sync::Arc::clone(&marker);
        let router = axum::Router::new().route(
            "/held",
            axum::routing::get(move || {
                let holders = std::sync::Arc::strong_count(&held);
                async move { holders.to_string() }
            }),
        );
        let second = super::net::SimNet::new().host("a", axum::Router::new());
        {
            let mut sim = Sim::from_seed(22);
            sim.net(first.clone());
            sim.mount_replica("a", crate::test::TestApp::new().merge(router));
            sim.net(second.clone());
        }
        drop(marker);
        assert!(
            freed.upgrade().is_none(),
            "the replica app is still alive in the network it joined"
        );
        assert!(first.service("a").is_none());
        assert!(
            second.service("a").is_some(),
            "a host of another network stays"
        );
    }
}